nautilus-execution 0.67.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

use std::{cell::RefCell, collections::VecDeque, fmt::Debug, rc::Rc};

use ahash::{AHashMap, AHashSet};
use nautilus_common::{
    cache::Cache,
    clock::Clock,
    logging::{CMD, EVT, RECV, SEND},
    messages::{
        data::{
            DataCommand, SubscribeCommand, SubscribeQuotes, SubscribeTrades, UnsubscribeCommand,
            UnsubscribeQuotes, UnsubscribeTrades,
        },
        execution::{
            BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder, SubmitOrder,
            SubmitOrderList, TradingCommand,
        },
    },
    msgbus::{
        self, MessagingSwitchboard, TypedHandler, TypedIntoHandler,
        switchboard::{
            get_event_order_topic, get_order_canceled_topic, get_quotes_topic, get_trades_topic,
        },
    },
};
use nautilus_core::{UUID4, WeakCell};
use nautilus_model::{
    data::{OrderBookDeltas, QuoteTick, TradeTick},
    enums::{ContingencyType, OrderSide, OrderStatus, OrderType, TimeInForce, TriggerType},
    events::{
        OrderCanceled, OrderEmulated, OrderEventAny, OrderInitialized, OrderReleased, OrderUpdated,
    },
    identifiers::{ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId},
    instruments::Instrument,
    orders::{LimitOrder, MarketOrder, Order, OrderAny},
    types::{Price, Quantity},
};
use ustr::Ustr;

use super::{PendingMessage, handlers::OrderEmulatorOnEventHandler};
use crate::{
    matching::{MatchAction, OrderMatchingCore, RestingOrder},
    order_manager::{OrderManagerAction, manager::OrderManager},
    trailing::{store_trailing_stop_activation, trailing_stop_calculate},
};

pub struct OrderEmulator {
    clock: Rc<RefCell<dyn Clock>>,
    cache: Rc<RefCell<Cache>>,
    manager: OrderManager,
    matching_cores: AHashMap<InstrumentId, OrderMatchingCore>,
    subscribed_quotes: AHashSet<InstrumentId>,
    subscribed_trades: AHashSet<InstrumentId>,
    subscribed_strategies: AHashSet<StrategyId>,
    monitored_positions: AHashSet<PositionId>,
    quote_tick_handler: Option<TypedHandler<QuoteTick>>,
    trade_tick_handler: Option<TypedHandler<TradeTick>>,
    quote_handlers: AHashMap<InstrumentId, TypedHandler<QuoteTick>>,
    trade_handlers: AHashMap<InstrumentId, TypedHandler<TradeTick>>,
    on_event_handler: Option<TypedHandler<OrderEventAny>>,
    pending_messages: Rc<RefCell<VecDeque<PendingMessage>>>,
}

impl Debug for OrderEmulator {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(OrderEmulator))
            .field("cores", &self.matching_cores.len())
            .field("subscribed_quotes", &self.subscribed_quotes.len())
            .field("subscribed_trades", &self.subscribed_trades.len())
            .field("subscribed_strategies", &self.subscribed_strategies.len())
            .finish()
    }
}

impl OrderEmulator {
    pub fn new(clock: Rc<RefCell<dyn Clock>>, cache: Rc<RefCell<Cache>>) -> Self {
        let active_local = true;
        let manager = OrderManager::new(clock.clone(), cache.clone(), active_local);

        Self {
            clock,
            cache,
            manager,
            matching_cores: AHashMap::new(),
            subscribed_quotes: AHashSet::new(),
            subscribed_trades: AHashSet::new(),
            subscribed_strategies: AHashSet::new(),
            monitored_positions: AHashSet::new(),
            quote_tick_handler: None,
            trade_tick_handler: None,
            quote_handlers: AHashMap::new(),
            trade_handlers: AHashMap::new(),
            on_event_handler: None,
            pending_messages: Rc::new(RefCell::new(VecDeque::new())),
        }
    }

    pub fn register_msgbus_handlers(emulator: &Rc<RefCell<Self>>) {
        let weak = WeakCell::from(Rc::downgrade(emulator));
        let pending_messages = emulator.borrow().pending_messages.clone();

        let execute_weak = weak.clone();
        let execute_pending = pending_messages.clone();
        let execute_handler = TypedIntoHandler::from(move |cmd: TradingCommand| {
            if let Some(emulator) = execute_weak.upgrade() {
                match emulator.try_borrow_mut() {
                    Ok(mut emulator) => emulator.execute(cmd),
                    Err(_) => execute_pending
                        .borrow_mut()
                        .push_back(PendingMessage::Command(Box::new(cmd))),
                }
            }
        });
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::order_emulator_execute(),
            execute_handler,
        );

        let quote_weak = weak.clone();
        let quote_handler = TypedHandler::from(move |quote: &QuoteTick| {
            if let Some(emulator) = quote_weak.upgrade() {
                emulator.borrow_mut().on_quote_tick(*quote);
            }
        });

        let trade_weak = weak.clone();
        let trade_handler = TypedHandler::from(move |trade: &TradeTick| {
            if let Some(emulator) = trade_weak.upgrade() {
                emulator.borrow_mut().on_trade_tick(*trade);
            }
        });

        let on_event_handler = TypedHandler::new(OrderEmulatorOnEventHandler::new(
            Ustr::from(UUID4::new().as_str()),
            weak,
            WeakCell::from(Rc::downgrade(&pending_messages)),
        ));

        let mut emulator = emulator.borrow_mut();
        emulator.quote_tick_handler = Some(quote_handler);
        emulator.trade_tick_handler = Some(trade_handler);
        emulator.on_event_handler = Some(on_event_handler);
    }

    pub fn set_on_event_handler(&mut self, handler: TypedHandler<OrderEventAny>) {
        self.on_event_handler = Some(handler);
    }

    /// Caches a submit order command for emulation tracking.
    pub fn cache_submit_order_command(&mut self, command: SubmitOrder) {
        self.manager.cache_submit_order_command(command);
    }

    /// Subscribes to quote data for the given instrument.
    fn subscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
        if self.quote_handlers.contains_key(&instrument_id) {
            log::warn!("OrderEmulator attempted duplicate quote subscription for {instrument_id}");
            return true;
        }

        let Some(handler) = self.quote_tick_handler.clone() else {
            log::warn!("Cannot subscribe to quotes: msgbus handlers not registered");
            return false;
        };

        let topic = get_quotes_topic(instrument_id);
        self.quote_handlers.insert(instrument_id, handler.clone());
        msgbus::subscribe_quotes(topic.into(), handler, None);
        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Quotes(
            SubscribeQuotes::new(
                instrument_id,
                None,
                Some(instrument_id.venue),
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                None,
                None,
            ),
        )));

        true
    }

    /// Subscribes to trade data for the given instrument.
    fn subscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) -> bool {
        if self.trade_handlers.contains_key(&instrument_id) {
            log::warn!("OrderEmulator attempted duplicate trade subscription for {instrument_id}");
            return true;
        }

        let Some(handler) = self.trade_tick_handler.clone() else {
            log::warn!("Cannot subscribe to trades: msgbus handlers not registered");
            return false;
        };

        let topic = get_trades_topic(instrument_id);
        self.trade_handlers.insert(instrument_id, handler.clone());
        msgbus::subscribe_trades(topic.into(), handler, None);
        self.send_data_command(DataCommand::Subscribe(SubscribeCommand::Trades(
            SubscribeTrades::new(
                instrument_id,
                None,
                Some(instrument_id.venue),
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                None,
                None,
            ),
        )));

        true
    }

    #[must_use]
    pub fn subscribed_quotes(&self) -> Vec<InstrumentId> {
        let mut quotes: Vec<InstrumentId> = self.subscribed_quotes.iter().copied().collect();
        quotes.sort();
        quotes
    }

    #[must_use]
    pub fn subscribed_trades(&self) -> Vec<InstrumentId> {
        let mut trades: Vec<_> = self.subscribed_trades.iter().copied().collect();
        trades.sort();
        trades
    }

    #[must_use]
    pub fn subscribed_strategy_count(&self) -> usize {
        self.subscribed_strategies.len()
    }

    #[must_use]
    pub fn monitored_position_count(&self) -> usize {
        self.monitored_positions.len()
    }

    #[must_use]
    pub fn get_submit_order_commands(&self) -> AHashMap<ClientOrderId, SubmitOrder> {
        self.manager.get_submit_order_commands()
    }

    #[must_use]
    pub fn get_matching_core(&self, instrument_id: &InstrumentId) -> Option<OrderMatchingCore> {
        self.matching_cores.get(instrument_id).cloned()
    }

    pub fn start(&mut self) {
        if let Err(e) = self.on_start() {
            log::error!("{e}");
        }

        log::info!("Started");
    }

    pub fn stop(&self) {
        self.on_stop();

        log::info!("Stopped");
    }

    pub fn reset(&mut self) {
        self.on_reset();

        log::info!("Reset");
    }

    pub fn dispose(&mut self) {
        self.on_dispose();

        log::info!("Disposed");
    }

    /// Reactivates emulated orders from cache on start.
    ///
    /// # Errors
    ///
    /// Returns an error if processing fails.
    pub fn on_start(&mut self) -> anyhow::Result<()> {
        let emulated_orders: Vec<ClientOrderId> = self
            .cache
            .borrow()
            .orders_emulated(None, None, None, None, None)
            .into_iter()
            .map(|order| order.client_order_id())
            .collect();

        if emulated_orders.is_empty() {
            log::debug!("No emulated orders to reactivate");
            return Ok(());
        }

        for client_order_id in emulated_orders {
            self.reactivate_order(client_order_id);
        }

        self.drain_pending_messages();

        Ok(())
    }

    fn reactivate_order(&mut self, client_order_id: ClientOrderId) {
        let Some(order) = self.cache.borrow().order_owned(&client_order_id) else {
            return;
        };

        if !matches!(
            order.status(),
            OrderStatus::Initialized | OrderStatus::Emulated
        ) {
            return; // No longer emulated
        }

        if let Some(parent_order_id) = &order.parent_order_id() {
            let parent_order = if let Some(order) = self.cache.borrow().order(parent_order_id) {
                order.clone()
            } else {
                log::error!("Cannot handle order: parent {parent_order_id} not found");
                return;
            };

            let is_position_closed = parent_order
                .position_id()
                .is_none_or(|id| self.cache.borrow().is_position_closed(&id));

            if parent_order.is_closed() && is_position_closed {
                let actions = self.manager.cancel_order(&order);
                self.dispatch_manager_actions(actions);
                return; // Parent already closed
            }

            if parent_order.contingency_type() == Some(ContingencyType::Oto)
                && (parent_order.is_active_local()
                    || parent_order.filled_qty() == Quantity::zero(0))
            {
                return; // Process contingency order later once parent triggered
            }
        }

        let position_id = self
            .cache
            .borrow()
            .position_id(&order.client_order_id())
            .copied();
        let client_id = self
            .cache
            .borrow()
            .client_id(&order.client_order_id())
            .copied();

        let command = SubmitOrder::new(
            order.trader_id(),
            client_id,
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            order.init_event().clone(),
            order.exec_algorithm_id(),
            position_id,
            None, // params
            UUID4::new(),
            self.clock.borrow().timestamp_ns(),
            None, // correlation_id
        );

        self.manager.cache_submit_order_command(command.clone());
        self.handle_submit_order(&command);
    }

    pub fn on_event(&mut self, event: &OrderEventAny) {
        log::info!("{RECV}{EVT} {event}");

        let actions = self.manager.handle_event(event);
        self.dispatch_manager_actions(actions);

        if let Some(order) = self.cache.borrow().order(&event.client_order_id())
            && order.is_closed()
            && let Some(matching_core) = self.matching_cores.get_mut(&order.instrument_id())
            && let Err(e) = matching_core.delete_order(event.client_order_id())
        {
            log::debug!("Error deleting order: {e}");
        }
        // else: Order not in cache yet

        self.drain_pending_messages();
    }

    fn drain_pending_messages(&mut self) {
        loop {
            let message = self.pending_messages.borrow_mut().pop_front();
            match message {
                Some(PendingMessage::Command(command)) => self.execute(*command),
                Some(PendingMessage::Event(event)) => self.on_event(&event),
                None => break,
            }
        }
    }

    pub const fn on_stop(&self) {}

    pub fn on_reset(&mut self) {
        self.manager.reset();
        self.pending_messages.borrow_mut().clear();
        self.matching_cores.clear();
        self.unsubscribe_all_market_data();
        self.unsubscribe_strategy_order_events();
        self.monitored_positions.clear();
    }

    pub fn on_dispose(&mut self) {
        self.on_reset();
    }

    fn unsubscribe_all_market_data(&mut self) {
        // Sort so the emitted unsubscribe commands, and the UUID4 draws they consume,
        // follow the same sequence across runs; the drained sets are AHash-backed.
        let mut quote_instrument_ids: Vec<_> = self.subscribed_quotes.drain().collect();
        quote_instrument_ids.sort();

        for instrument_id in quote_instrument_ids {
            self.unsubscribe_quotes_for_instrument(instrument_id);
        }

        let mut trade_instrument_ids: Vec<_> = self.subscribed_trades.drain().collect();
        trade_instrument_ids.sort();

        for instrument_id in trade_instrument_ids {
            self.unsubscribe_trades_for_instrument(instrument_id);
        }
    }

    fn unsubscribe_quotes_for_instrument(&mut self, instrument_id: InstrumentId) {
        if let Some(handler) = self.quote_handlers.remove(&instrument_id) {
            let topic = get_quotes_topic(instrument_id);
            msgbus::unsubscribe_quotes(topic.into(), &handler);
        }

        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(
            UnsubscribeQuotes::new(
                instrument_id,
                None,
                Some(instrument_id.venue),
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                None,
                None,
            ),
        )));
    }

    fn unsubscribe_trades_for_instrument(&mut self, instrument_id: InstrumentId) {
        if let Some(handler) = self.trade_handlers.remove(&instrument_id) {
            let topic = get_trades_topic(instrument_id);
            msgbus::unsubscribe_trades(topic.into(), &handler);
        }

        self.send_data_command(DataCommand::Unsubscribe(UnsubscribeCommand::Trades(
            UnsubscribeTrades::new(
                instrument_id,
                None,
                Some(instrument_id.venue),
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                None,
                None,
            ),
        )));
    }

    fn unsubscribe_strategy_order_events(&mut self) {
        let mut strategy_ids: Vec<_> = self.subscribed_strategies.drain().collect();
        strategy_ids.sort();
        let Some(handler) = &self.on_event_handler else {
            return;
        };

        for strategy_id in strategy_ids {
            msgbus::unsubscribe_order_events(format!("events.order.{strategy_id}").into(), handler);
        }
    }

    pub fn execute(&mut self, command: TradingCommand) {
        log::info!("{RECV}{CMD} {command}");

        match command {
            TradingCommand::SubmitOrder(command) => self.handle_submit_order(&command),
            TradingCommand::SubmitOrderList(ref command) => self.handle_submit_order_list(command),
            TradingCommand::ModifyOrder(ref command) => self.handle_modify_order(command),
            TradingCommand::ModifyOrders(ref command) => self.handle_batch_modify_orders(command),
            TradingCommand::CancelOrder(command) => self.handle_cancel_order(command),
            TradingCommand::CancelAllOrders(ref command) => self.handle_cancel_all_orders(command),
            _ => log::error!("Cannot handle command: unrecognized {command}"),
        }

        self.drain_pending_messages();
    }

    fn dispatch_manager_actions(&mut self, actions: Vec<OrderManagerAction>) {
        for action in actions {
            self.dispatch_manager_action(action);
        }
    }

    fn dispatch_manager_action(&mut self, action: OrderManagerAction) {
        match action {
            OrderManagerAction::PublishInitialized(event) => publish_order_event(&event),
            OrderManagerAction::SubmitToEmulator(command) => self.handle_submit_order(&command),
            OrderManagerAction::SubmitToRisk(command) => {
                self.send_risk_command(TradingCommand::SubmitOrder(command));
            }
            OrderManagerAction::SubmitToAlgorithm {
                command,
                exec_algorithm_id,
            } => self.send_algo_command(command, exec_algorithm_id),
            OrderManagerAction::CancelLocal(order) => self.cancel_order(&order),
            OrderManagerAction::ModifyLocalQuantity {
                mut order,
                quantity,
            } => {
                self.update_order(&mut order, quantity);
            }
        }
    }

    fn create_matching_core(
        &mut self,
        instrument_id: InstrumentId,
        price_increment: Price,
        emulation_trigger: TriggerType,
    ) -> OrderMatchingCore {
        let mut matching_core = OrderMatchingCore::new(instrument_id, price_increment);
        let cache = self.cache.borrow();
        if emulation_trigger == TriggerType::LastPrice {
            if let Some(trade) = cache.trade(&instrument_id) {
                matching_core.set_bid_raw(trade.price);
                matching_core.set_ask_raw(trade.price);
                matching_core.set_last_raw(trade.price);
            }
        } else if let Some(quote) = cache.quote(&instrument_id) {
            matching_core.set_bid_raw(quote.bid_price);
            matching_core.set_ask_raw(quote.ask_price);
        }

        drop(cache);
        self.matching_cores
            .insert(instrument_id, matching_core.clone());

        log::info!("Creating matching core for {instrument_id:?}");
        matching_core
    }

    /// # Panics
    ///
    /// Panics if the emulation trigger type is not set or if the order is not in the cache.
    pub fn handle_submit_order(&mut self, command: &SubmitOrder) {
        let client_order_id = command.client_order_id;

        let mut order = self
            .cache
            .borrow()
            .order(&client_order_id)
            .map(|o| o.clone())
            .expect("order must exist in cache");

        if order.is_closed() {
            return;
        }

        let emulation_trigger = order.emulation_trigger();

        if !self
            .manager
            .get_submit_order_commands()
            .contains_key(&client_order_id)
        {
            self.manager.cache_submit_order_command(command.clone());
        }

        assert!(
            emulation_trigger.is_some(),
            "order.emulation_trigger must be set"
        );

        if !matches!(
            emulation_trigger,
            Some(TriggerType::Default | TriggerType::BidAsk | TriggerType::LastPrice)
        ) {
            log::error!("Cannot emulate order: `TriggerType` {emulation_trigger:?} not supported");
            let actions = self.manager.cancel_order(&order);
            self.dispatch_manager_actions(actions);
            return;
        }

        let emulation_trigger = emulation_trigger.expect("emulation trigger must be set");
        let strategy_id = command.strategy_id;
        let position_id = command.position_id;

        let trigger_instrument_id = order
            .trigger_instrument_id()
            .unwrap_or_else(|| order.instrument_id());

        let matching_core = self.matching_cores.get(&trigger_instrument_id).cloned();

        let mut matching_core = if let Some(core) = matching_core {
            core
        } else {
            // Handle synthetic instruments
            let (instrument_id, price_increment) = if trigger_instrument_id.is_synthetic() {
                let synthetic = self
                    .cache
                    .borrow()
                    .try_synthetic(&trigger_instrument_id)
                    .cloned();

                match synthetic {
                    Ok(synthetic) => (synthetic.id, synthetic.price_increment),
                    Err(e) => {
                        log::error!("Cannot emulate order: {e}");
                        let actions = self.manager.cancel_order(&order);
                        self.dispatch_manager_actions(actions);
                        return;
                    }
                }
            } else {
                let instrument = self
                    .cache
                    .borrow()
                    .instrument(&trigger_instrument_id)
                    .cloned();

                if let Some(instrument) = instrument {
                    (instrument.id(), instrument.price_increment())
                } else {
                    log::error!(
                        "Cannot emulate order: no instrument {trigger_instrument_id} for trigger"
                    );
                    let actions = self.manager.cancel_order(&order);
                    self.dispatch_manager_actions(actions);
                    return;
                }
            };

            self.create_matching_core(instrument_id, price_increment, emulation_trigger)
        };

        // Update trailing stop
        if matches!(
            order.order_type(),
            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
        ) {
            self.update_trailing_stop_order(&mut order);
            if order.trigger_price().is_none() && is_order_activated(&order) {
                log::error!(
                    "Cannot handle trailing stop order with no trigger_price and no market updates"
                );

                let actions = self.manager.cancel_order(&order);
                self.dispatch_manager_actions(actions);
                return;
            }
            // Not yet activated: held inert until the activation price is touched
        }

        self.check_monitoring(strategy_id, position_id);

        // Check if immediately marketable
        let is_activated = is_order_activated(&order);

        let match_info = RestingOrder::new_with_trigger_type(
            order.client_order_id(),
            order.order_side(),
            order.order_type(),
            order.emulation_trigger(),
            if is_activated {
                order.trigger_price()
            } else {
                None
            },
            if is_activated { order.price() } else { None },
            is_activated,
        );

        if let Some(action) = matching_core.match_order(&match_info) {
            self.dispatch_match_action(action);
        }

        // Handle data subscriptions
        match emulation_trigger {
            TriggerType::Default | TriggerType::BidAsk => {
                if !self.subscribed_quotes.contains(&trigger_instrument_id)
                    && self.subscribe_quotes_for_instrument(trigger_instrument_id)
                {
                    self.subscribed_quotes.insert(trigger_instrument_id);
                }
            }
            TriggerType::LastPrice => {
                if !self.subscribed_trades.contains(&trigger_instrument_id)
                    && self.subscribe_trades_for_instrument(trigger_instrument_id)
                {
                    self.subscribed_trades.insert(trigger_instrument_id);
                }
            }
            _ => {
                log::error!("Invalid TriggerType: {emulation_trigger:?}");
                return;
            }
        }

        // Check if order was already released
        if !self
            .manager
            .get_submit_order_commands()
            .contains_key(&order.client_order_id())
        {
            return; // Already released
        }

        // Hold in matching core
        matching_core.add_order(match_info);

        if order.status() == OrderStatus::Initialized {
            let event = OrderEmulated::new(
                order.trader_id(),
                order.strategy_id(),
                order.instrument_id(),
                order.client_order_id(),
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                self.clock.borrow().timestamp_ns(),
            );

            let event = OrderEventAny::Emulated(event);

            order = match self.cache.borrow_mut().update_order(&event) {
                Ok(order) => order,
                Err(e) => {
                    log::error!("Cannot apply order event: {e:?}");
                    return;
                }
            };

            self.send_risk_event(event.clone());

            let topic = get_event_order_topic(order.strategy_id());
            msgbus::publish_order_event(topic, &event);
        }

        // Since we are cloning the matching core, we need to insert it back into the original hashmap
        self.matching_cores
            .insert(trigger_instrument_id, matching_core);

        log::info!("Emulating {order}");
    }

    fn handle_submit_order_list(&mut self, command: &SubmitOrderList) {
        self.check_monitoring(command.strategy_id, command.position_id);

        let orders: Vec<OrderAny> = self
            .cache
            .borrow()
            .orders_for_ids(&command.order_list.client_order_ids, &command);

        for order in &orders {
            let order = self.cache.borrow().order_owned(&order.client_order_id());

            let Some(order) = order else {
                continue;
            };

            if order.is_closed() {
                continue;
            }

            if let Some(parent_order_id) = order.parent_order_id() {
                let cache = self.cache.borrow();
                let parent_order = if let Some(parent_order) = cache.order(&parent_order_id) {
                    parent_order
                } else {
                    log::error!("Parent order for {} not found", order.client_order_id());
                    continue;
                };

                if parent_order.contingency_type() == Some(ContingencyType::Oto) {
                    continue; // Process contingency order later once parent triggered
                }
            }

            match self.manager.create_new_submit_order(
                &order,
                command.position_id,
                command.client_id,
                command.correlation_id,
            ) {
                Ok(actions) => self.dispatch_manager_actions(actions),
                Err(e) => log::error!("Error creating new submit order: {e}"),
            }
        }
    }

    fn handle_modify_order(&mut self, command: &ModifyOrder) {
        let order = self
            .cache
            .borrow()
            .order(&command.client_order_id)
            .map(|order| order.clone());

        if let Some(order) = order {
            let price = match command.price {
                Some(price) => Some(price),
                None => order.price(),
            };

            let trigger_price = match command.trigger_price {
                Some(trigger_price) => Some(trigger_price),
                None => order.trigger_price(),
            };

            let ts_now = self.clock.borrow().timestamp_ns();
            let event = OrderUpdated::new(
                order.trader_id(),
                order.strategy_id(),
                order.instrument_id(),
                order.client_order_id(),
                command.quantity.unwrap_or(order.quantity()),
                UUID4::new(),
                ts_now,
                ts_now,
                false,
                order.venue_order_id(),
                order.account_id(),
                price,
                trigger_price,
                None,
                order.is_quote_quantity(),
            );

            let event = OrderEventAny::Updated(event);

            // Apply locally as for cancels, sending to the execution engine would re-enter it
            // when a strategy modifies from an order or position event handler.
            self.apply_order_event(&event);

            // A synchronous event handler may supersede this update before dispatch returns
            let order = self
                .cache
                .borrow()
                .order(&command.client_order_id)
                .filter(|order| order.last_event() == &event)
                .map(|order| order.clone());

            let Some(order) = order else {
                return;
            };

            if !self
                .manager
                .get_submit_order_commands()
                .contains_key(&command.client_order_id)
            {
                return;
            }

            let trigger_instrument_id = order
                .trigger_instrument_id()
                .unwrap_or_else(|| order.instrument_id());

            let (action, match_info) = if let Some(matching_core) =
                self.matching_cores.get_mut(&trigger_instrument_id)
            {
                if let Err(e) = matching_core.delete_order(order.client_order_id()) {
                    log::debug!("Cannot update order match info: {e:?}");
                }

                let is_activated = is_order_activated(&order);

                let match_info = RestingOrder::new_with_trigger_type(
                    order.client_order_id(),
                    order.order_side(),
                    order.order_type(),
                    order.emulation_trigger(),
                    if is_activated {
                        order.trigger_price()
                    } else {
                        None
                    },
                    if is_activated { order.price() } else { None },
                    is_activated,
                );

                (matching_core.match_order(&match_info), match_info)
            } else {
                log::error!(
                    "Cannot handle `ModifyOrder`: no matching core for trigger instrument {trigger_instrument_id}"
                );
                return;
            };

            if let Some(action) = action {
                self.dispatch_match_action(action);
            }

            // Hold in matching core unless the match released the order
            if self
                .manager
                .get_submit_order_commands()
                .contains_key(&command.client_order_id)
                && let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
            {
                matching_core.add_order(match_info);
            }
        } else {
            log::error!("Cannot modify order: {} not found", command.client_order_id);
        }
    }

    fn handle_batch_modify_orders(&mut self, command: &BatchModifyOrders) {
        for modify in &command.modifies {
            self.handle_modify_order(modify);
        }
    }

    pub fn handle_cancel_order(&mut self, command: CancelOrder) {
        let order = if let Some(order) = self.cache.borrow().order(&command.client_order_id) {
            order.clone()
        } else {
            log::error!("Cannot cancel order: {} not found", command.client_order_id);
            return;
        };

        let trigger_instrument_id = order
            .trigger_instrument_id()
            .unwrap_or_else(|| order.instrument_id());

        let matching_core = if let Some(core) = self.matching_cores.get(&trigger_instrument_id) {
            core
        } else {
            let actions = self.manager.cancel_order(&order);
            self.dispatch_manager_actions(actions);
            return;
        };

        if !matching_core.order_exists(order.client_order_id())
            && order.is_open()
            && !order.is_pending_cancel()
        {
            // Order not held in the emulator
            self.send_exec_command(TradingCommand::CancelOrder(command));
        } else {
            let actions = self.manager.cancel_order(&order);
            self.dispatch_manager_actions(actions);
        }
    }

    fn handle_cancel_all_orders(&mut self, command: &CancelAllOrders) {
        let mut ids_to_cancel: Vec<ClientOrderId> = self
            .matching_cores
            .values()
            .flat_map(OrderMatchingCore::iter_orders)
            .map(|order| order.client_order_id)
            .collect();
        ids_to_cancel.sort_unstable();
        ids_to_cancel.dedup();

        {
            let cache = self.cache.borrow();
            ids_to_cancel.retain(|client_order_id| {
                let Some(order) = cache.order(client_order_id) else {
                    return false;
                };

                order.instrument_id() == command.instrument_id
                    && command
                        .order_side
                        .is_none_or(|side| order.order_side() == side)
                    && command.client_id.is_none_or(|client_id| {
                        cache
                            .client_id(client_order_id)
                            .is_some_and(|order_client_id| *order_client_id == client_id)
                    })
            });
        }

        for id in ids_to_cancel {
            let Some(order) = self.cache.borrow().order_owned(&id) else {
                continue;
            };
            let actions = self.manager.cancel_order(&order);
            self.dispatch_manager_actions(actions);
        }
    }

    pub fn update_order(&mut self, order: &mut OrderAny, new_quantity: Quantity) {
        log::info!(
            "Updating order {} quantity to {new_quantity}",
            order.client_order_id(),
        );

        let ts_now = self.clock.borrow().timestamp_ns();
        let event = OrderUpdated::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            new_quantity,
            UUID4::new(),
            ts_now,
            ts_now,
            false,
            None,
            order.account_id(),
            None,
            None,
            None,
            order.is_quote_quantity(),
        );

        let event = OrderEventAny::Updated(event);

        *order = match self.cache.borrow_mut().update_order(&event) {
            Ok(order) => order,
            Err(e) => {
                log::error!("Cannot apply order event: {e:?}");
                return;
            }
        };

        self.send_risk_event(event);
    }

    pub fn on_order_book_deltas(&mut self, deltas: &OrderBookDeltas) {
        log::debug!("Processing {deltas:?}");

        let instrument_id = &deltas.instrument_id;
        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
            if let Some(book) = self.cache.borrow().order_book(instrument_id) {
                let best_bid = book.best_bid_price();
                let best_ask = book.best_ask_price();

                if let Some(best_bid) = best_bid {
                    matching_core.set_bid_raw(best_bid);
                }

                if let Some(best_ask) = best_ask {
                    matching_core.set_ask_raw(best_ask);
                }
            } else {
                log::error!(
                    "Cannot handle `OrderBookDeltas`: no book being maintained for {}",
                    deltas.instrument_id
                );
            }

            self.iterate_orders(instrument_id);
        } else {
            log::error!(
                "Cannot handle `OrderBookDeltas`: no matching core for instrument {}",
                deltas.instrument_id
            );
        }
    }

    pub fn on_quote_tick(&mut self, quote: QuoteTick) {
        log::debug!("Processing {quote}:?");

        let instrument_id = &quote.instrument_id;
        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
            matching_core.set_bid_raw(quote.bid_price);
            matching_core.set_ask_raw(quote.ask_price);

            self.iterate_orders(instrument_id);
        } else {
            log::error!(
                "Cannot handle `QuoteTick`: no matching core for instrument {}",
                quote.instrument_id
            );
        }
    }

    pub fn on_trade_tick(&mut self, trade: TradeTick) {
        log::debug!("Processing {trade:?}");

        let instrument_id = &trade.instrument_id;
        if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
            matching_core.set_last_raw(trade.price);

            if !self.subscribed_quotes.contains(instrument_id) {
                matching_core.set_bid_raw(trade.price);
                matching_core.set_ask_raw(trade.price);
            }

            self.iterate_orders(instrument_id);
        } else {
            log::error!(
                "Cannot handle `TradeTick`: no matching core for instrument {}",
                trade.instrument_id
            );
        }
    }

    fn iterate_orders(&mut self, instrument_id: &InstrumentId) {
        // Process bid actions before ask actions so cross-side
        // contingencies (OCO/OUO) mutate state between sides
        let bid_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
            matching_core.iterate_bids()
        } else {
            log::error!("Cannot iterate orders: no matching core for instrument {instrument_id}");
            return;
        };

        for action in bid_actions {
            self.dispatch_held_match(action, instrument_id);
        }

        let ask_actions = if let Some(matching_core) = self.matching_cores.get_mut(instrument_id) {
            matching_core.iterate_asks()
        } else {
            return;
        };

        for action in ask_actions {
            self.dispatch_held_match(action, instrument_id);
        }

        // Re-snapshot orders after actions to avoid stale trailing stop updates
        let orders = if let Some(matching_core) = self.matching_cores.get(instrument_id) {
            matching_core.get_orders()
        } else {
            return;
        };

        for match_info in orders {
            if !matches!(
                match_info.order_type,
                OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
            ) {
                continue;
            }

            let mut order = match self
                .cache
                .borrow()
                .order(&match_info.client_order_id)
                .map(|o| o.clone())
            {
                Some(order) => order,
                None => continue,
            };

            if order.is_closed() {
                continue;
            }

            self.update_trailing_stop_order(&mut order);
        }

        self.drain_pending_messages();
    }

    fn dispatch_held_match(&mut self, action: MatchAction, instrument_id: &InstrumentId) {
        let client_order_id = match action {
            MatchAction::FillLimit(id) | MatchAction::TriggerStop(id) => id,
        };

        let Some(core) = self.matching_cores.get(instrument_id) else {
            return;
        };

        let Some(order) = core.get_order(client_order_id) else {
            return;
        };

        let Some(action) = core.match_order(order) else {
            return;
        };

        self.dispatch_match_action(action);
    }

    fn dispatch_match_action(&mut self, action: MatchAction) {
        match action {
            MatchAction::FillLimit(id) => self.fill_limit_order(id),
            MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
        }

        self.drain_pending_messages();
    }

    pub fn cancel_order(&mut self, order: &OrderAny) {
        log::info!("Canceling order {}", order.client_order_id());

        let mut order = order.clone();
        order.set_emulation_trigger(None);

        let trigger_instrument_id = order
            .trigger_instrument_id()
            .unwrap_or(order.instrument_id());

        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id)
            && let Err(e) = matching_core.delete_order(order.client_order_id())
        {
            log::debug!("Cannot delete order: {e:?}");
        }

        self.manager
            .pop_submit_order_command(order.client_order_id());

        self.cache
            .borrow_mut()
            .update_order_pending_cancel_local(&order);

        let ts_now = self.clock.borrow().timestamp_ns();
        let event = OrderCanceled::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            UUID4::new(),
            ts_now,
            ts_now,
            false,
            order.venue_order_id(),
            order.account_id(),
            None,
        );

        self.apply_order_event(&OrderEventAny::Canceled(event));
    }

    fn check_monitoring(&mut self, strategy_id: StrategyId, position_id: Option<PositionId>) {
        if !self.subscribed_strategies.contains(&strategy_id) {
            // Subscribe to strategy order events
            if let Some(handler) = &self.on_event_handler {
                msgbus::subscribe_order_events(
                    format!("events.order.{strategy_id}").into(),
                    handler.clone(),
                    None,
                );
                self.subscribed_strategies.insert(strategy_id);
                log::info!("Subscribed to strategy {strategy_id} order events");
            }
        }

        if let Some(position_id) = position_id
            && !self.monitored_positions.contains(&position_id)
        {
            self.monitored_positions.insert(position_id);
        }
    }

    /// # Panics
    ///
    /// Panics if the order type is invalid for a stop order.
    pub fn trigger_stop_order(&mut self, client_order_id: ClientOrderId) {
        let order = match self
            .cache
            .borrow()
            .order(&client_order_id)
            .map(|o| o.clone())
        {
            Some(order) => order,
            None => {
                log::error!(
                    "Cannot trigger stop order: order {client_order_id} not found in cache"
                );
                return;
            }
        };

        match order.order_type() {
            OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
                self.fill_limit_order(client_order_id);
            }
            OrderType::Market
            | OrderType::MarketIfTouched
            | OrderType::StopMarket
            | OrderType::TrailingStopMarket => self.fill_market_order(client_order_id),
            _ => panic!("invalid `OrderType`, was {}", order.order_type()),
        }
    }

    /// Releases an emulated limit-type order as a limit order,
    /// or a plain limit order as a market order.
    ///
    /// Retains the command when market data is unavailable or the limit order cannot be built.
    /// Returns without releasing a limit-type order when its command is missing.
    ///
    /// # Panics
    ///
    /// Panics if a plain limit order is ready for market release but its command is missing.
    pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
        self.release_order(client_order_id, ReleaseType::Limit);
    }

    /// Releases an emulated order as a market order, converting GTD to GTC.
    ///
    /// Retains the command when market data is unavailable.
    ///
    /// # Panics
    ///
    /// Panics if an order is ready for market release but its command is missing.
    pub fn fill_market_order(&mut self, client_order_id: ClientOrderId) {
        self.release_order(client_order_id, ReleaseType::Market);
    }

    fn release_order(&mut self, client_order_id: ClientOrderId, release_type: ReleaseType) {
        let kind = match release_type {
            ReleaseType::Limit => "limit",
            ReleaseType::Market => "market",
        };

        let Some(mut order) = self.cache.borrow().order_owned(&client_order_id) else {
            log::error!("Cannot fill {kind} order: order {client_order_id} not found in cache");
            return;
        };

        if matches!(
            (release_type, order.order_type()),
            (ReleaseType::Limit, OrderType::Limit)
        ) {
            self.fill_market_order(client_order_id);
            return;
        }

        let trigger_instrument_id = order
            .trigger_instrument_id()
            .unwrap_or(order.instrument_id());

        let Some(matching_core) = self.matching_cores.get(&trigger_instrument_id) else {
            log::error!(
                "Cannot fill {kind} order: no matching core for instrument {trigger_instrument_id}"
            );
            return;
        };

        let Some(released_price) =
            self.validate_release(&order, matching_core, trigger_instrument_id)
        else {
            return;
        };

        // Build before taking the command, so a failed transform leaves the order emulated
        let limit_order = match release_type {
            ReleaseType::Limit => {
                let Some(transformed) = self.transform_limit_order(&order, trigger_instrument_id)
                else {
                    return;
                };

                Some(transformed)
            }
            ReleaseType::Market => None,
        };

        let command = self.manager.pop_submit_order_command(client_order_id);

        let command = match release_type {
            ReleaseType::Limit => {
                let Some(command) = command else {
                    return;
                };

                command
            }
            ReleaseType::Market => {
                command.expect("invalid operation `fill_market_order` with no command")
            }
        };

        let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) else {
            return;
        };

        if let Err(e) = matching_core.delete_order(client_order_id) {
            match release_type {
                ReleaseType::Limit => log::debug!("Error deleting order: {e:?}"),
                ReleaseType::Market => log::debug!("Cannot delete order: {e:?}"),
            }
        }

        let transformed = match limit_order {
            Some(mut transformed) => {
                transformed.liquidity_side = order.liquidity_side();

                // TODO: fix
                // let triggered_price = order.trigger_price();
                // if triggered_price.is_some() {
                //     transformed.trigger_price() = (triggered_price.unwrap());
                // }

                transformed.prepend_events(order.events().into_iter().cloned());
                OrderAny::Limit(transformed)
            }
            None => {
                order.set_emulation_trigger(None);
                let mut transformed = self.transform_market_order(&order);
                transformed.prepend_events(order.events().into_iter().cloned());
                OrderAny::Market(transformed)
            }
        };

        self.submit_released_order(&order, &transformed, command, released_price, release_type);
    }

    fn validate_release(
        &self,
        order: &OrderAny,
        matching_core: &OrderMatchingCore,
        trigger_instrument_id: InstrumentId,
    ) -> Option<Price> {
        let released_price = match order.order_side() {
            OrderSide::Buy => matching_core.ask,
            OrderSide::Sell => matching_core.bid,
        };

        if released_price.is_none() {
            log::warn!(
                "Cannot release order {} yet: no market data available for {trigger_instrument_id}, will retry on next update",
                order.client_order_id(),
            );
            return None;
        }

        released_price
    }

    fn transform_limit_order(
        &self,
        order: &OrderAny,
        trigger_instrument_id: InstrumentId,
    ) -> Option<LimitOrder> {
        let client_order_id = order.client_order_id();

        // A trailing stop limit has no limit price until its trailing calculation succeeds
        let Some(price) = order.price() else {
            log::warn!(
                "Cannot release order {client_order_id} yet: no limit price, will retry on next update"
            );
            return None;
        };

        let transformed = match LimitOrder::new_checked(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            order.order_side(),
            order.quantity(),
            price,
            order.time_in_force(),
            order.expire_time(),
            order.is_post_only(),
            order.is_reduce_only(),
            order.is_quote_quantity(),
            order.display_qty(),
            None,
            Some(trigger_instrument_id),
            order.contingency_type(),
            order.order_list_id(),
            order.linked_order_ids().map(Vec::from),
            order.parent_order_id(),
            order.exec_algorithm_id(),
            order.exec_algorithm_params().cloned(),
            order.exec_spawn_id(),
            order.tags().map(Vec::from),
            UUID4::new(),
            self.clock.borrow().timestamp_ns(),
        ) {
            Ok(transformed) => transformed,
            Err(e) => {
                log::error!("Cannot release order {client_order_id} as a limit order: {e}");
                return None;
            }
        };

        Some(transformed)
    }

    fn transform_market_order(&self, order: &OrderAny) -> MarketOrder {
        // Market orders reject GTD, so a released GTD order becomes GTC
        let time_in_force = match order.time_in_force() {
            TimeInForce::Gtd => TimeInForce::Gtc,
            time_in_force => time_in_force,
        };

        MarketOrder::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            order.order_side(),
            order.quantity(),
            time_in_force,
            UUID4::new(),
            self.clock.borrow().timestamp_ns(),
            order.is_reduce_only(),
            order.is_quote_quantity(),
            order.contingency_type(),
            order.order_list_id(),
            order.linked_order_ids().map(Vec::from),
            order.parent_order_id(),
            order.exec_algorithm_id(),
            order.exec_algorithm_params().cloned(),
            order.exec_spawn_id(),
            order.tags().map(Vec::from),
        )
    }

    fn submit_released_order(
        &self,
        order: &OrderAny,
        transformed: &OrderAny,
        mut command: SubmitOrder,
        released_price: Price,
        release_type: ReleaseType,
    ) {
        let client_order_id = order.client_order_id();
        command.order_init = OrderInitialized::from(transformed);
        let add_result = self.cache.borrow_mut().add_order(
            transformed.clone(),
            command.position_id,
            command.client_id,
            true,
        );

        if let Err(e) = add_result {
            log::error!("Failed to add order: {e}");
        } else {
            let topic = get_event_order_topic(order.strategy_id());
            msgbus::publish_order_event(topic, transformed.last_event());
        }

        // Preserve the release branches' distinct clock-read and event-ID ordering
        let (event_id, ts_event, ts_init) = match release_type {
            ReleaseType::Limit => (
                UUID4::new(),
                self.clock.borrow().timestamp_ns(),
                self.clock.borrow().timestamp_ns(),
            ),
            ReleaseType::Market => {
                let ts_now = self.clock.borrow().timestamp_ns();
                (UUID4::new(), ts_now, ts_now)
            }
        };

        let event = OrderEventAny::Released(OrderReleased::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            client_order_id,
            released_price,
            event_id,
            ts_event,
            ts_init,
        ));

        let transformed = match self.cache.borrow_mut().update_order(&event) {
            Ok(order) => order,
            Err(e) => {
                log::error!("Failed to apply order event: {e}");
                return;
            }
        };

        self.send_risk_event(event.clone());

        log::info!("Releasing order {client_order_id}");

        let strategy_id = match release_type {
            ReleaseType::Limit => transformed.strategy_id(),
            ReleaseType::Market => order.strategy_id(),
        };

        let topic = get_event_order_topic(strategy_id);
        msgbus::publish_order_event(topic, &event);

        if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
            self.send_algo_command(command, exec_algorithm_id);
        } else {
            self.send_risk_command(TradingCommand::SubmitOrder(command));
        }
    }

    fn update_trailing_stop_order(&mut self, order: &mut OrderAny) {
        let trigger_instrument_id = order
            .trigger_instrument_id()
            .unwrap_or_else(|| order.instrument_id());

        let Some(matching_core) = self.matching_cores.get(&trigger_instrument_id) else {
            log::error!(
                "Cannot update trailing-stop order: no matching core for instrument {trigger_instrument_id}"
            );
            return;
        };

        let mut bid = matching_core.bid;
        let mut ask = matching_core.ask;
        let mut last = matching_core.last;
        let price_increment = matching_core.price_increment;
        let instrument_id = matching_core.instrument_id;
        if bid.is_none() || ask.is_none() || last.is_none() {
            if let Some(q) = self.cache.borrow().quote(&instrument_id) {
                bid.get_or_insert(q.bid_price);
                ask.get_or_insert(q.ask_price);
            }

            if let Some(t) = self.cache.borrow().trade(&instrument_id) {
                last.get_or_insert(t.price);
            }
        }

        let was_activated = is_order_activated(order);

        if !self.maybe_activate_trailing_stop(order, bid, ask, last) {
            return; // Not yet activated
        }

        if !was_activated {
            // Activation transitioned: re-key the held entry from its inert
            // pre-activation state regardless of the trailing outcome below.
            self.refresh_matching_core_entry(order, trigger_instrument_id);
        }

        let (new_trigger_px, new_limit_px) = match trailing_stop_calculate(
            price_increment,
            order.trigger_price(),
            order,
            bid,
            ask,
            last,
        ) {
            Ok(pair) => pair,
            Err(e) => {
                log::warn!("Cannot calculate trailing-stop update: {e}");
                return;
            }
        };

        if new_trigger_px.is_none() && new_limit_px.is_none() {
            return;
        }

        let ts_now = self.clock.borrow().timestamp_ns();
        let update = OrderUpdated::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            order.quantity(),
            UUID4::new(),
            ts_now,
            ts_now,
            false,
            order.venue_order_id(),
            order.account_id(),
            new_limit_px,
            new_trigger_px,
            None,
            order.is_quote_quantity(),
        );
        let wrapped = OrderEventAny::Updated(update);

        *order = match self.cache.borrow_mut().update_order(&wrapped) {
            Ok(order) => order,
            Err(e) => {
                log::error!("Failed to apply order event: {e}");
                return;
            }
        };

        self.refresh_matching_core_entry(order, trigger_instrument_id);

        self.send_risk_event(wrapped);
    }

    fn refresh_matching_core_entry(
        &mut self,
        order: &OrderAny,
        trigger_instrument_id: InstrumentId,
    ) {
        if let Some(matching_core) = self.matching_cores.get_mut(&trigger_instrument_id) {
            if let Err(e) = matching_core.delete_order(order.client_order_id()) {
                log::debug!("Cannot update trailing-stop match info: {e:?}");
            }

            // A trailing stop with no trigger price yet must stay inert:
            // its limit price must never match as a plain limit order.
            let (trigger_price, limit_price) = if order.trigger_price().is_some() {
                (order.trigger_price(), order.price())
            } else {
                (None, None)
            };

            matching_core.add_order(RestingOrder::new_with_trigger_type(
                order.client_order_id(),
                order.order_side(),
                order.order_type(),
                order.emulation_trigger(),
                trigger_price,
                limit_price,
                is_order_activated(order),
            ));
        }
    }

    /// Returns true if the trailing stop is (or has just become) activated.
    ///
    /// With no `activation_price` the order activates immediately at the current
    /// market price; otherwise it activates only once the market touches the
    /// activation price.
    fn maybe_activate_trailing_stop(
        &self,
        order: &mut OrderAny,
        bid: Option<Price>,
        ask: Option<Price>,
        last: Option<Price>,
    ) -> bool {
        let (is_activated, activation_price, trigger_type, order_side) = match order {
            OrderAny::TrailingStopMarket(inner) => (
                inner.is_activated,
                inner.activation_price,
                inner.trigger_type,
                inner.order_side(),
            ),
            OrderAny::TrailingStopLimit(inner) => (
                inner.is_activated,
                inner.activation_price,
                inner.trigger_type,
                inner.order_side(),
            ),
            _ => return true,
        };

        if is_activated {
            return true;
        }

        if let Some(activation_price) = activation_price {
            let hit = match order_side {
                OrderSide::Buy => ask.is_some_and(|a| a <= activation_price),
                OrderSide::Sell => bid.is_some_and(|b| b >= activation_price),
            };

            if hit {
                Self::set_trailing_stop_activated(order, None);
                store_trailing_stop_activation(&mut self.cache.borrow_mut(), order);
            }
            return hit;
        }

        let market_price = match trigger_type {
            TriggerType::LastPrice => last,
            _ => match order_side {
                OrderSide::Buy => ask,
                OrderSide::Sell => bid,
            },
        };

        let Some(market_price) = market_price else {
            log::error!(
                "Cannot activate trailing stop {}: no market price available",
                order.client_order_id()
            );
            return false;
        };

        Self::set_trailing_stop_activated(order, Some(market_price));
        store_trailing_stop_activation(&mut self.cache.borrow_mut(), order);
        true
    }

    fn set_trailing_stop_activated(order: &mut OrderAny, activation_price: Option<Price>) {
        match order {
            OrderAny::TrailingStopMarket(inner) => {
                if let Some(price) = activation_price {
                    inner.activation_price = Some(price);
                }
                inner.set_activated();
            }
            OrderAny::TrailingStopLimit(inner) => {
                if let Some(price) = activation_price {
                    inner.activation_price = Some(price);
                }
                inner.set_activated();
            }
            _ => {}
        }
    }

    fn send_algo_command(&self, command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
        let id = command.strategy_id;
        log::info!("{id} {CMD}{SEND} {command}");

        let endpoint = format!("{exec_algorithm_id}.execute");
        msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
    }

    fn send_risk_command(&self, command: TradingCommand) {
        log_cmd_send(&command);
        let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
        msgbus::send_trading_command(endpoint, command);
    }

    fn send_exec_command(&self, command: TradingCommand) {
        log_cmd_send(&command);
        let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
        msgbus::send_trading_command(endpoint, command);
    }

    fn send_risk_event(&self, event: OrderEventAny) {
        log_evt_send(&event);
        let endpoint = MessagingSwitchboard::risk_engine_process();
        msgbus::send_order_event(endpoint, event);
    }

    // Applies an event the emulator generated, then notifies the portfolio and the strategy
    fn apply_order_event(&self, event: &OrderEventAny) {
        if let Err(e) = self.cache.borrow_mut().update_order(event) {
            log::error!("Failed to apply order event: {e}");
            return;
        }

        self.send_portfolio_order_event(event.clone());
        publish_order_event(event);
    }

    fn send_portfolio_order_event(&self, event: OrderEventAny) {
        log_evt_send(&event);
        let endpoint = MessagingSwitchboard::portfolio_update_order();
        msgbus::send_order_event(endpoint, event);
    }

    fn send_data_command(&self, command: DataCommand) {
        log::info!("{CMD}{SEND} {command:?}");
        let endpoint = MessagingSwitchboard::data_engine_queue_execute();
        msgbus::send_data_command(endpoint, command);
    }
}

#[derive(Clone, Copy)]
enum ReleaseType {
    Limit,
    Market,
}

fn publish_order_event(event: &OrderEventAny) {
    msgbus::publish_order_event(get_event_order_topic(event.strategy_id()), event);

    if let OrderEventAny::Canceled(_) = event {
        msgbus::publish_order_event(get_order_canceled_topic(event.instrument_id()), event);
    }
}

fn is_order_activated(order: &OrderAny) -> bool {
    match order {
        OrderAny::TrailingStopMarket(o) => o.is_activated,
        OrderAny::TrailingStopLimit(o) => o.is_activated,
        _ => true,
    }
}

fn log_cmd_send(command: &TradingCommand) {
    if let Some(id) = command.strategy_id() {
        log::info!("{id} {CMD}{SEND} {command}");
    } else {
        log::info!("{CMD}{SEND} {command}");
    }
}

fn log_evt_send(event: &OrderEventAny) {
    let id = event.strategy_id();
    log::info!("{id} {EVT}{SEND} {event}");
}

#[cfg(test)]
mod tests {
    use std::{cell::RefCell, rc::Rc};

    use nautilus_common::{
        cache::Cache,
        clock::VirtualClock,
        messages::data::{DataCommand, SubscribeCommand, UnsubscribeCommand},
        msgbus::{
            MessagingSwitchboard,
            stubs::{
                TypedIntoMessageSavingHandler, get_any_saving_handler,
                get_typed_into_message_saving_handler,
            },
        },
    };
    use nautilus_core::{Params, UUID4, UnixNanos};
    use nautilus_model::{
        data::{QuoteTick, TradeTick},
        enums::{
            AggressorSide, OrderSide, OrderType, TimeInForce, TrailingOffsetType, TriggerType,
        },
        identifiers::{
            AccountId, ClientId, ClientOrderId, OrderListId, StrategyId, Symbol, TradeId, TraderId,
        },
        instruments::{
            CryptoPerpetual, Instrument, InstrumentAny, SyntheticInstrument,
            stubs::{btcusd_bybit, crypto_perpetual_ethusdt},
        },
        orders::{OrderList, OrderTestBuilder, stubs::TestOrderEventStubs},
        types::{Price, Quantity},
    };
    use rstest::{fixture, rstest};
    use rust_decimal_macros::dec;
    use ustr::Ustr;

    use super::*;

    #[fixture]
    fn instrument() -> CryptoPerpetual {
        crypto_perpetual_ethusdt()
    }

    #[expect(clippy::type_complexity)]
    fn create_emulator() -> (
        Rc<RefCell<dyn Clock>>,
        Rc<RefCell<Cache>>,
        Rc<RefCell<OrderEmulator>>,
    ) {
        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
        let cache = Rc::new(RefCell::new(Cache::new(None, None)));
        let emulator = Rc::new(RefCell::new(OrderEmulator::new(
            clock.clone(),
            cache.clone(),
        )));

        OrderEmulator::register_msgbus_handlers(&emulator);

        (clock, cache, emulator)
    }

    fn create_stop_market_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
        OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(trigger)
            .build()
    }

    fn create_stop_limit_order(instrument: &CryptoPerpetual, trigger: TriggerType) -> OrderAny {
        OrderTestBuilder::new(OrderType::StopLimit)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5100.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(trigger)
            .build()
    }

    fn create_list_stop_market_order(
        instrument: &CryptoPerpetual,
        client_order_id: &str,
        order_list_id: OrderListId,
    ) -> OrderAny {
        OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from(client_order_id))
            .order_list_id(order_list_id)
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .build()
    }

    fn create_submit_order(instrument: &CryptoPerpetual, order: &OrderAny) -> SubmitOrder {
        SubmitOrder::new(
            TraderId::from("TRADER-001"),
            None,
            StrategyId::from("STRATEGY-001"),
            instrument.id(),
            order.client_order_id(),
            order.init_event().clone(),
            None,
            None,
            None,
            UUID4::new(),
            0.into(),
            None, // correlation_id
        )
    }

    fn create_quote_tick(instrument: &CryptoPerpetual, bid: &str, ask: &str) -> QuoteTick {
        QuoteTick::new(
            instrument.id(),
            Price::from(bid),
            Price::from(ask),
            Quantity::from(10),
            Quantity::from(10),
            0.into(),
            0.into(),
        )
    }

    fn create_trade_tick(instrument: &CryptoPerpetual, price: &str) -> TradeTick {
        TradeTick::new(
            instrument.id(),
            Price::from(price),
            Quantity::from(1),
            AggressorSide::Buy,
            TradeId::from("T-001"),
            0.into(),
            0.into(),
        )
    }

    fn add_instrument_to_cache(cache: &Rc<RefCell<Cache>>, instrument: &CryptoPerpetual) {
        cache
            .borrow_mut()
            .add_instrument(InstrumentAny::CryptoPerpetual(instrument.clone()))
            .unwrap();
    }

    fn register_risk_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
        let (handler, saving_handler) =
            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
        msgbus::register_order_event_endpoint(MessagingSwitchboard::risk_engine_process(), handler);
        saving_handler
    }

    fn register_exec_event_handler(
        cache: Rc<RefCell<Cache>>,
        id: &str,
    ) -> TypedIntoMessageSavingHandler<OrderEventAny> {
        let messages = Rc::new(RefCell::new(Vec::new()));
        let messages_for_handler = messages.clone();
        msgbus::register_order_event_endpoint(
            MessagingSwitchboard::exec_engine_process(),
            TypedIntoHandler::from(move |event: OrderEventAny| {
                cache.borrow_mut().update_order(&event).unwrap();
                messages_for_handler.borrow_mut().push(event);
            }),
        );
        TypedIntoMessageSavingHandler::new_with_messages(Some(Ustr::from(id)), messages)
    }

    fn register_portfolio_event_handler(id: &str) -> TypedIntoMessageSavingHandler<OrderEventAny> {
        let (handler, saving_handler) =
            get_typed_into_message_saving_handler::<OrderEventAny>(Some(Ustr::from(id)));
        msgbus::register_order_event_endpoint(
            MessagingSwitchboard::portfolio_update_order(),
            handler,
        );
        saving_handler
    }

    fn register_data_command_handler(id: &str) -> TypedIntoMessageSavingHandler<DataCommand> {
        let (handler, saving_handler) =
            get_typed_into_message_saving_handler::<DataCommand>(Some(Ustr::from(id)));
        msgbus::register_data_command_endpoint(
            MessagingSwitchboard::data_engine_queue_execute(),
            handler,
        );
        saving_handler
    }

    fn subscribe_order_topic(
        strategy_id: StrategyId,
    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
        let events = Rc::new(RefCell::new(Vec::new()));
        let handler = TypedHandler::from({
            let events = events.clone();
            move |event: &OrderEventAny| {
                events.borrow_mut().push(event.clone());
            }
        });
        msgbus::subscribe_order_events(
            format!("events.order.{strategy_id}").into(),
            handler.clone(),
            None,
        );
        (handler, events)
    }

    fn subscribe_order_cancel_topic(
        instrument_id: InstrumentId,
    ) -> (TypedHandler<OrderEventAny>, Rc<RefCell<Vec<OrderEventAny>>>) {
        let events = Rc::new(RefCell::new(Vec::new()));
        let handler = TypedHandler::from({
            let events = events.clone();
            move |event: &OrderEventAny| {
                events.borrow_mut().push(event.clone());
            }
        });
        msgbus::subscribe_order_events(
            get_order_canceled_topic(instrument_id).into(),
            handler.clone(),
            None,
        );
        (handler, events)
    }

    #[rstest]
    fn test_dispatch_manager_publish_initialized_publishes_order_event(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, _cache, emulator) = create_emulator();
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let strategy_id = order.strategy_id();
        let client_order_id = order.client_order_id();
        let event = OrderEventAny::Initialized(order.init_event().clone());
        let (order_handler, order_events) = subscribe_order_topic(strategy_id);

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::PublishInitialized(event));
        msgbus::unsubscribe_order_events(
            format!("events.order.{strategy_id}").into(),
            &order_handler,
        );
        let order_events = order_events.borrow();

        assert_eq!(order_events.len(), 1);
        assert!(matches!(
            &order_events[0],
            OrderEventAny::Initialized(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_dispatch_manager_submit_to_emulator_applies_emulated_event(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_emulated");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::SubmitToEmulator(command));
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();

        assert_eq!(cached_order.status(), OrderStatus::Emulated);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
    }

    #[rstest]
    fn test_registered_execute_endpoint_routes_submit_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.endpoint_emulated");
        let data_commands = register_data_command_handler("DataEngine.queue_execute.endpoint");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());

        msgbus::send_trading_command(
            MessagingSwitchboard::order_emulator_execute(),
            TradingCommand::SubmitOrder(command),
        );
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();
        let data_commands = data_commands.get_messages();

        assert_eq!(cached_order.status(), OrderStatus::Emulated);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
        assert_eq!(data_commands.len(), 1);
        assert!(matches!(
            &data_commands[0],
            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
                if command.instrument_ids == [instrument.id()]
        ));
    }

    #[rstest]
    fn test_cancel_all_orders_filters_emulated_orders_by_client(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let _risk_events = register_risk_event_handler("RiskEngine.process.cancel_all_client");
        let portfolio_events =
            register_portfolio_event_handler("Portfolio.update_order.cancel_all_client");
        let _data_commands = register_data_command_handler("DataEngine.queue_execute.cancel_all");
        add_instrument_to_cache(&cache, &instrument);

        let selected_client = ClientId::from("CLIENT-001");
        let other_client = ClientId::from("CLIENT-002");
        let make_order = |client_order_id| {
            OrderTestBuilder::new(OrderType::StopMarket)
                .instrument_id(instrument.id())
                .client_order_id(client_order_id)
                .side(OrderSide::Buy)
                .trigger_price(Price::from("5100.00"))
                .quantity(Quantity::from(1))
                .emulation_trigger(TriggerType::BidAsk)
                .build()
        };
        let selected_order = make_order(ClientOrderId::from("O-EMULATED-SELECTED"));
        let other_order = make_order(ClientOrderId::from("O-EMULATED-OTHER"));
        let unclaimed_order = make_order(ClientOrderId::from("O-EMULATED-UNCLAIMED"));
        let synthetic_formula = format!("{} * 1.0", instrument.id());
        let synthetic = SyntheticInstrument::builder()
            .symbol(Symbol::from("ETH-INDEX"))
            .price_precision(instrument.price_precision())
            .components(vec![instrument.id()])
            .formula(&synthetic_formula)
            .ts_event(0.into())
            .ts_init(0.into())
            .build()
            .unwrap();
        let trigger_instrument_id = synthetic.id;
        let cross_trigger_order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-EMULATED-CROSS-TRIGGER"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .trigger_instrument_id(trigger_instrument_id)
            .build();
        let mut selected_submit = create_submit_order(&instrument, &selected_order);
        selected_submit.client_id = Some(selected_client);
        let mut other_submit = create_submit_order(&instrument, &other_order);
        other_submit.client_id = Some(other_client);
        let unclaimed_submit = create_submit_order(&instrument, &unclaimed_order);
        let mut cross_trigger_submit = create_submit_order(&instrument, &cross_trigger_order);
        cross_trigger_submit.client_id = Some(selected_client);
        {
            let mut cache = cache.borrow_mut();
            cache.add_synthetic(synthetic).unwrap();
            cache
                .add_order(selected_order.clone(), None, Some(selected_client), false)
                .unwrap();
            cache
                .add_order(other_order.clone(), None, Some(other_client), false)
                .unwrap();
            cache
                .add_order(unclaimed_order.clone(), None, None, false)
                .unwrap();
            cache
                .add_order(
                    cross_trigger_order.clone(),
                    None,
                    Some(selected_client),
                    false,
                )
                .unwrap();
        }
        emulator.borrow_mut().handle_submit_order(&selected_submit);
        emulator.borrow_mut().handle_submit_order(&other_submit);
        emulator.borrow_mut().handle_submit_order(&unclaimed_submit);
        emulator
            .borrow_mut()
            .handle_submit_order(&cross_trigger_submit);

        emulator
            .borrow_mut()
            .execute(TradingCommand::CancelAllOrders(CancelAllOrders::new(
                TraderId::from("TRADER-001"),
                Some(selected_client),
                StrategyId::from("CALLER-001"),
                trigger_instrument_id,
                None,
                UUID4::new(),
                0.into(),
                None,
                None,
            )));

        assert_eq!(
            cache
                .borrow()
                .order(&cross_trigger_order.client_order_id())
                .unwrap()
                .status(),
            OrderStatus::Emulated
        );

        emulator
            .borrow_mut()
            .execute(TradingCommand::CancelAllOrders(CancelAllOrders::new(
                TraderId::from("TRADER-001"),
                Some(selected_client),
                StrategyId::from("CALLER-001"),
                instrument.id(),
                None,
                UUID4::new(),
                0.into(),
                None,
                None,
            )));

        let cache = cache.borrow();
        assert_eq!(
            cache
                .order(&selected_order.client_order_id())
                .unwrap()
                .status(),
            OrderStatus::Canceled
        );
        assert_eq!(
            cache
                .order(&other_order.client_order_id())
                .unwrap()
                .status(),
            OrderStatus::Emulated
        );
        assert_eq!(
            cache
                .order(&unclaimed_order.client_order_id())
                .unwrap()
                .status(),
            OrderStatus::Emulated
        );
        assert_eq!(
            cache
                .order(&cross_trigger_order.client_order_id())
                .unwrap()
                .status(),
            OrderStatus::Canceled
        );
        let portfolio_events = portfolio_events.get_messages();
        assert_eq!(portfolio_events.len(), 2);
        let canceled_ids: AHashSet<_> = portfolio_events
            .iter()
            .filter_map(|event| match event {
                OrderEventAny::Canceled(event) => Some(event.client_order_id),
                _ => None,
            })
            .collect();
        assert_eq!(
            canceled_ids,
            AHashSet::from_iter([
                selected_order.client_order_id(),
                cross_trigger_order.client_order_id(),
            ])
        );
    }

    #[rstest]
    fn test_dispatch_manager_submit_to_risk_uses_risk_queue(instrument: CryptoPerpetual) {
        let (_clock, _cache, emulator) = create_emulator();
        let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        let order = create_stop_market_order(&instrument, TriggerType::Default);
        let command = create_submit_order(&instrument, &order);
        let client_order_id = command.client_order_id;

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::SubmitToRisk(command));

        let messages = messages.get_messages();
        assert_eq!(messages.len(), 1);
        assert!(matches!(
            messages.first(),
            Some(TradingCommand::SubmitOrder(command))
                if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_dispatch_manager_submit_to_algorithm_uses_dynamic_endpoint(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, _cache, emulator) = create_emulator();
        let exec_algorithm_id = ExecAlgorithmId::from("ALG-001");
        let endpoint = format!("{exec_algorithm_id}.execute");
        let (handler, messages) =
            get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALG-001.execute")));
        msgbus::register_any(endpoint.into(), handler);
        let order = create_stop_market_order(&instrument, TriggerType::Default);
        let command = create_submit_order(&instrument, &order);
        let client_order_id = command.client_order_id;

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::SubmitToAlgorithm {
                command,
                exec_algorithm_id,
            });

        let messages = messages.get_messages();
        assert_eq!(messages.len(), 1);
        assert!(matches!(
            messages.first(),
            Some(TradingCommand::SubmitOrder(command))
                if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_dispatch_manager_cancel_local_applies_and_publishes_event(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let portfolio_events =
            register_portfolio_event_handler("Portfolio.update_order.dispatch_canceled");
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let instrument_id = order.instrument_id();
        let command = create_submit_order(&instrument, &order);
        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
        let (cancel_handler, cancel_events) = subscribe_order_cancel_topic(instrument_id);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator.borrow_mut().cache_submit_order_command(command);

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::CancelLocal(order));
        msgbus::unsubscribe_order_events(get_event_order_topic(strategy_id).into(), &order_handler);
        msgbus::unsubscribe_order_events(
            get_order_canceled_topic(instrument_id).into(),
            &cancel_handler,
        );
        let cached_order = cache
            .borrow()
            .order(&client_order_id)
            .map(|order| order.clone())
            .unwrap();
        let portfolio_events = portfolio_events.get_messages();
        let order_events = order_events.borrow();
        let cancel_events = cancel_events.borrow();
        let commands = emulator.borrow().get_submit_order_commands();

        assert_eq!(cached_order.status(), OrderStatus::Canceled);
        assert_eq!(portfolio_events.len(), 1);
        assert_eq!(order_events.len(), 1);
        assert_eq!(cancel_events.len(), 1);
        assert!(matches!(
            &portfolio_events[0],
            OrderEventAny::Canceled(event) if event.client_order_id == client_order_id
        ));
        assert_eq!(order_events[0].client_order_id(), client_order_id);
        assert_eq!(cancel_events[0].client_order_id(), client_order_id);
        assert!(!commands.contains_key(&client_order_id));
    }

    #[rstest]
    fn test_dispatch_manager_modify_local_quantity_updates_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.dispatch_updated");
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let quantity = Quantity::from(2);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .dispatch_manager_action(OrderManagerAction::ModifyLocalQuantity { order, quantity });
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();

        assert_eq!(cached_order.quantity(), quantity);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(
            &risk_events[0],
            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_subscribed_quotes_initially_empty() {
        let (_clock, _cache, emulator) = create_emulator();

        assert!(emulator.borrow().subscribed_quotes().is_empty());
    }

    #[rstest]
    fn test_subscribed_trades_initially_empty() {
        let (_clock, _cache, emulator) = create_emulator();

        assert!(emulator.borrow().subscribed_trades().is_empty());
    }

    #[rstest]
    fn test_get_submit_order_commands_initially_empty() {
        let (_clock, _cache, emulator) = create_emulator();

        assert!(emulator.borrow().get_submit_order_commands().is_empty());
    }

    #[rstest]
    fn test_get_matching_core_returns_none_when_not_created(instrument: CryptoPerpetual) {
        let (_clock, _cache, emulator) = create_emulator();

        assert!(
            emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .is_none()
        );
    }

    #[rstest]
    fn test_create_matching_core(instrument: CryptoPerpetual) {
        let (_clock, _cache, emulator) = create_emulator();

        emulator.borrow_mut().create_matching_core(
            instrument.id(),
            instrument.price_increment,
            TriggerType::BidAsk,
        );

        assert!(
            emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .is_some()
        );
    }

    #[rstest]
    fn test_on_quote_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
        let (_clock, _cache, emulator) = create_emulator();
        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");

        emulator.borrow_mut().on_quote_tick(quote);
    }

    #[rstest]
    fn test_on_trade_tick_no_matching_core_does_not_panic(instrument: CryptoPerpetual) {
        let (_clock, _cache, emulator) = create_emulator();
        let trade = create_trade_tick(&instrument, "5065.00");

        emulator.borrow_mut().on_trade_tick(trade);
    }

    #[rstest]
    fn test_submit_order_bid_ask_trigger_creates_matching_core(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        assert!(
            emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .is_some()
        );
    }

    #[rstest]
    fn test_submit_order_bid_ask_trigger_tracks_quote_subscription(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let data_commands = register_data_command_handler("DataEngine.queue_execute.quotes");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        assert_eq!(emulator.borrow().subscribed_quotes(), vec![instrument.id()]);
        assert!(emulator.borrow().subscribed_trades().is_empty());

        let commands = data_commands.get_messages();
        assert_eq!(commands.len(), 1);
        assert!(matches!(
            &commands[0],
            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
                if command.instrument_ids == [instrument.id()]
        ));

        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert_eq!(core.bid, Some(Price::from("5060.00")));
        assert_eq!(core.ask, Some(Price::from("5070.00")));
    }

    #[rstest]
    fn test_submit_order_last_price_trigger_tracks_trade_subscription(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let data_commands = register_data_command_handler("DataEngine.queue_execute.trades");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        assert!(emulator.borrow().subscribed_quotes().is_empty());
        assert_eq!(emulator.borrow().subscribed_trades(), vec![instrument.id()]);

        let commands = data_commands.get_messages();
        assert_eq!(commands.len(), 1);
        assert!(matches!(
            &commands[0],
            DataCommand::Subscribe(SubscribeCommand::Trades(command))
                if command.instrument_ids == [instrument.id()]
        ));

        let trade = create_trade_tick(&instrument, "5065.00");
        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert_eq!(core.last, Some(Price::from("5065.00")));
    }

    #[rstest]
    fn test_reset_unsubscribes_market_data_and_clears_state(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let data_commands = register_data_command_handler("DataEngine.queue_execute.reset");
        add_instrument_to_cache(&cache, &instrument);
        let quote_order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let quote_command = create_submit_order(&instrument, &quote_order);
        let trade_order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-RESET-TRADE"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::LastPrice)
            .build();
        let trade_command = create_submit_order(&instrument, &trade_order);
        cache
            .borrow_mut()
            .add_order(quote_order, None, None, false)
            .unwrap();
        cache
            .borrow_mut()
            .add_order(trade_order, None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(quote_command.clone());
        emulator.borrow_mut().handle_submit_order(&quote_command);
        emulator
            .borrow_mut()
            .cache_submit_order_command(trade_command.clone());
        emulator.borrow_mut().handle_submit_order(&trade_command);
        data_commands.clear();

        emulator.borrow_mut().reset();
        let commands = data_commands.get_messages();
        let emulator_ref = emulator.borrow();

        assert!(emulator_ref.subscribed_quotes.is_empty());
        assert!(emulator_ref.subscribed_trades.is_empty());
        assert!(emulator_ref.subscribed_strategies.is_empty());
        assert!(emulator_ref.monitored_positions.is_empty());
        assert!(emulator_ref.quote_handlers.is_empty());
        assert!(emulator_ref.trade_handlers.is_empty());
        assert!(emulator_ref.get_submit_order_commands().is_empty());
        assert!(emulator_ref.get_matching_core(&instrument.id()).is_none());
        assert!(commands.iter().any(|command| matches!(
            command,
            DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command))
                if command.instrument_ids == [instrument.id()]
        )));
        assert!(commands.iter().any(|command| matches!(
            command,
            DataCommand::Unsubscribe(UnsubscribeCommand::Trades(command))
                if command.instrument_ids == [instrument.id()]
        )));

        drop(emulator_ref);
        emulator.borrow_mut().create_matching_core(
            instrument.id(),
            instrument.price_increment,
            TriggerType::BidAsk,
        );
        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
        let trade = create_trade_tick(&instrument, "5065.00");
        msgbus::publish_quote(get_quotes_topic(instrument.id()), &quote);
        msgbus::publish_trade(get_trades_topic(instrument.id()), &trade);
        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert_eq!(core.bid, None);
        assert_eq!(core.ask, None);
        assert_eq!(core.last, None);
    }

    #[rstest]
    fn test_reset_unsubscribes_in_sorted_instrument_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let data_commands = register_data_command_handler("DataEngine.queue_execute.reset_order");

        // Registered in an order that is neither sorted nor reverse sorted, and five
        // entries leave a 1-in-120 chance that AHash order coincides with the expectation.
        let ids = [
            "SOLUSDT-PERP.BINANCE",
            "ADAUSDT-PERP.BINANCE",
            "XRPUSDT-PERP.BINANCE",
            "BTCUSDT-PERP.BINANCE",
            "DOTUSDT-PERP.BINANCE",
        ];

        for (i, id) in ids.iter().enumerate() {
            let mut variant = instrument.clone();
            variant.id = InstrumentId::from(*id);
            add_instrument_to_cache(&cache, &variant);

            let order = OrderTestBuilder::new(OrderType::StopMarket)
                .instrument_id(variant.id)
                .client_order_id(ClientOrderId::from(format!("O-RESET-{i}").as_str()))
                .side(OrderSide::Buy)
                .trigger_price(Price::from("5100.00"))
                .quantity(Quantity::from(1))
                .emulation_trigger(TriggerType::BidAsk)
                .build();
            let command = create_submit_order(&variant, &order);
            cache
                .borrow_mut()
                .add_order(order, None, None, false)
                .unwrap();
            emulator
                .borrow_mut()
                .cache_submit_order_command(command.clone());
            emulator.borrow_mut().handle_submit_order(&command);
        }
        data_commands.clear();

        emulator.borrow_mut().reset();

        let unsubscribed: Vec<InstrumentId> = data_commands
            .get_messages()
            .iter()
            .filter_map(|command| match command {
                DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(command)) => {
                    Some(command.single_instrument_id().unwrap())
                }
                _ => None,
            })
            .collect();

        let mut expected: Vec<InstrumentId> =
            ids.iter().map(|id| InstrumentId::from(*id)).collect();
        expected.sort();

        assert_eq!(unsubscribed, expected);
    }

    #[rstest]
    fn test_submit_order_list_handles_reentrant_order_events(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let data_commands = register_data_command_handler("DataEngine.queue_execute.list");
        add_instrument_to_cache(&cache, &instrument);
        let order_list_id = OrderListId::from("OL-EMULATOR-001");
        let first_order = create_list_stop_market_order(&instrument, "O-LIST-001", order_list_id);
        let second_order = create_list_stop_market_order(&instrument, "O-LIST-002", order_list_id);
        let orders = vec![first_order.clone(), second_order.clone()];
        let order_list = OrderList::from_orders(&orders, 0.into());
        let order_inits = orders
            .iter()
            .map(|order| order.init_event().clone())
            .collect();
        cache
            .borrow_mut()
            .add_order(first_order.clone(), None, None, false)
            .unwrap();
        cache
            .borrow_mut()
            .add_order(second_order.clone(), None, None, false)
            .unwrap();
        let command = SubmitOrderList::new(
            TraderId::from("TRADER-001"),
            None,
            StrategyId::from("STRATEGY-001"),
            order_list,
            order_inits,
            None,
            None,
            None,
            UUID4::new(),
            0.into(),
            None,
        );

        emulator
            .borrow_mut()
            .execute(TradingCommand::SubmitOrderList(command));

        let commands = data_commands.get_messages();
        let cache = cache.borrow();
        let first_status = cache
            .order(&first_order.client_order_id())
            .unwrap()
            .status();
        let second_status = cache
            .order(&second_order.client_order_id())
            .unwrap()
            .status();
        drop(cache);
        let emulator = emulator.borrow();

        assert_eq!(first_status, OrderStatus::Emulated);
        assert_eq!(second_status, OrderStatus::Emulated);
        assert!(emulator.get_matching_core(&instrument.id()).is_some());
        assert_eq!(emulator.subscribed_quotes(), vec![instrument.id()]);
        assert!(commands.iter().any(|command| matches!(
            command,
            DataCommand::Subscribe(SubscribeCommand::Quotes(command))
                if command.instrument_ids == [instrument.id()]
        )));
    }

    #[rstest]
    fn test_submit_order_caches_command(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        let commands = emulator.borrow().get_submit_order_commands();
        assert!(commands.contains_key(&client_order_id));
    }

    #[rstest]
    fn test_handle_submit_order_applies_emulated_event_to_cache(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.emulated");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();
        let (order_handler, order_events) = subscribe_order_topic(strategy_id);

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
        msgbus::unsubscribe_order_events(
            format!("events.order.{strategy_id}").into(),
            &order_handler,
        );
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();
        let order_events = order_events.borrow();

        assert_eq!(cached_order.status(), OrderStatus::Emulated);
        assert_eq!(cached_order.event_count(), 2);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Emulated(_)));
        assert_eq!(order_events.len(), 1);
        assert!(matches!(order_events[0], OrderEventAny::Emulated(_)));
    }

    #[rstest]
    fn test_submit_order_missing_synthetic_trigger_cancels_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let synthetic_id = InstrumentId::from("BTC-ETH-INDEX.SYNTH");
        let order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .trigger_instrument_id(synthetic_id)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let commands = emulator.get_submit_order_commands();

        assert_eq!(cached_order.status(), OrderStatus::Canceled);
        assert!(emulator.get_matching_core(&synthetic_id).is_none());
        assert!(emulator.subscribed_quotes().is_empty());
        assert!(emulator.subscribed_trades().is_empty());
        assert!(!commands.contains_key(&client_order_id));
    }

    #[rstest]
    fn test_submit_order_releases_when_immediately_triggered(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events =
            register_risk_event_handler("RiskEngine.process.submit_immediately_triggered");
        add_instrument_to_cache(&cache, &instrument);
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5099.00"));
        core.set_ask_raw(Price::from("5101.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
        let risk_events = risk_events.get_messages();
        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(cached_order.order_type(), OrderType::Market);
        assert!(!matching_core.order_exists(client_order_id));
        assert_eq!(emulator.subscribed_strategy_count(), 1);
        assert!(
            !emulator
                .get_submit_order_commands()
                .contains_key(&client_order_id)
        );
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(
            &risk_events[0],
            OrderEventAny::Released(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    #[case::bid_ask_quote_below(TriggerType::BidAsk, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(false), Some(true)))]
    #[case::bid_ask_quote_above(TriggerType::BidAsk, Some(("5101.00", "5102.00")), Some("5099.00"), (Some(true), Some(false)))]
    #[case::default_quote(TriggerType::Default, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(false), Some(true)))]
    #[case::last_price_trade_below(TriggerType::LastPrice, Some(("5101.00", "5102.00")), Some("5099.00"), (Some(false), Some(true)))]
    #[case::last_price_trade_above(TriggerType::LastPrice, Some(("5098.00", "5099.00")), Some("5101.00"), (Some(true), Some(false)))]
    #[case::last_price_quote_only(TriggerType::LastPrice, Some(("5098.00", "5099.00")), None, (None, None))]
    #[case::last_price_trade_only(
        TriggerType::LastPrice,
        None,
        Some("5101.00"),
        (Some(true), Some(false))
    )]
    #[case::bid_ask_trade_only(TriggerType::BidAsk, None, Some("5101.00"), (None, None))]
    #[case::default_trade_only(TriggerType::Default, None, Some("5101.00"), (None, None))]
    #[case::bid_ask_no_data(TriggerType::BidAsk, None, None, (None, None))]
    #[case::default_no_data(TriggerType::Default, None, None, (None, None))]
    #[case::last_price_no_data(TriggerType::LastPrice, None, None, (None, None))]
    fn test_submit_order_uses_cached_emulation_source(
        instrument: CryptoPerpetual,
        #[case] emulation_trigger: TriggerType,
        #[case] quote: Option<(&str, &str)>,
        #[case] trade: Option<&str>,
        #[case] stop_matches: (Option<bool>, Option<bool>),
        #[values(
            OrderType::Limit,
            OrderType::StopMarket,
            OrderType::StopLimit,
            OrderType::MarketIfTouched,
            OrderType::LimitIfTouched
        )]
        order_type: OrderType,
        #[values(OrderSide::Buy, OrderSide::Sell)] side: OrderSide,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.submit_cached_source");
        add_instrument_to_cache(&cache, &instrument);
        if let Some((bid, ask)) = quote {
            cache
                .borrow_mut()
                .add_quote(create_quote_tick(&instrument, bid, ask))
                .unwrap();
        }

        if let Some(price) = trade {
            cache
                .borrow_mut()
                .add_trade(create_trade_tick(&instrument, price))
                .unwrap();
        }

        let order = OrderTestBuilder::new(order_type)
            .instrument_id(instrument.id())
            .side(side)
            .price(Price::from("5100.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .emulation_trigger(emulation_trigger)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);

        let stop_matches = match side {
            OrderSide::Buy => stop_matches.0,
            OrderSide::Sell => stop_matches.1,
        };

        let should_release = stop_matches.is_some_and(|matches| match order_type {
            OrderType::StopMarket | OrderType::StopLimit => matches,
            _ => !matches,
        });

        let expected_status = if should_release {
            OrderStatus::Released
        } else {
            OrderStatus::Emulated
        };

        let expected_type = if should_release {
            match order_type {
                OrderType::StopLimit | OrderType::LimitIfTouched => OrderType::Limit,
                _ => OrderType::Market,
            }
        } else {
            order_type
        };

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let core = emulator.get_matching_core(&instrument.id()).unwrap();
        let events = risk_events.get_messages();
        assert_eq!(cached_order.status(), expected_status);
        assert_eq!(cached_order.order_type(), expected_type);
        assert_eq!(cached_order.quantity(), Quantity::from(3));
        assert_eq!(core.order_exists(client_order_id), !should_release);
        assert_eq!(
            emulator
                .get_submit_order_commands()
                .contains_key(&client_order_id),
            !should_release
        );
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].client_order_id(), client_order_id);
        assert_eq!(
            matches!(&events[0], OrderEventAny::Released(_)),
            should_release
        );
        assert_eq!(
            matches!(&events[0], OrderEventAny::Emulated(_)),
            !should_release
        );

        let expected_price = match (emulation_trigger, side) {
            (TriggerType::LastPrice, _) => trade.map(Price::from),
            (_, OrderSide::Buy) => quote.map(|(_, ask)| Price::from(ask)),
            (_, OrderSide::Sell) => quote.map(|(bid, _)| Price::from(bid)),
        };

        let released_price = match &events[0] {
            OrderEventAny::Released(event) => Some(event.released_price),
            _ => None,
        };

        assert_eq!(released_price, expected_price.filter(|_| should_release));
    }

    #[rstest]
    #[case(TriggerType::Default, "5099.00", "5101.00", None)]
    #[case(TriggerType::BidAsk, "5099.00", "5101.00", None)]
    #[case(TriggerType::LastPrice, "5098.00", "5098.00", Some("5098.00"))]
    fn test_create_matching_core_seeds_cached_prices(
        instrument: CryptoPerpetual,
        #[case] emulation_trigger: TriggerType,
        #[case] expected_bid: &str,
        #[case] expected_ask: &str,
        #[case] expected_last: Option<&str>,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, "5099.00", "5101.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5098.00"))
            .unwrap();

        let core = emulator.borrow_mut().create_matching_core(
            instrument.id(),
            instrument.price_increment,
            emulation_trigger,
        );

        let emulator = emulator.borrow();
        let stored_core = emulator.get_matching_core(&instrument.id()).unwrap();
        assert_eq!(core.bid, Some(Price::from(expected_bid)));
        assert_eq!(core.ask, Some(Price::from(expected_ask)));
        assert_eq!(core.last, expected_last.map(Price::from));
        assert_eq!(stored_core.bid, core.bid);
        assert_eq!(stored_core.ask, core.ask);
        assert_eq!(stored_core.last, core.last);
    }

    #[rstest]
    fn test_submit_order_seeds_cached_trigger_instrument(
        instrument: CryptoPerpetual,
        #[values(TriggerType::BidAsk, TriggerType::LastPrice)] emulation_trigger: TriggerType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let _risk_events =
            register_risk_event_handler("RiskEngine.process.cached_trigger_instrument");
        let trigger_instrument = btcusd_bybit();
        add_instrument_to_cache(&cache, &instrument);
        add_instrument_to_cache(&cache, &trigger_instrument);
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, "5099.00", "5101.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5101.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&trigger_instrument, "5098.00", "5099.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&trigger_instrument, "5099.00"))
            .unwrap();
        let order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .trigger_instrument_id(trigger_instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(emulation_trigger)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);

        let emulator = emulator.borrow();
        let core = emulator
            .get_matching_core(&trigger_instrument.id())
            .unwrap();

        let expected_bid = if emulation_trigger == TriggerType::LastPrice {
            "5099.00"
        } else {
            "5098.00"
        };

        assert_eq!(core.bid, Some(Price::from(expected_bid)));
        assert_eq!(core.ask, Some(Price::from("5099.00")));
        assert_eq!(
            core.last,
            (emulation_trigger == TriggerType::LastPrice).then(|| Price::from("5099.00"))
        );
        assert_eq!(
            cache.borrow().order(&client_order_id).unwrap().status(),
            OrderStatus::Emulated
        );
        assert!(emulator.get_matching_core(&instrument.id()).is_none());
    }

    #[rstest]
    fn test_quote_emulated_trailing_stop_uses_latest_cached_trade(
        instrument: CryptoPerpetual,
        #[values(TriggerType::Default, TriggerType::BidAsk)] emulation_trigger: TriggerType,
        #[values(TriggerType::LastPrice, TriggerType::LastOrBidAsk)] trigger_type: TriggerType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let _risk_events = register_risk_event_handler("RiskEngine.process.cached_trailing_trade");
        add_instrument_to_cache(&cache, &instrument);
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, "5000.00", "5001.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5000.00"))
            .unwrap();
        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
            .instrument_id(instrument.id())
            .side(OrderSide::Sell)
            .quantity(Quantity::from(3))
            .trigger_type(trigger_type)
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(emulation_trigger)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);

        assert_eq!(
            cache
                .borrow()
                .order(&client_order_id)
                .unwrap()
                .trigger_price(),
            Some(Price::from("4990.00"))
        );

        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5050.00"))
            .unwrap();
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5010.00", "5011.00"));

        let cache = cache.borrow();
        let order = cache.order(&client_order_id).unwrap();
        assert_eq!(order.trigger_price(), Some(Price::from("5040.00")));
        assert_eq!(order.quantity(), Quantity::from(3));
        assert_eq!(order.status(), OrderStatus::Emulated);
    }

    #[rstest]
    fn test_submit_order_preserves_existing_matching_core(
        instrument: CryptoPerpetual,
        #[values(TriggerType::BidAsk, TriggerType::LastPrice)] emulation_trigger: TriggerType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let _risk_events = register_risk_event_handler("RiskEngine.process.cached_existing_core");
        add_instrument_to_cache(&cache, &instrument);
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, "5098.00", "5099.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5099.00"))
            .unwrap();

        let first = create_stop_market_order(&instrument, emulation_trigger);
        let first_id = first.client_order_id();
        let first_command = create_submit_order(&instrument, &first);
        cache
            .borrow_mut()
            .add_order(first, None, None, false)
            .unwrap();
        emulator.borrow_mut().handle_submit_order(&first_command);
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, "5101.00", "5102.00"))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, "5102.00"))
            .unwrap();
        let second = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-CACHED-SECOND"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(emulation_trigger)
            .build();
        let second_id = second.client_order_id();
        let second_command = create_submit_order(&instrument, &second);
        cache
            .borrow_mut()
            .add_order(second, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&second_command);

        let emulator = emulator.borrow();
        let core = emulator.get_matching_core(&instrument.id()).unwrap();
        assert_eq!(core.ask, Some(Price::from("5099.00")));
        assert_eq!(
            core.last,
            (emulation_trigger == TriggerType::LastPrice).then(|| Price::from("5099.00"))
        );
        assert!(core.order_exists(first_id));
        assert!(core.order_exists(second_id));
        assert_eq!(
            cache.borrow().order(&first_id).unwrap().status(),
            OrderStatus::Emulated
        );
        assert_eq!(
            cache.borrow().order(&second_id).unwrap().status(),
            OrderStatus::Emulated
        );
    }

    #[rstest]
    #[case(OrderType::StopMarket, ("5099.00", "5101.00"), "5099.00", "5101.00")]
    #[case(OrderType::Limit, ("5098.00", "5099.00"), "5101.00", "5099.00")]
    fn test_modify_last_price_order_ignores_cached_quote_before_first_trade(
        instrument: CryptoPerpetual,
        #[case] order_type: OrderType,
        #[case] quote: (&str, &str),
        #[case] trade: &str,
        #[case] next_trade: &str,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.modify_cached_quote");
        add_instrument_to_cache(&cache, &instrument);
        cache
            .borrow_mut()
            .add_quote(create_quote_tick(&instrument, quote.0, quote.1))
            .unwrap();
        cache
            .borrow_mut()
            .add_trade(create_trade_tick(&instrument, trade))
            .unwrap();
        let order = OrderTestBuilder::new(order_type)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5150.00"))
            .price(Price::from("5050.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::LastPrice)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator.borrow_mut().handle_submit_order(&command);

        let modify = ModifyOrder::new(
            order.trader_id(),
            None,
            order.strategy_id(),
            instrument.id(),
            client_order_id,
            None,
            None,
            (order_type == OrderType::Limit).then(|| Price::from("5100.00")),
            (order_type == OrderType::StopMarket).then(|| Price::from("5100.00")),
            UUID4::new(),
            0.into(),
            None,
            None,
        );

        emulator.borrow_mut().handle_modify_order(&modify);

        {
            let cache = cache.borrow();
            let cached_order = cache.order(&client_order_id).unwrap();
            let core = emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .unwrap();
            assert_eq!(core.bid, Some(Price::from(trade)));
            assert_eq!(core.ask, Some(Price::from(trade)));
            assert_eq!(core.last, Some(Price::from(trade)));
            assert_eq!(cached_order.status(), OrderStatus::Emulated);
            assert_eq!(
                if order_type == OrderType::Limit {
                    cached_order.price()
                } else {
                    cached_order.trigger_price()
                },
                Some(Price::from("5100.00"))
            );
            assert!(matches!(risk_events.get_messages().as_slice(),
                [OrderEventAny::Emulated(event)] if event.client_order_id == client_order_id
            ));
        }

        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, next_trade));

        assert_eq!(
            cache.borrow().order(&client_order_id).unwrap().status(),
            OrderStatus::Released
        );
        assert!(matches!(risk_events.get_messages().as_slice(),
            [OrderEventAny::Emulated(_), OrderEventAny::Released(event)]
                if event.client_order_id == client_order_id
                    && event.released_price == Price::from(next_trade)
        ));
    }

    #[rstest]
    fn test_submit_limit_order_releases_when_immediately_fillable(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events =
            register_risk_event_handler("RiskEngine.process.submit_immediately_fillable");
        add_instrument_to_cache(&cache, &instrument);
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5099.00"));
        core.set_ask_raw(Price::from("5101.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);
        let order = OrderTestBuilder::new(OrderType::Limit)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5101.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
        let risk_events = risk_events.get_messages();
        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(cached_order.order_type(), OrderType::Market);
        assert!(!matching_core.order_exists(client_order_id));
        assert!(
            !emulator
                .get_submit_order_commands()
                .contains_key(&client_order_id)
        );
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(
            &risk_events[0],
            OrderEventAny::Released(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_modify_order_reindexes_matching_state(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator.borrow_mut().handle_submit_order(&command);
        let exec_events =
            register_exec_event_handler(cache.clone(), "ExecEngine.process.modify_reindex");
        let portfolio_events = register_portfolio_event_handler("Portfolio.modify_reindex");
        let new_trigger = Price::from("5200.00");
        let modify = ModifyOrder::new(
            order.trader_id(),
            None,
            order.strategy_id(),
            instrument.id(),
            client_order_id,
            None,
            None,
            None,
            Some(new_trigger),
            UUID4::new(),
            0.into(),
            None,
            None,
        );

        emulator.borrow_mut().handle_modify_order(&modify);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
        let match_info = matching_core.get_order(client_order_id).unwrap();
        let portfolio_events = portfolio_events.get_messages();
        assert_eq!(cached_order.trigger_price(), Some(new_trigger));
        assert_eq!(match_info.trigger_price, Some(new_trigger));
        assert_eq!(matching_core.get_orders().len(), 1);
        assert!(exec_events.get_messages().is_empty());
        assert_eq!(portfolio_events.len(), 1);
        assert!(matches!(
            &portfolio_events[0],
            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_modify_order_releases_when_updated_trigger_matches(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events =
            register_risk_event_handler("RiskEngine.process.modify_immediately_triggered");
        add_instrument_to_cache(&cache, &instrument);
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5099.00"));
        core.set_ask_raw(Price::from("5101.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);
        let order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5200.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator.borrow_mut().handle_submit_order(&command);
        risk_events.clear();
        let exec_events = register_exec_event_handler(
            cache.clone(),
            "ExecEngine.process.modify_immediately_triggered",
        );
        let portfolio_events =
            register_portfolio_event_handler("Portfolio.modify_immediately_triggered");

        let modify = ModifyOrder::new(
            order.trader_id(),
            None,
            order.strategy_id(),
            instrument.id(),
            client_order_id,
            None,
            None,
            None,
            Some(Price::from("5100.00")),
            UUID4::new(),
            0.into(),
            None,
            None,
        );

        emulator.borrow_mut().handle_modify_order(&modify);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let emulator = emulator.borrow();
        let matching_core = emulator.get_matching_core(&instrument.id()).unwrap();
        let portfolio_events = portfolio_events.get_messages();
        let risk_events = risk_events.get_messages();
        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(cached_order.order_type(), OrderType::Market);
        assert!(!matching_core.order_exists(client_order_id));
        assert!(
            !emulator
                .get_submit_order_commands()
                .contains_key(&client_order_id)
        );
        assert!(exec_events.get_messages().is_empty());
        assert_eq!(portfolio_events.len(), 1);
        assert!(matches!(
            &portfolio_events[0],
            OrderEventAny::Updated(event) if event.client_order_id == client_order_id
        ));
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(
            &risk_events[0],
            OrderEventAny::Released(event) if event.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_update_order_applies_updated_event_to_cache(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.updated");
        let mut order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .update_order(&mut order, Quantity::from(2));
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();

        assert_eq!(order.quantity(), Quantity::from(2));
        assert_eq!(cached_order.quantity(), Quantity::from(2));
        assert_eq!(cached_order.status(), OrderStatus::Initialized);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Updated(_)));
    }

    #[rstest]
    fn test_fill_market_order_applies_released_event_to_cache(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.released");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
        risk_events.clear();
        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
        {
            let mut emulator = emulator.borrow_mut();
            emulator
                .matching_cores
                .get_mut(&instrument.id())
                .unwrap()
                .set_ask_raw(Price::from("5100.00"));
            emulator.fill_market_order(client_order_id);
        }

        msgbus::unsubscribe_order_events(
            format!("events.order.{strategy_id}").into(),
            &order_handler,
        );
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();
        let order_events = order_events.borrow();

        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
        assert_eq!(order_events.len(), 2);
        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
    }

    #[rstest]
    fn test_fill_limit_order_publishes_transformed_initialized_before_released(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.limit_released");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
        risk_events.clear();
        let (order_handler, order_events) = subscribe_order_topic(strategy_id);
        {
            let mut emulator = emulator.borrow_mut();
            emulator
                .matching_cores
                .get_mut(&instrument.id())
                .unwrap()
                .set_ask_raw(Price::from("5100.00"));
            emulator.fill_limit_order(client_order_id);
        }

        msgbus::unsubscribe_order_events(
            format!("events.order.{strategy_id}").into(),
            &order_handler,
        );
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let risk_events = risk_events.get_messages();
        let order_events = order_events.borrow();

        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(risk_events.len(), 1);
        assert!(matches!(risk_events[0], OrderEventAny::Released(_)));
        assert_eq!(order_events.len(), 2);
        assert!(matches!(order_events[0], OrderEventAny::Initialized(_)));
        assert!(matches!(order_events[1], OrderEventAny::Released(_)));
    }

    fn register_risk_command_handler(id: &str) -> TypedIntoMessageSavingHandler<TradingCommand> {
        let (handler, saving_handler) =
            get_typed_into_message_saving_handler::<TradingCommand>(Some(Ustr::from(id)));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        saving_handler
    }

    fn submit_to_emulator(
        cache: &Rc<RefCell<Cache>>,
        emulator: &Rc<RefCell<OrderEmulator>>,
        instrument: &CryptoPerpetual,
        order: &OrderAny,
    ) {
        let command = create_submit_order(instrument, order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
    }

    #[rstest]
    fn test_release_without_market_data_retains_command_and_retries(
        instrument: CryptoPerpetual,
        #[values(OrderType::StopLimit, OrderType::StopMarket, OrderType::Limit)]
        order_type: OrderType,
        #[values(OrderSide::Buy, OrderSide::Sell)] side: OrderSide,
        #[values(false, true)] use_algorithm: bool,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.release_retry");
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.release_retry");
        let endpoint = Ustr::from("ALG-RETRY.execute");
        let (handler, algo_commands) = get_any_saving_handler::<TradingCommand>(Some(endpoint));
        msgbus::register_any(endpoint.into(), handler);
        add_instrument_to_cache(&cache, &instrument);
        let mut builder = OrderTestBuilder::new(order_type);
        builder
            .instrument_id(instrument.id())
            .side(side)
            .price(match side {
                OrderSide::Buy => Price::from("5102.00"),
                OrderSide::Sell => Price::from("5098.00"),
            })
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .emulation_trigger(TriggerType::BidAsk);

        if use_algorithm {
            builder
                .exec_algorithm_id(ExecAlgorithmId::from("ALG-RETRY"))
                .exec_spawn_id(ClientOrderId::from("O-SPAWN-RETRY"));
        }

        let order = builder.build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator.borrow_mut().handle_submit_order(&command);
        let held_order = cache.borrow().order(&client_order_id).unwrap().clone();
        let held_entry = *emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap()
            .get_order(client_order_id)
            .unwrap();
        risk_events.clear();
        let (order_handler, order_events) = subscribe_order_topic(order.strategy_id());

        {
            let mut emulator = emulator.borrow_mut();
            let core = emulator.matching_cores.get_mut(&instrument.id()).unwrap();

            // The opposite quote side cannot supply the release price
            match side {
                OrderSide::Buy => core.set_bid_raw(Price::from("5099.00")),
                OrderSide::Sell => core.set_ask_raw(Price::from("5101.00")),
            }

            match order_type {
                OrderType::StopMarket => emulator.fill_market_order(client_order_id),
                _ => emulator.fill_limit_order(client_order_id),
            }
        }

        {
            let emulator = emulator.borrow();
            let cache = cache.borrow();
            let cached_order = cache.order(&client_order_id).unwrap();
            assert_eq!(cached_order.status(), OrderStatus::Emulated);
            assert_eq!(cached_order.order_type(), order_type);
            assert_eq!(cached_order.events(), held_order.events());
            assert_eq!(
                emulator.get_submit_order_commands().get(&client_order_id),
                Some(&command),
            );
            assert_eq!(
                emulator
                    .get_matching_core(&instrument.id())
                    .unwrap()
                    .get_order(client_order_id),
                Some(&held_entry),
            );
            assert!(risk_events.get_messages().is_empty());
            assert!(order_events.borrow().is_empty());
            assert!(risk_commands.get_messages().is_empty());
            assert!(algo_commands.get_messages().is_empty());
        }

        let quote = create_quote_tick(&instrument, "5099.00", "5101.00");
        emulator.borrow_mut().on_quote_tick(quote);
        emulator.borrow_mut().on_quote_tick(quote);
        msgbus::unsubscribe_order_events(
            format!("events.order.{}", order.strategy_id()).into(),
            &order_handler,
        );
        let risk_commands = risk_commands.get_messages();
        let algo_commands = algo_commands.get_messages();
        let (commands, other_commands) = if use_algorithm {
            (&algo_commands, &risk_commands)
        } else {
            (&risk_commands, &algo_commands)
        };

        let [TradingCommand::SubmitOrder(released_command)] = commands.as_slice() else {
            panic!("Expected one released SubmitOrder, was {commands:?}");
        };

        let emulator = emulator.borrow();
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let events = cached_order.events();
        let order_events = order_events.borrow();
        let mut expected_command = command;
        expected_command.order_init = OrderInitialized::from(&*cached_order);

        let released_price = match side {
            OrderSide::Buy => Price::from("5101.00"),
            OrderSide::Sell => Price::from("5099.00"),
        };

        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(
            cached_order.order_type(),
            match order_type {
                OrderType::StopLimit => OrderType::Limit,
                _ => OrderType::Market,
            },
        );
        assert_eq!(cached_order.emulation_trigger(), None);
        assert_eq!(&events[..held_order.event_count()], held_order.events());
        assert_eq!(events.len(), held_order.event_count() + 2);
        assert_eq!(
            order_events.as_slice(),
            [
                OrderEventAny::Initialized(released_command.order_init.clone()),
                cached_order.last_event().clone(),
            ],
        );
        assert!(matches!(
            cached_order.last_event(),
            OrderEventAny::Released(event) if event.released_price == released_price
        ));
        assert_eq!(
            risk_events.get_messages(),
            [cached_order.last_event().clone()]
        );
        assert_eq!(released_command, &expected_command);
        assert!(other_commands.is_empty());
        assert!(emulator.get_submit_order_commands().is_empty());
        assert_eq!(
            emulator
                .get_matching_core(&instrument.id())
                .unwrap()
                .get_order(client_order_id),
            None,
        );
    }

    #[rstest]
    fn test_release_updates_cache_before_publication_and_defers_reentrant_commands(
        instrument: CryptoPerpetual,
        #[values(OrderType::StopLimit, OrderType::StopMarket)] order_type: OrderType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_commands =
            register_risk_command_handler("RiskEngine.queue_execute.release_events");
        add_instrument_to_cache(&cache, &instrument);
        let order = OrderTestBuilder::new(order_type)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5102.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let sibling = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-REENTRANT-CANCEL"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5200.00"))
            .quantity(Quantity::from(2))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        submit_to_emulator(&cache, &emulator, &instrument, &order);
        submit_to_emulator(&cache, &emulator, &instrument, &sibling);
        let client_order_id = order.client_order_id();

        let cancel = CancelOrder::new(
            sibling.trader_id(),
            None,
            sibling.strategy_id(),
            sibling.instrument_id(),
            sibling.client_order_id(),
            None,
            UUID4::new(),
            0.into(),
            None,
            None,
        );
        let observed = Rc::new(RefCell::new(Vec::new()));

        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            TypedIntoHandler::from({
                let cache = cache.clone();
                let observed = observed.clone();
                let exec_handler = risk_commands.handler();
                move |command: TradingCommand| {
                    let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
                    observed.borrow_mut().push((
                        "exec",
                        cached_order.status(),
                        cached_order.order_type(),
                        cached_order.last_event().clone(),
                    ));
                    exec_handler.handle(command);
                }
            }),
        );

        let order_handler = TypedHandler::from({
            let cache = cache.clone();
            let observed = observed.clone();
            move |event: &OrderEventAny| {
                if event.client_order_id() != client_order_id {
                    return;
                }

                let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
                observed.borrow_mut().push((
                    "order",
                    cached_order.status(),
                    cached_order.order_type(),
                    cached_order.last_event().clone(),
                ));

                if matches!(event, OrderEventAny::Initialized(_)) {
                    msgbus::send_trading_command(
                        MessagingSwitchboard::order_emulator_execute(),
                        TradingCommand::CancelOrder(cancel.clone()),
                    );
                }
            }
        });

        msgbus::subscribe_order_events(
            format!("events.order.{}", order.strategy_id()).into(),
            order_handler.clone(),
            None,
        );
        msgbus::register_order_event_endpoint(
            MessagingSwitchboard::risk_engine_process(),
            TypedIntoHandler::from({
                let cache = cache.clone();
                let observed = observed.clone();
                move |event: OrderEventAny| {
                    if event.client_order_id() != client_order_id {
                        return;
                    }

                    let cached_order = cache.borrow().order_owned(&client_order_id).unwrap();
                    observed.borrow_mut().push((
                        "risk",
                        cached_order.status(),
                        cached_order.order_type(),
                        cached_order.last_event().clone(),
                    ));
                }
            }),
        );

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5099.00", "5101.00"));
        msgbus::unsubscribe_order_events(
            format!("events.order.{}", order.strategy_id()).into(),
            &order_handler,
        );
        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let events = cached_order.events();

        let released_type = match order_type {
            OrderType::StopLimit => OrderType::Limit,
            _ => OrderType::Market,
        };

        assert_eq!(
            observed.borrow().as_slice(),
            [
                (
                    "order",
                    OrderStatus::Initialized,
                    released_type,
                    events[events.len() - 2].clone(),
                ),
                (
                    "risk",
                    OrderStatus::Released,
                    released_type,
                    cached_order.last_event().clone(),
                ),
                (
                    "order",
                    OrderStatus::Released,
                    released_type,
                    cached_order.last_event().clone(),
                ),
                (
                    "exec",
                    OrderStatus::Released,
                    released_type,
                    cached_order.last_event().clone(),
                ),
            ],
        );
        assert_eq!(
            cache.order(&sibling.client_order_id()).unwrap().status(),
            OrderStatus::Canceled,
        );
        assert!(matches!(
            risk_commands.get_messages().as_slice(),
            [TradingCommand::SubmitOrder(command)] if command.client_order_id == client_order_id
        ));
        assert!(emulator.borrow().get_submit_order_commands().is_empty());
        assert!(
            emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .unwrap()
                .get_orders()
                .is_empty()
        );
    }

    #[rstest]
    fn test_release_stops_when_cached_order_rejects_released_event(
        instrument: CryptoPerpetual,
        #[values(OrderType::StopLimit, OrderType::StopMarket)] order_type: OrderType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.release_rejected");
        let risk_commands =
            register_risk_command_handler("RiskEngine.queue_execute.release_rejected");
        add_instrument_to_cache(&cache, &instrument);
        let order = OrderTestBuilder::new(order_type)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5102.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        submit_to_emulator(&cache, &emulator, &instrument, &order);
        let client_order_id = order.client_order_id();
        let held_order = cache.borrow().order_owned(&client_order_id).unwrap();
        let canceled =
            TestOrderEventStubs::canceled(&order, AccountId::from("ACCOUNT-RELEASE"), None);
        let observed = Rc::new(RefCell::new(Vec::new()));

        let order_handler = TypedHandler::from({
            let cache = cache.clone();
            let canceled = canceled.clone();
            let observed = observed.clone();
            move |event: &OrderEventAny| {
                observed.borrow_mut().push(event.clone());

                if matches!(event, OrderEventAny::Initialized(_)) {
                    cache.borrow_mut().update_order(&canceled).unwrap();
                }
            }
        });

        msgbus::subscribe_order_events(
            format!("events.order.{}", order.strategy_id()).into(),
            order_handler.clone(),
            None,
        );
        risk_events.clear();

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5099.00", "5101.00"));
        msgbus::unsubscribe_order_events(
            format!("events.order.{}", order.strategy_id()).into(),
            &order_handler,
        );

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let events = cached_order.events();
        let mut expected_events = held_order.events();
        expected_events.push(events[held_order.events().len()]);
        expected_events.push(&canceled);

        assert_eq!(cached_order.status(), OrderStatus::Canceled);
        assert_eq!(events, expected_events);
        assert!(matches!(
            events[events.len() - 2],
            OrderEventAny::Initialized(_)
        ));
        assert_eq!(
            observed.borrow().as_slice(),
            [events[events.len() - 2].clone()]
        );
        assert!(risk_events.get_messages().is_empty());
        assert!(risk_commands.get_messages().is_empty());
        assert!(emulator.borrow().get_submit_order_commands().is_empty());
        assert!(
            emulator
                .borrow()
                .get_matching_core(&instrument.id())
                .unwrap()
                .get_orders()
                .is_empty()
        );
    }

    #[rstest]
    fn test_fill_limit_order_without_command_retains_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.no_command");
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        submit_to_emulator(&cache, &emulator, &instrument, &order);
        let held_order = cache.borrow().order(&client_order_id).unwrap().clone();
        let mut emulator = emulator.borrow_mut();
        let command = emulator
            .manager
            .pop_submit_order_command(client_order_id)
            .unwrap();
        emulator
            .matching_cores
            .get_mut(&instrument.id())
            .unwrap()
            .set_ask_raw(Price::from("5101.00"));

        emulator.fill_limit_order(client_order_id);

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        assert_eq!(command.client_order_id, client_order_id);
        assert_eq!(cached_order.status(), OrderStatus::Emulated);
        assert_eq!(cached_order.order_type(), OrderType::StopLimit);
        assert_eq!(cached_order.events(), held_order.events());
        assert!(
            emulator
                .get_matching_core(&instrument.id())
                .unwrap()
                .order_exists(client_order_id)
        );
        assert!(risk_commands.get_messages().is_empty());
    }

    #[rstest]
    #[should_panic(expected = "invalid operation `fill_market_order` with no command")]
    fn test_fill_market_order_without_command_panics(
        instrument: CryptoPerpetual,
        #[values(OrderType::StopMarket, OrderType::Limit)] order_type: OrderType,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = OrderTestBuilder::new(order_type)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5102.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        submit_to_emulator(&cache, &emulator, &instrument, &order);
        let mut emulator = emulator.borrow_mut();
        emulator
            .manager
            .pop_submit_order_command(client_order_id)
            .unwrap();
        emulator
            .matching_cores
            .get_mut(&instrument.id())
            .unwrap()
            .set_ask_raw(Price::from("5101.00"));

        match order_type {
            OrderType::Limit => emulator.fill_limit_order(client_order_id),
            _ => emulator.fill_market_order(client_order_id),
        }
    }

    #[rstest]
    fn test_release_gtd_stop_market_as_gtc_market_order(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.gtd");
        add_instrument_to_cache(&cache, &instrument);
        let order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(1))
            .time_in_force(TimeInForce::Gtd)
            .expire_time(UnixNanos::from(1_000_000_000_000))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        submit_to_emulator(&cache, &emulator, &instrument, &order);

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        assert_eq!(cached_order.order_type(), OrderType::Market);
        assert_eq!(cached_order.time_in_force(), TimeInForce::Gtc);
        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert!(matches!(
            risk_commands.get_messages().as_slice(),
            [TradingCommand::SubmitOrder(command)] if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_last_price_emulated_stop_triggers_on_trades_not_quotes(
        instrument: CryptoPerpetual,
        #[values(false, true)] modified: bool,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let _data_commands = register_data_command_handler("DataEngine.queue_execute.last_price");
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.last_price");
        add_instrument_to_cache(&cache, &instrument);

        // A quote-triggered order on the same instrument subscribes the emulator to quotes
        let quote_order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-QUOTE"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5200.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .build();

        // The modified case re-indexes the order through the modify path
        let initial_trigger = if modified { "5150.00" } else { "5100.00" };
        let last_order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-LAST"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from(initial_trigger))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::LastPrice)
            .build();
        submit_to_emulator(&cache, &emulator, &instrument, &quote_order);
        submit_to_emulator(&cache, &emulator, &instrument, &last_order);

        if modified {
            let modify = ModifyOrder::new(
                last_order.trader_id(),
                None,
                last_order.strategy_id(),
                instrument.id(),
                last_order.client_order_id(),
                None,
                None,
                None,
                Some(Price::from("5100.00")),
                UUID4::new(),
                0.into(),
                None,
                None,
            );
            emulator.borrow_mut().handle_modify_order(&modify);
        }

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));
        let commands_after_quote = risk_commands.get_messages().len();
        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, "5101.00"));

        assert_eq!(commands_after_quote, 0);
        assert!(matches!(
            risk_commands.get_messages().as_slice(),
            [TradingCommand::SubmitOrder(command)]
                if command.client_order_id == last_order.client_order_id()
        ));
    }

    #[rstest]
    fn test_last_price_emulated_trailing_stop_triggers_on_trades_not_quotes(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let _data_commands = register_data_command_handler("DataEngine.queue_execute.trailing");
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.trailing");
        add_instrument_to_cache(&cache, &instrument);
        let quote_order = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-QUOTE"))
            .side(OrderSide::Buy)
            .trigger_price(Price::from("5200.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let trailing_order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_type(TriggerType::LastPrice)
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::LastPrice)
            .build();
        submit_to_emulator(&cache, &emulator, &instrument, &quote_order);
        submit_to_emulator(&cache, &emulator, &instrument, &trailing_order);

        // The first trade activates the stop and trails its trigger to 4990
        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, "5000.00"));
        let trigger_price = cache
            .borrow()
            .order(&trailing_order.client_order_id())
            .unwrap()
            .trigger_price();
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "4980.00", "4981.00"));
        let commands_after_quote = risk_commands.get_messages().len();
        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, "4985.00"));

        assert_eq!(trigger_price, Some(Price::from("4990.00")));
        assert_eq!(commands_after_quote, 0);
        assert!(matches!(
            risk_commands.get_messages().as_slice(),
            [TradingCommand::SubmitOrder(command)]
                if command.client_order_id == trailing_order.client_order_id()
        ));
    }

    #[rstest]
    fn test_failed_limit_release_keeps_order_emulated(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_commands = register_risk_command_handler("RiskEngine.queue_execute.failed");
        add_instrument_to_cache(&cache, &instrument);
        let mut order = OrderTestBuilder::new(OrderType::StopLimit)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5100.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(10))
            .display_qty(Quantity::from(5))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        submit_to_emulator(&cache, &emulator, &instrument, &order);

        // A display quantity above the reduced quantity cannot form a limit order on release
        emulator
            .borrow_mut()
            .update_order(&mut order, Quantity::from(3));

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5100.00", "5101.00"));

        let emulator = emulator.borrow();
        let cache = cache.borrow();
        assert_eq!(
            cache.order(&client_order_id).unwrap().status(),
            OrderStatus::Emulated
        );
        assert!(
            emulator
                .get_submit_order_commands()
                .contains_key(&client_order_id)
        );
        assert!(
            emulator
                .get_matching_core(&instrument.id())
                .unwrap()
                .order_exists(client_order_id)
        );
        assert!(risk_commands.get_messages().is_empty());
    }

    #[rstest]
    #[case(
        OrderType::StopMarket,
        OrderType::Market,
        TimeInForce::Gtc,
        TimeInForce::Gtc
    )]
    #[case(
        OrderType::StopMarket,
        OrderType::Market,
        TimeInForce::Gtd,
        TimeInForce::Gtc
    )]
    #[case(
        OrderType::StopLimit,
        OrderType::Limit,
        TimeInForce::Gtc,
        TimeInForce::Gtc
    )]
    #[case(
        OrderType::StopLimit,
        OrderType::Limit,
        TimeInForce::Gtd,
        TimeInForce::Gtd
    )]
    fn test_release_order_sends_transformed_command(
        instrument: CryptoPerpetual,
        #[case] order_type: OrderType,
        #[case] released_type: OrderType,
        #[case] time_in_force: TimeInForce,
        #[case] released_time_in_force: TimeInForce,
        #[values(false, true)] use_algorithm: bool,
    ) {
        let (clock, cache, emulator) = create_emulator();
        let release_time = UnixNanos::from(29);
        clock
            .borrow_mut()
            .as_any_mut()
            .downcast_mut::<VirtualClock>()
            .unwrap()
            .advance_time(release_time, true);
        let _risk_events = register_risk_event_handler("RiskEngine.process.release_command");
        let risk_commands =
            register_risk_command_handler("RiskEngine.queue_execute.release_command");
        let exec_algorithm_id = ExecAlgorithmId::from("ALG-RELEASE");
        let endpoint = Ustr::from("ALG-RELEASE.execute");
        let (handler, algo_commands) = get_any_saving_handler::<TradingCommand>(Some(endpoint));
        msgbus::register_any(endpoint.into(), handler);
        add_instrument_to_cache(&cache, &instrument);
        let trigger_instrument = btcusd_bybit();
        add_instrument_to_cache(&cache, &trigger_instrument);
        let mut builder = OrderTestBuilder::new(order_type);
        builder
            .instrument_id(instrument.id())
            .trigger_instrument_id(trigger_instrument.id())
            .side(OrderSide::Buy)
            .price(Price::from("5099.00"))
            .trigger_price(Price::from("5100.00"))
            .quantity(Quantity::from(3))
            .ts_init(UnixNanos::from(11))
            .time_in_force(time_in_force)
            .reduce_only(true)
            .quote_quantity(true)
            .tags(vec![Ustr::from("release-tag")])
            .emulation_trigger(TriggerType::BidAsk);

        if order_type == OrderType::StopLimit {
            builder.post_only(true).display_qty(Quantity::from(1));
        }

        if time_in_force == TimeInForce::Gtd {
            builder.expire_time(UnixNanos::from(1_000_000_000_000));
        }

        if use_algorithm {
            builder
                .exec_algorithm_id(exec_algorithm_id)
                .exec_algorithm_params(
                    [(Ustr::from("release-algo-param"), Ustr::from("preserved"))].into(),
                )
                .exec_spawn_id(ClientOrderId::from("O-SPAWN-RELEASE"));
        }

        let mut order = builder.build();
        let client_order_id = order.client_order_id();
        let mut command = create_submit_order(&instrument, &order);
        command.client_id = Some(ClientId::from("EXECUTION-RELEASE"));
        command.position_id = Some(PositionId::from("P-RELEASE"));
        command.exec_algorithm_id = order.exec_algorithm_id();
        let params = Params::new([("release-param".to_string(), "preserved".into())].into());
        command.params = Some(params);
        command.ts_init = UnixNanos::from(123);
        command.correlation_id = Some(UUID4::new());
        command.causation_id = Some(UUID4::new());
        let original_command = command.clone();
        cache
            .borrow_mut()
            .add_order(order.clone(), command.position_id, command.client_id, false)
            .unwrap();

        emulator.borrow_mut().handle_submit_order(&command);
        emulator
            .borrow_mut()
            .update_order(&mut order, Quantity::from(2));
        emulator.borrow_mut().on_quote_tick(create_quote_tick(
            &trigger_instrument,
            "5100.00",
            "5101.00",
        ));

        let risk_commands = risk_commands.get_messages();
        let algo_commands = algo_commands.get_messages();
        let (commands, other_commands) = if use_algorithm {
            (&algo_commands, &risk_commands)
        } else {
            (&risk_commands, &algo_commands)
        };

        let [TradingCommand::SubmitOrder(released_command)] = commands.as_slice() else {
            panic!("expected one released SubmitOrder, was {commands:?}");
        };

        let cache = cache.borrow();
        let cached_order = cache.order(&client_order_id).unwrap();
        let events = cached_order.events();
        let mut expected_command = original_command;
        let expected_init = &mut expected_command.order_init;
        expected_init.order_type = released_type;
        expected_init.quantity = Quantity::from(2);
        expected_init.time_in_force = released_time_in_force;
        expected_init.trigger_price = None;
        expected_init.trigger_type = None;
        expected_init.emulation_trigger = None;
        expected_init.event_id = released_command.order_init.event_id;
        expected_init.ts_event = release_time;
        expected_init.ts_init = release_time;
        if released_type == OrderType::Market {
            expected_init.price = None;
            expected_init.expire_time = None;
            expected_init.trigger_instrument_id = None;
        }

        assert_eq!(cached_order.status(), OrderStatus::Released);
        assert_eq!(
            events[events.len() - 2],
            &OrderEventAny::Initialized(released_command.order_init.clone()),
        );
        assert_eq!(released_command, &expected_command);
        assert_eq!(
            OrderInitialized::from(&*cached_order),
            expected_command.order_init
        );
        assert!(other_commands.is_empty());
    }

    #[rstest]
    fn test_quote_tick_updates_matching_core_prices(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        let quote = create_quote_tick(&instrument, "5060.00", "5070.00");
        emulator.borrow_mut().on_quote_tick(quote);

        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert_eq!(core.bid, Some(Price::from("5060.00")));
        assert_eq!(core.ask, Some(Price::from("5070.00")));
    }

    #[rstest]
    fn test_trade_tick_updates_matching_core_last_price(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::LastPrice);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        let trade = create_trade_tick(&instrument, "5065.00");
        emulator.borrow_mut().on_trade_tick(trade);

        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert_eq!(core.last, Some(Price::from("5065.00")));
    }

    #[rstest]
    fn test_cancel_order_removes_from_matching_core(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        emulator.borrow_mut().cancel_order(&order);

        let core = emulator
            .borrow()
            .get_matching_core(&instrument.id())
            .unwrap();
        assert!(core.get_orders().is_empty());
    }

    #[rstest]
    fn test_cancel_order_removes_cached_command(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        emulator.borrow_mut().cancel_order(&order);

        let commands = emulator.borrow().get_submit_order_commands();
        assert!(!commands.contains_key(&client_order_id));
    }

    #[rstest]
    fn test_trailing_stop_waits_for_activation_price(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let risk_events = register_risk_event_handler("RiskEngine.process.trailing_activation");
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-1"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        {
            let cache = cache.borrow();
            let cached = cache.order(&client_order_id).unwrap();
            let is_activated = match &*cached {
                OrderAny::TrailingStopMarket(o) => o.is_activated,
                _ => panic!("expected trailing stop market order"),
            };
            assert!(
                !is_activated,
                "must not activate before the activation price trades"
            );
            assert!(cached.trigger_price().is_none());
        }
        assert!(
            !risk_events
                .get_messages()
                .iter()
                .any(|e| matches!(e, OrderEventAny::Updated(_))),
            "no trailing update before activation"
        );

        // Market rallies through the 5050 activation price: activates and starts trailing
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));

        {
            let cache = cache.borrow();
            let cached = cache.order(&client_order_id).unwrap();
            let is_activated = match &*cached {
                OrderAny::TrailingStopMarket(o) => o.is_activated,
                _ => panic!("expected trailing stop market order"),
            };
            assert!(is_activated);
            assert_eq!(cached.trigger_price(), Some(Price::from("5045.00")));
        }

        assert!(risk_commands.get_messages().is_empty());

        // Market falls through the trailed trigger: order releases
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));

        let commands = risk_commands.get_messages();
        assert_eq!(commands.len(), 1);
        assert!(matches!(
            &commands[0],
            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_pending_activation_trailing_stop_limit_not_matched_as_plain_limit(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        // Market is already through the limit price, but below the activation price
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-1"))
            .side(OrderSide::Sell)
            .price(Price::from("5010.00"))
            .quantity(Quantity::from(1))
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .limit_offset(dec!(5))
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        // The market falls below where the trailed trigger would sit (4990)
        // while staying below the 5050 activation price: must not release.
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "4985.00", "4986.00"));

        assert!(risk_commands.get_messages().is_empty());
        assert!(
            emulator
                .borrow()
                .get_submit_order_commands()
                .contains_key(&client_order_id),
            "order must remain held by the emulator before activation"
        );
    }

    #[rstest]
    fn test_trailing_stop_with_preset_trigger_activates_and_triggers(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        // Both prices preset: the 5045 trigger must stay inert until 5050 activates
        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-2"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_price(Price::from("5045.00"))
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        assert!(
            risk_commands.get_messages().is_empty(),
            "preset trigger must not release before activation"
        );

        // Rally touches the 5050 activation price; the trailed candidate does
        // not improve on the preset trigger, so no price update is emitted.
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));

        assert!(risk_commands.get_messages().is_empty());

        // Bid falls through the preset trigger: the order must release
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));

        let commands = risk_commands.get_messages();
        assert_eq!(commands.len(), 1);
        assert!(matches!(
            &commands[0],
            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_trailing_stop_activates_despite_trailing_calculate_error(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        // Quote-only data: `LastOrBidAsk` trailing calculation errors without a last
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-3"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_price(Price::from("5045.00"))
            .trigger_type(TriggerType::LastOrBidAsk)
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        // Activation touches while the trailing calculation errors: the preset
        // trigger must still be honored once the market crosses it.
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));

        let commands = risk_commands.get_messages();
        assert_eq!(commands.len(), 1);
        assert!(matches!(
            &commands[0],
            TradingCommand::SubmitOrder(command) if command.client_order_id == client_order_id
        ));
    }

    #[rstest]
    fn test_basis_point_trailing_stop_limit_releases_on_price_increment(
        mut instrument: CryptoPerpetual,
    ) {
        instrument.price_increment = Price::from("0.25");
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("1500.00"));
        core.set_ask_raw(Price::from("1500.25"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-BPS"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_type(TriggerType::BidAsk)
            .trailing_offset(dec!(33))
            .limit_offset(dec!(37))
            .trailing_offset_type(TrailingOffsetType::BasisPoints)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "1500.00", "1500.25"));

        // From the 1500.00 bid, 33 bps is 1495.05 and 37 bps is 1494.45, both between increments
        let trailed_trigger = cache
            .borrow()
            .order(&client_order_id)
            .unwrap()
            .trigger_price();
        let trailed_price = cache.borrow().order(&client_order_id).unwrap().price();

        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "1494.00", "1494.25"));

        assert_eq!(trailed_trigger, Some(Price::from("1495.00")));
        assert_eq!(trailed_price, Some(Price::from("1494.50")));
        let commands = risk_commands.get_messages();

        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
            panic!("expected one released SubmitOrder, was {commands:?}");
        };

        assert_eq!(released.client_order_id, client_order_id);
        let cache = cache.borrow();
        let released_order = cache.order(&client_order_id).unwrap();
        assert_eq!(released_order.order_type(), OrderType::Limit);
        assert_eq!(released_order.price(), Some(Price::from("1494.50")));
    }

    #[rstest]
    fn test_trailing_stop_limit_without_limit_price_triggered_after_submit(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-1"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_price(Price::from("5045.00"))
            .trigger_type(TriggerType::LastOrBidAsk)
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .limit_offset(dec!(5))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        // Quote-only data: activation touches but `LastOrBidAsk` cannot price the limit
        // without a last, so the crossed trigger must hold the order rather than release it.
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));

        let held_price = cache.borrow().order(&client_order_id).unwrap().price();
        let held_commands = risk_commands.get_messages().len();

        // The first trade prices the limit, and the next update releases it
        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, "5044.00"));
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5044.00", "5045.00"));

        assert_eq!(held_price, None);
        assert_eq!(held_commands, 0);
        let commands = risk_commands.get_messages();

        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
            panic!("expected one released SubmitOrder, was {commands:?}");
        };

        assert_eq!(released.client_order_id, client_order_id);
        let cache = cache.borrow();
        let released_order = cache.order(&client_order_id).unwrap();
        assert_eq!(released_order.order_type(), OrderType::Limit);
        assert_eq!(released_order.price(), Some(Price::from("5039.00")));
    }

    #[rstest]
    fn test_trailing_stop_limit_without_limit_price_triggered_by_modify(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        let (handler, risk_commands): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
            get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::risk_engine_queue_execute(),
            handler,
        );
        add_instrument_to_cache(&cache, &instrument);

        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5000.00"));
        core.set_ask_raw(Price::from("5001.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = OrderTestBuilder::new(OrderType::TrailingStopLimit)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-TRAILING-LIMIT-2"))
            .side(OrderSide::Sell)
            .quantity(Quantity::from(1))
            .trigger_type(TriggerType::LastOrBidAsk)
            .activation_price(Price::from("5050.00"))
            .trailing_offset(dec!(10))
            .limit_offset(dec!(5))
            .trailing_offset_type(TrailingOffsetType::Price)
            .emulation_trigger(TriggerType::BidAsk)
            .build();
        let client_order_id = order.client_order_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);
        let _exec_events =
            register_exec_event_handler(cache.clone(), "ExecEngine.process.trailing_modify");
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));

        let modify = ModifyOrder::new(
            order.trader_id(),
            None,
            order.strategy_id(),
            instrument.id(),
            client_order_id,
            None,
            None,
            None,
            Some(Price::from("5060.00")),
            UUID4::new(),
            0.into(),
            None,
            None,
        );
        emulator.borrow_mut().handle_modify_order(&modify);

        let held_commands = risk_commands.get_messages().len();
        let held_in_core =
            emulator.borrow().matching_cores[&instrument.id()].order_exists(client_order_id);

        let (held_status, held_trigger, held_price) = {
            let cache = cache.borrow();
            let held_order = cache.order(&client_order_id).unwrap();
            (
                held_order.status(),
                held_order.trigger_price(),
                held_order.price(),
            )
        };

        // The first trade prices the limit, and the next update releases it
        emulator
            .borrow_mut()
            .on_trade_tick(create_trade_tick(&instrument, "5055.00"));
        emulator
            .borrow_mut()
            .on_quote_tick(create_quote_tick(&instrument, "5055.00", "5056.00"));

        assert_eq!(held_commands, 0);
        assert!(held_in_core);
        assert_eq!(held_status, OrderStatus::Emulated);
        assert_eq!(held_trigger, Some(Price::from("5060.00")));
        assert_eq!(held_price, None);
        let commands = risk_commands.get_messages();

        let [TradingCommand::SubmitOrder(released)] = commands.as_slice() else {
            panic!("expected one released SubmitOrder, was {commands:?}");
        };

        assert_eq!(released.client_order_id, client_order_id);
        let cache = cache.borrow();
        let released_order = cache.order(&client_order_id).unwrap();
        assert_eq!(released_order.order_type(), OrderType::Limit);
        assert_eq!(released_order.price(), Some(Price::from("5050.00")));
    }

    #[rstest]
    fn test_cancel_emulated_oco_leg_cancels_sibling(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let client_order_id_a = ClientOrderId::from("O-OCO-A");
        let client_order_id_b = ClientOrderId::from("O-OCO-B");
        let order_a = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .strategy_id(StrategyId::from("STRATEGY-001"))
            .client_order_id(client_order_id_a)
            .side(OrderSide::Sell)
            .trigger_price(Price::from("4900.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .contingency_type(ContingencyType::Oco)
            .linked_order_ids(vec![client_order_id_b])
            .build();
        let order_b = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .strategy_id(StrategyId::from("STRATEGY-001"))
            .client_order_id(client_order_id_b)
            .side(OrderSide::Sell)
            .trigger_price(Price::from("4950.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .contingency_type(ContingencyType::Oco)
            .linked_order_ids(vec![client_order_id_a])
            .build();
        cache
            .borrow_mut()
            .add_order(order_a.clone(), None, None, false)
            .unwrap();
        cache
            .borrow_mut()
            .add_order(order_b.clone(), None, None, false)
            .unwrap();

        for order in [&order_a, &order_b] {
            let command = create_submit_order(&instrument, order);
            emulator
                .borrow_mut()
                .cache_submit_order_command(command.clone());
            emulator.borrow_mut().handle_submit_order(&command);
        }
        assert_eq!(
            cache.borrow().order(&client_order_id_a).unwrap().status(),
            OrderStatus::Emulated
        );
        assert_eq!(
            cache.borrow().order(&client_order_id_b).unwrap().status(),
            OrderStatus::Emulated
        );

        let cancel = CancelOrder::new(
            order_a.trader_id(),
            None,
            order_a.strategy_id(),
            instrument.id(),
            client_order_id_a,
            None,
            UUID4::new(),
            0.into(),
            None,
            None,
        );
        msgbus::send_trading_command(
            MessagingSwitchboard::order_emulator_execute(),
            TradingCommand::CancelOrder(cancel),
        );

        let cache = cache.borrow();
        assert_eq!(
            cache.order(&client_order_id_a).unwrap().status(),
            OrderStatus::Canceled
        );
        assert_eq!(
            cache.order(&client_order_id_b).unwrap().status(),
            OrderStatus::Canceled,
            "OCO sibling must be canceled through the deferred self-published event"
        );
    }

    #[rstest]
    fn test_reentrant_command_from_event_handler_does_not_panic(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let order = create_stop_market_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());

        // A strategy-style subscriber issuing an emulator command synchronously
        // from within its own order-event handler.
        let cancel = CancelOrder::new(
            order.trader_id(),
            None,
            strategy_id,
            instrument.id(),
            client_order_id,
            None,
            UUID4::new(),
            0.into(),
            None,
            None,
        );
        let handler = TypedHandler::from(move |event: &OrderEventAny| {
            if matches!(event, OrderEventAny::Emulated(_)) {
                msgbus::send_trading_command(
                    MessagingSwitchboard::order_emulator_execute(),
                    TradingCommand::CancelOrder(cancel.clone()),
                );
            }
        });
        msgbus::subscribe_order_events(format!("events.order.{strategy_id}").into(), handler, None);

        // Must not panic on the reentrant borrow; the command is deferred and drained
        msgbus::send_trading_command(
            MessagingSwitchboard::order_emulator_execute(),
            TradingCommand::SubmitOrder(command),
        );

        assert!(
            cache.borrow().order(&client_order_id).unwrap().is_closed(),
            "deferred cancel must be processed after the active call completes"
        );
    }

    #[rstest]
    fn test_released_order_preserves_chronological_event_history(instrument: CryptoPerpetual) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);
        let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
        core.set_bid_raw(Price::from("5099.00"));
        core.set_ask_raw(Price::from("5101.00"));
        emulator
            .borrow_mut()
            .matching_cores
            .insert(instrument.id(), core);

        let order = create_stop_limit_order(&instrument, TriggerType::BidAsk);
        let client_order_id = order.client_order_id();
        let original_init = OrderEventAny::Initialized(order.init_event().clone());
        let command = create_submit_order(&instrument, &order);
        cache
            .borrow_mut()
            .add_order(order, None, None, false)
            .unwrap();

        emulator
            .borrow_mut()
            .cache_submit_order_command(command.clone());
        emulator.borrow_mut().handle_submit_order(&command);

        let cache = cache.borrow();
        let transformed = cache.order(&client_order_id).unwrap();
        let events = transformed.events();

        assert_eq!(
            events.first(),
            Some(&&original_init),
            "released order history must start with the original OrderInitialized event, was {events:?}",
        );
    }

    #[rstest]
    fn test_on_start_cancels_emulated_child_of_closed_positionless_parent(
        instrument: CryptoPerpetual,
    ) {
        let (_clock, cache, emulator) = create_emulator();
        add_instrument_to_cache(&cache, &instrument);

        let mut parent = OrderTestBuilder::new(OrderType::Limit)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-PARENT"))
            .side(OrderSide::Buy)
            .price(Price::from("5000.00"))
            .quantity(Quantity::from(1))
            .submit(true)
            .build();
        parent
            .apply(OrderEventAny::Canceled(OrderCanceled::new(
                parent.trader_id(),
                parent.strategy_id(),
                parent.instrument_id(),
                parent.client_order_id(),
                UUID4::new(),
                0.into(),
                0.into(),
                false,
                None,
                None,
                None,
            )))
            .unwrap();
        assert!(parent.is_closed());
        assert!(parent.position_id().is_none());

        let mut child = OrderTestBuilder::new(OrderType::StopMarket)
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from("O-CHILD"))
            .side(OrderSide::Sell)
            .trigger_price(Price::from("4900.00"))
            .quantity(Quantity::from(1))
            .emulation_trigger(TriggerType::BidAsk)
            .parent_order_id(parent.client_order_id())
            .build();
        child
            .apply(OrderEventAny::Emulated(OrderEmulated::new(
                child.trader_id(),
                child.strategy_id(),
                child.instrument_id(),
                child.client_order_id(),
                UUID4::new(),
                0.into(),
                0.into(),
            )))
            .unwrap();
        assert_eq!(child.status(), OrderStatus::Emulated);

        cache
            .borrow_mut()
            .add_order(parent.clone(), None, None, false)
            .unwrap();
        cache
            .borrow_mut()
            .add_order(child.clone(), None, None, false)
            .unwrap();

        emulator.borrow_mut().on_start().unwrap();

        assert!(
            cache
                .borrow()
                .order(&child.client_order_id())
                .unwrap()
                .is_closed(),
            "emulated child of a closed position-less parent must be canceled on reactivation"
        );
    }
}