nautilus_execution/engine/

mod.rs

// -------------------------------------------------------------------------------------------------
//  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.
// -------------------------------------------------------------------------------------------------

//! Provides a generic `ExecutionEngine` for all environments.
//!
//! The execution engines primary responsibility is to orchestrate interactions
//! between the `ExecutionClient` instances, and the rest of the platform. This
//! includes sending commands to, and receiving events from, the trading venue
//! endpoints via its registered execution clients.

pub mod config;
pub mod position;
pub mod stubs;

use std::{
    cell::{Cell, RefCell, RefMut},
    collections::{HashMap, HashSet},
    fmt::{Debug, Display},
    rc::Rc,
    time::SystemTime,
};

use ahash::{AHashMap, AHashSet};
use config::ExecutionEngineConfig;
use futures::future::join_all;
use indexmap::{IndexMap, IndexSet};
use nautilus_common::{
    cache::{Cache, PositionRef},
    clients::ExecutionClient,
    clock::Clock,
    enums::LogColor,
    generators::position_id::PositionIdGenerator,
    log_info,
    logging::{CMD, EVT, RECV, SEND},
    messages::{
        ExecutionReport,
        execution::{
            BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
            QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList, TradingCommand,
        },
    },
    msgbus::{
        self, MessagingSwitchboard, TypedHandler, TypedIntoHandler, get_message_bus,
        switchboard::{self},
    },
    runner::{
        TradingCommandMessage, capture_trading_cmd, trading_cmd_is_dispatching,
        try_get_trading_cmd_sender,
    },
    timer::{TimeEvent, TimeEventCallback},
};
use nautilus_core::{
    DurationNanos, UUID4, UnixNanos, WeakCell,
    datetime::{mins_to_secs, secs_to_nanos},
};
use nautilus_model::{
    accounts::Account,
    enums::{
        AccountType, ContingencyType, OmsType, OrderStatus, OrderType, PositionSide, TimeInForce,
    },
    events::{
        OrderAccepted, OrderDenied, OrderDeniedReason, OrderEvent, OrderEventAny, OrderFillVoided,
        OrderFilled, OrderInitialized, PositionChanged, PositionClosed, PositionEvent,
        PositionOpened,
    },
    identifiers::{
        AccountId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId,
        TradeId, Venue, VenueOrderId,
    },
    instruments::{Instrument, InstrumentAny},
    orderbook::own::{OwnBookOrder, OwnOrderBook, should_handle_own_book_order},
    orders::{Order, OrderAny, OrderError},
    position::{Position, PositionReplayEvent},
    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
    types::{Money, Quantity},
};
use position::CorrectedPosition;
pub use position::{PositionStateSnapshot, SnapshotAnchorer};
use rust_decimal::Decimal;

use crate::{
    client::ExecutionClientAdapter,
    reconciliation::{
        check_position_reconciliation, generate_external_order_status_events,
        generate_reconciliation_order_events, generate_reconciliation_order_pre_fill_events,
        generate_reconciliation_order_snapshot_events, reconcile_fill_report as reconcile_fill,
    },
};

const TIMER_SNAPSHOT_POSITIONS: &str = "ExecEngine_SNAPSHOT_POSITIONS";
const TIMER_PURGE_CLOSED_ORDERS: &str = "ExecEngine_PURGE_CLOSED_ORDERS";
const TIMER_PURGE_CLOSED_POSITIONS: &str = "ExecEngine_PURGE_CLOSED_POSITIONS";
const TIMER_PURGE_ACCOUNT_EVENTS: &str = "ExecEngine_PURGE_ACCOUNT_EVENTS";

/// Central execution engine responsible for orchestrating order routing and execution.
///
/// The execution engine manages the entire order lifecycle from submission to completion,
/// handling routing to appropriate execution clients, position management, and event
/// processing. It supports multiple execution venues through registered clients and
/// provides sophisticated order management capabilities.
///
/// An order dispatched to a registered or external execution client keeps its `Initialized` (or
/// `Released`) status until the client's first status event is applied, so cached status alone
/// cannot distinguish it from an order that has not been routed yet. The engine records each
/// dispatch until that first status transition, and a later submit command naming the order is
/// stale for it: the engine neither routes the order to a client again nor denies it when the
/// later command fails validation.
pub struct ExecutionEngine {
    clock: Rc<RefCell<dyn Clock>>,
    cache: Rc<RefCell<Cache>>,
    clients: IndexMap<ClientId, ExecutionClientAdapter>,
    default_client_id: Option<ClientId>,
    routing_map: AHashMap<Venue, ClientId>,
    instrument_venues: AHashSet<Venue>,
    oms_overrides: AHashMap<StrategyId, OmsType>,
    external_clients: HashSet<ClientId>,
    pos_id_generator: PositionIdGenerator,
    config: ExecutionEngineConfig,
    orders_dispatched: RefCell<AHashSet<ClientOrderId>>,
    command_count: Cell<u64>,
    event_count: u64,
    report_count: u64,
    filtered_unclaimed_external_order_count: u64,
    snapshot_anchorer: Option<SnapshotAnchorer>,
}

impl Debug for ExecutionEngine {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(ExecutionEngine))
            .field("client_count", &self.clients.len())
            .finish()
    }
}

impl ExecutionEngine {
    /// Creates a new [`ExecutionEngine`] instance.
    pub fn new(
        clock: Rc<RefCell<dyn Clock>>,
        cache: Rc<RefCell<Cache>>,
        config: Option<ExecutionEngineConfig>,
    ) -> Self {
        let trader_id = get_message_bus().borrow().trader_id;
        Self {
            clock: clock.clone(),
            cache,
            clients: IndexMap::new(),
            default_client_id: None,
            routing_map: AHashMap::new(),
            instrument_venues: AHashSet::new(),
            oms_overrides: AHashMap::new(),
            external_clients: config
                .as_ref()
                .and_then(|c| c.external_clients.clone())
                .unwrap_or_default()
                .into_iter()
                .collect(),
            pos_id_generator: PositionIdGenerator::new(trader_id, clock),
            config: config.unwrap_or_default(),
            orders_dispatched: RefCell::new(AHashSet::new()),
            command_count: Cell::new(0),
            event_count: 0,
            report_count: 0,
            filtered_unclaimed_external_order_count: 0,
            snapshot_anchorer: None,
        }
    }

    /// Registers all message bus handlers for the execution engine.
    pub fn register_msgbus_handlers(engine: &Rc<RefCell<Self>>) {
        let weak = WeakCell::from(Rc::downgrade(engine));

        let weak1 = weak.clone();
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::exec_engine_execute(),
            TypedIntoHandler::from(move |cmd: TradingCommand| {
                if let Some(rc) = weak1.upgrade() {
                    rc.borrow().execute(cmd);
                }
            }),
        );

        // Queued endpoint for deferred command execution (re-entrancy safe),
        // with direct dispatch when no sender is installed.
        msgbus::register_trading_command_endpoint(
            MessagingSwitchboard::exec_engine_queue_execute(),
            TypedIntoHandler::from(move |cmd: TradingCommand| {
                let endpoint = MessagingSwitchboard::exec_engine_execute();
                if trading_cmd_is_dispatching() {
                    capture_trading_cmd(TradingCommandMessage::new(endpoint, cmd));
                } else if let Some(sender) = try_get_trading_cmd_sender() {
                    sender.execute(TradingCommandMessage::new(endpoint, cmd));
                } else {
                    msgbus::send_trading_command(endpoint, cmd);
                }
            }),
        );

        let weak2 = weak.clone();
        msgbus::register_order_event_endpoint(
            MessagingSwitchboard::exec_engine_process(),
            TypedIntoHandler::from(move |event: OrderEventAny| {
                if let Some(rc) = weak2.upgrade() {
                    rc.borrow_mut().process(&event);
                }
            }),
        );

        let weak3 = weak;
        msgbus::register_execution_report_endpoint(
            MessagingSwitchboard::exec_engine_reconcile_execution_report(),
            TypedIntoHandler::from(move |report: ExecutionReport| {
                if let Some(rc) = weak3.upgrade() {
                    rc.borrow_mut().reconcile_execution_report(&report);
                }
            }),
        );
    }

    /// Returns the total count of trading commands received by the engine.
    #[must_use]
    pub fn command_count(&self) -> u64 {
        self.command_count.get()
    }

    /// Returns the total count of order events received by the engine.
    #[must_use]
    pub const fn event_count(&self) -> u64 {
        self.event_count
    }

    /// Returns the total count of execution reports received by the engine.
    #[must_use]
    pub const fn report_count(&self) -> u64 {
        self.report_count
    }

    /// Returns the count of unclaimed external venue orders filtered by execution reconciliation.
    #[must_use]
    pub const fn filtered_unclaimed_external_order_count(&self) -> u64 {
        self.filtered_unclaimed_external_order_count
    }

    /// Subscribes to instrument updates for a venue via the message bus.
    ///
    /// When instruments are published by the `DataEngine`, the handler routes
    /// them to every client whose own venue matches, plus the client routed to that
    /// venue, if any. Repeated subscriptions for the same venue are ignored.
    pub fn subscribe_venue_instruments(engine: &Rc<RefCell<Self>>, venue: Venue) {
        if !engine.borrow_mut().instrument_venues.insert(venue) {
            return;
        }

        let weak = WeakCell::from(Rc::downgrade(engine));
        let pattern = switchboard::get_instruments_pattern(venue);

        let handler = TypedHandler::from(move |instrument: &InstrumentAny| {
            if let Some(rc) = weak.upgrade() {
                let venue = instrument.id().venue;
                let mut engine = rc.borrow_mut();
                let routed_client = engine.routing_map.get(&venue).copied();
                for adapter in engine.clients.values_mut() {
                    if adapter.venue == venue || Some(adapter.client_id) == routed_client {
                        adapter.on_instrument(instrument.clone());
                    }
                }
            }
        });

        msgbus::subscribe_instruments(pattern, handler, None);
        log::info!("Subscribed to instrument updates for venue {venue}");
    }

    #[must_use]
    /// Returns the position ID count for the specified strategy.
    pub fn position_id_count(&self, strategy_id: StrategyId) -> usize {
        self.pos_id_generator.count(strategy_id)
    }

    #[must_use]
    /// Returns a reference to the cache.
    pub fn cache(&self) -> &Rc<RefCell<Cache>> {
        &self.cache
    }

    #[must_use]
    /// Returns a reference to the configuration.
    pub const fn config(&self) -> &ExecutionEngineConfig {
        &self.config
    }

    /// Sets the cache snapshot anchorer.
    ///
    /// The system event-store integration installs this while a run is open. Passing
    /// `None` disables anchor recording for later cache snapshots.
    pub fn set_snapshot_anchorer(&mut self, anchorer: Option<SnapshotAnchorer>) {
        self.snapshot_anchorer = anchorer;
    }

    #[must_use]
    /// Checks the integrity of cached execution data.
    pub fn check_integrity(&self) -> bool {
        self.cache.borrow_mut().check_integrity()
    }

    #[must_use]
    /// Returns true if all registered execution clients are connected.
    pub fn check_connected(&self) -> bool {
        self.clients.values().all(|c| c.is_connected())
    }

    #[must_use]
    /// Returns true if all registered execution clients are disconnected.
    pub fn check_disconnected(&self) -> bool {
        self.clients.values().all(|c| !c.is_connected())
    }

    /// Returns connection status for each registered client.
    #[must_use]
    pub fn client_connection_status(&self) -> Vec<(ClientId, bool)> {
        self.clients
            .values()
            .map(|c| (c.client_id(), c.is_connected()))
            .collect()
    }

    #[must_use]
    /// Checks for residual positions and orders in the cache.
    pub fn check_residuals(&self) -> bool {
        self.cache.borrow().check_residuals()
    }

    #[must_use]
    /// Returns the set of instruments that have external order claims.
    pub fn get_external_order_claims_instruments(&self) -> HashSet<InstrumentId> {
        self.cache
            .borrow()
            .external_order_claim_instrument_ids(None)
            .into_iter()
            .collect()
    }

    #[must_use]
    /// Returns the configured external client IDs.
    pub fn get_external_client_ids(&self) -> HashSet<ClientId> {
        self.external_clients.clone()
    }

    #[must_use]
    /// Returns any external order claim for the given instrument ID.
    pub fn get_external_order_claim(&self, instrument_id: &InstrumentId) -> Option<StrategyId> {
        self.cache.borrow().external_order_claim(instrument_id)
    }

    /// Registers a new execution client without assigning venue or default routing.
    ///
    /// Callers configure venue and fallback routing separately with
    /// [`Self::register_venue_routing`] and [`Self::set_default_client`].
    ///
    /// # Errors
    ///
    /// Returns an error if a client with the same ID is already registered.
    pub fn register_client(&mut self, client: Box<dyn ExecutionClient>) -> anyhow::Result<()> {
        let client_id = client.client_id();

        if self.clients.contains_key(&client_id) {
            anyhow::bail!("Client already registered with ID {client_id}");
        }

        let adapter = ExecutionClientAdapter::new(client);

        log::debug!("Registered client {client_id}");
        self.clients.insert(client_id, adapter);
        Ok(())
    }

    /// Registers a default execution client for fallback routing.
    pub fn register_default_client(&mut self, client: Box<dyn ExecutionClient>) {
        let client_id = client.client_id();
        let adapter = ExecutionClientAdapter::new(client);

        self.clients.insert(client_id, adapter);
        self.default_client_id = Some(client_id);
        log::debug!("Registered default client {client_id}");
    }

    /// Marks an already-registered client as the default for fallback routing.
    ///
    /// # Errors
    ///
    /// Returns an error if no client is registered with the given ID, or a default
    /// client has already been set.
    pub fn set_default_client(&mut self, client_id: ClientId) -> anyhow::Result<()> {
        if self.default_client_id.is_some() {
            anyhow::bail!("default client already registered");
        }

        if !self.clients.contains_key(&client_id) {
            anyhow::bail!("No client registered with ID {client_id}");
        }
        self.default_client_id = Some(client_id);
        log::debug!("Set client {client_id} as default");
        Ok(())
    }

    #[must_use]
    /// Returns a reference to the execution client registered with the given ID.
    pub fn get_client(&self, client_id: &ClientId) -> Option<&dyn ExecutionClient> {
        self.clients.get(client_id).map(|a| a.client.as_ref())
    }

    #[must_use]
    /// Returns a mutable reference to the execution client adapter registered with the given ID.
    pub fn get_client_adapter_mut(
        &mut self,
        client_id: &ClientId,
    ) -> Option<&mut ExecutionClientAdapter> {
        self.clients.get_mut(client_id)
    }

    /// Generates mass status for the given client.
    ///
    /// # Errors
    ///
    /// Returns an error if the client is not found or mass status generation fails.
    pub async fn generate_mass_status(
        &mut self,
        client_id: &ClientId,
        lookback_mins: Option<u64>,
    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
        if let Some(client) = self.get_client_adapter_mut(client_id) {
            client.generate_mass_status(lookback_mins).await
        } else {
            anyhow::bail!("Client {client_id} not found")
        }
    }

    /// Registers an external order with the execution client for tracking.
    ///
    /// This is called after reconciliation creates an external order, allowing the
    /// execution client to track it for subsequent events (e.g., cancellations).
    pub fn register_external_order(
        &self,
        client_order_id: ClientOrderId,
        venue_order_id: VenueOrderId,
        instrument_id: InstrumentId,
        strategy_id: StrategyId,
        ts_init: UnixNanos,
    ) {
        let venue = instrument_id.venue;
        // Prefer the cached origin over venue routing so tracking lands on the
        // client whose stream materialized the order.
        let client_id = self
            .cache
            .borrow()
            .client_id(&client_order_id)
            .copied()
            .or_else(|| self.routing_map.get(&venue).copied())
            .or(self.default_client_id);

        if let Some(client_id) = client_id
            && let Some(client) = self.clients.get(&client_id)
        {
            client.register_external_order(
                client_order_id,
                venue_order_id,
                instrument_id,
                strategy_id,
                ts_init,
            );
        }
    }

    #[must_use]
    /// Returns all registered execution client IDs.
    pub fn client_ids(&self) -> Vec<ClientId> {
        self.clients.keys().copied().collect()
    }

    #[must_use]
    /// Returns mutable access to all registered execution clients.
    pub fn get_clients_mut(&mut self) -> Vec<&mut ExecutionClientAdapter> {
        self.clients.values_mut().collect()
    }

    /// Returns all registered execution clients.
    #[must_use]
    pub fn get_all_clients(&self) -> Vec<&dyn ExecutionClient> {
        self.clients.values().map(|a| a.client.as_ref()).collect()
    }

    #[must_use]
    /// Returns execution clients that would handle the given orders.
    ///
    /// This method first attempts to resolve each order's originating client from the cache,
    /// then falls back to venue routing for any orders without a cached client.
    pub fn get_clients_for_orders(&self, orders: &[OrderAny]) -> Vec<&dyn ExecutionClient> {
        let mut client_ids: IndexSet<ClientId> = IndexSet::new();
        let mut venues: IndexSet<Venue> = IndexSet::new();

        // Collect client IDs from cache and venues for fallback
        for order in orders {
            venues.insert(order.instrument_id().venue);
            if let Some(client_id) = self.cache.borrow().client_id(&order.client_order_id()) {
                client_ids.insert(*client_id);
            }
        }

        let mut clients: Vec<&dyn ExecutionClient> = Vec::new();

        // Add clients for cached client IDs (orders go back to originating client)
        for client_id in &client_ids {
            if let Some(adapter) = self.clients.get(client_id)
                && !clients.iter().any(|c| c.client_id() == adapter.client_id)
            {
                clients.push(adapter.client.as_ref());
            }
        }

        // Add clients for venue routing (for orders not in cache)
        for venue in &venues {
            let resolved_id = self
                .routing_map
                .get(venue)
                .copied()
                .or(self.default_client_id);

            if let Some(adapter) = resolved_id.and_then(|id| self.clients.get(&id))
                && !clients.iter().any(|c| c.client_id() == adapter.client_id)
            {
                clients.push(adapter.client.as_ref());
            }
        }

        clients
    }

    /// Sets routing for a specific venue to a given client ID.
    ///
    /// # Errors
    ///
    /// Returns an error if the client ID is not registered or the venue already
    /// routes to a different client.
    pub fn register_venue_routing(
        &mut self,
        client_id: ClientId,
        venue: Venue,
    ) -> anyhow::Result<()> {
        if !self.clients.contains_key(&client_id) {
            anyhow::bail!("No client registered with ID {client_id}");
        }

        if let Some(existing_client_id) = self.routing_map.get(&venue)
            && *existing_client_id != client_id
        {
            anyhow::bail!(
                "Venue {venue} already routed to {existing_client_id}, \
                 cannot re-route to {client_id}"
            );
        }

        self.routing_map.insert(venue, client_id);
        log::info!("Set client {client_id} routing for {venue}");
        Ok(())
    }

    /// Registers the OMS (Order Management System) type for a strategy.
    ///
    /// If an OMS type is already registered for this strategy, it will be overridden.
    pub fn register_oms_type(&mut self, strategy_id: StrategyId, oms_type: OmsType) {
        self.oms_overrides.insert(strategy_id, oms_type);
        log::info!("Registered OMS::{oms_type:?} for {strategy_id}");
    }

    /// Registers external order claims for a strategy.
    ///
    /// Venue-sourced external orders, fills, and materialized reconciliation activity for matching
    /// instruments will be associated with the strategy.
    ///
    /// This operation is atomic: either all instruments are registered or none are.
    ///
    /// # Errors
    ///
    /// Returns an error if any instrument already has a registered claim.
    pub fn register_external_order_claims(
        &mut self,
        strategy_id: StrategyId,
        instrument_ids: &HashSet<InstrumentId>,
    ) -> anyhow::Result<()> {
        let instrument_ids: Vec<_> = instrument_ids.iter().copied().collect();
        self.cache
            .borrow_mut()
            .register_external_order_claims(strategy_id, &instrument_ids)?;

        if !instrument_ids.is_empty() {
            log::info!("Registered external order claims for {strategy_id}: {instrument_ids:?}");
        }

        Ok(())
    }

    /// Deregisters all external order claims owned by `strategy_id`.
    ///
    /// # Panics
    ///
    /// Panics if the shared cache is already borrowed.
    pub fn deregister_external_order_claims(&mut self, strategy_id: StrategyId) {
        self.cache
            .borrow_mut()
            .set_external_order_claims(strategy_id, &[])
            .expect("clearing external order claims cannot fail");
    }

    /// # Errors
    ///
    /// Returns an error if no client is registered with the given ID.
    pub fn deregister_client(&mut self, client_id: ClientId) -> anyhow::Result<()> {
        if self.clients.shift_remove(&client_id).is_some() {
            if self.default_client_id == Some(client_id) {
                self.default_client_id = None;
            }

            // Remove from routing map if present
            self.routing_map
                .retain(|_, mapped_id| mapped_id != &client_id);
            log::info!("Deregistered client {client_id}");
            Ok(())
        } else {
            anyhow::bail!("No client registered with ID {client_id}")
        }
    }

    /// Connects all registered execution clients concurrently.
    ///
    /// Connection failures are logged but do not prevent the node from running.
    pub async fn connect(&mut self) {
        let futures: Vec<_> = self
            .get_clients_mut()
            .into_iter()
            .map(ExecutionClientAdapter::connect)
            .collect();

        let results = join_all(futures).await;

        for error in results.into_iter().filter_map(Result::err) {
            log::error!("Failed to connect execution client: {error:#}");
        }
    }

    /// Disconnects all registered execution clients concurrently.
    ///
    /// # Errors
    ///
    /// Returns an error if any client fails to disconnect.
    pub async fn disconnect(&mut self) -> anyhow::Result<()> {
        let futures: Vec<_> = self
            .get_clients_mut()
            .into_iter()
            .map(ExecutionClientAdapter::disconnect)
            .collect();

        let results = join_all(futures).await;
        let errors: Vec<_> = results.into_iter().filter_map(Result::err).collect();

        if errors.is_empty() {
            Ok(())
        } else {
            let error_msgs: Vec<_> = errors.iter().map(ToString::to_string).collect();
            anyhow::bail!(
                "Failed to disconnect execution clients: {}",
                error_msgs.join("; ")
            )
        }
    }

    /// Sets the `manage_own_order_books` configuration option.
    pub fn set_manage_own_order_books(&mut self, value: bool) {
        self.config.manage_own_order_books = value;
    }

    /// Starts the position snapshot timer if configured.
    #[expect(
        clippy::missing_panics_doc,
        reason = "timer registration is not expected to fail"
    )]
    pub fn start_snapshot_timer(&mut self) {
        if let Some(interval_secs) = self
            .config
            .snapshot_positions_interval_secs
            .filter(|&secs| secs > 0.0)
            && !self
                .clock
                .borrow()
                .timer_names()
                .contains(&TIMER_SNAPSHOT_POSITIONS)
        {
            let interval_ns = match secs_to_nanos(interval_secs) {
                Ok(ns) => ns,
                Err(e) => {
                    log::error!("Cannot start position snapshots timer: {e}");
                    return;
                }
            };
            let clock = self.clock.clone();
            let cache = self.cache.clone();
            let debug = self.config.debug;

            let callback_fn: Rc<dyn Fn(TimeEvent)> = Rc::new(move |_event| {
                Self::snapshot_open_positions(&clock, &cache, debug);
            });
            let callback = TimeEventCallback::from(callback_fn);

            log::info!("Starting position snapshots timer at {interval_secs} second intervals");
            self.clock
                .borrow_mut()
                .set_timer_ns(
                    TIMER_SNAPSHOT_POSITIONS,
                    DurationNanos::new(interval_ns),
                    None,
                    None,
                    Some(callback),
                    None,
                    None,
                )
                .expect("Failed to set position snapshots timer");
        }
    }

    /// Stops the position snapshot timer if running.
    pub fn stop_snapshot_timer(&mut self) {
        let timer_registered = self
            .clock
            .borrow()
            .timer_names()
            .contains(&TIMER_SNAPSHOT_POSITIONS);

        if timer_registered {
            log::info!("Canceling position snapshots timer");
            self.clock
                .borrow_mut()
                .cancel_timer(TIMER_SNAPSHOT_POSITIONS);
        }
    }

    /// Starts the purge timers if configured.
    pub fn start_purge_timers(&mut self) {
        if let Some(interval_mins) = self
            .config
            .purge_closed_orders_interval_mins
            .filter(|&m| m > 0)
            && !self
                .clock
                .borrow()
                .timer_names()
                .contains(&TIMER_PURGE_CLOSED_ORDERS)
        {
            'purge_closed_orders: {
                let Ok(interval_ns) = DurationNanos::try_from_mins(u64::from(interval_mins)) else {
                    log::error!(
                        "Invalid purge_closed_orders_interval_mins {interval_mins}: minutes to nanoseconds conversion overflow"
                    );
                    break 'purge_closed_orders;
                };
                let buffer_mins = self.config.purge_closed_orders_buffer_mins.unwrap_or(0);
                let buffer_secs = mins_to_secs(u64::from(buffer_mins));
                let cache = self.cache.clone();
                let clock = self.clock.clone();

                let callback_fn: Rc<dyn Fn(TimeEvent)> = Rc::new(move |_event| {
                    let ts_now = clock.borrow().timestamp_ns();
                    cache.borrow_mut().purge_closed_orders(ts_now, buffer_secs);
                });
                let callback = TimeEventCallback::from(callback_fn);

                log::info!(
                    "Starting purge closed orders timer at {interval_mins} minute intervals"
                );

                if let Err(e) = self.clock.borrow_mut().set_timer_ns(
                    TIMER_PURGE_CLOSED_ORDERS,
                    interval_ns,
                    None,
                    None,
                    Some(callback),
                    None,
                    None,
                ) {
                    log::error!("Failed to set {TIMER_PURGE_CLOSED_ORDERS} timer: {e}");
                }
            }
        }

        if let Some(interval_mins) = self
            .config
            .purge_closed_positions_interval_mins
            .filter(|&m| m > 0)
            && !self
                .clock
                .borrow()
                .timer_names()
                .contains(&TIMER_PURGE_CLOSED_POSITIONS)
        {
            'purge_closed_positions: {
                let Ok(interval_ns) = DurationNanos::try_from_mins(u64::from(interval_mins)) else {
                    log::error!(
                        "Invalid purge_closed_positions_interval_mins {interval_mins}: minutes to nanoseconds conversion overflow"
                    );
                    break 'purge_closed_positions;
                };
                let buffer_mins = self.config.purge_closed_positions_buffer_mins.unwrap_or(0);
                let buffer_secs = mins_to_secs(u64::from(buffer_mins));
                let cache = self.cache.clone();
                let clock = self.clock.clone();

                let callback_fn: Rc<dyn Fn(TimeEvent)> = Rc::new(move |_event| {
                    let ts_now = clock.borrow().timestamp_ns();
                    cache
                        .borrow_mut()
                        .purge_closed_positions(ts_now, buffer_secs);
                });
                let callback = TimeEventCallback::from(callback_fn);

                log::info!(
                    "Starting purge closed positions timer at {interval_mins} minute intervals"
                );

                if let Err(e) = self.clock.borrow_mut().set_timer_ns(
                    TIMER_PURGE_CLOSED_POSITIONS,
                    interval_ns,
                    None,
                    None,
                    Some(callback),
                    None,
                    None,
                ) {
                    log::error!("Failed to set {TIMER_PURGE_CLOSED_POSITIONS} timer: {e}");
                }
            }
        }

        if let Some(interval_mins) = self
            .config
            .purge_account_events_interval_mins
            .filter(|&m| m > 0)
            && !self
                .clock
                .borrow()
                .timer_names()
                .contains(&TIMER_PURGE_ACCOUNT_EVENTS)
        {
            'purge_account_events: {
                let Ok(interval_ns) = DurationNanos::try_from_mins(u64::from(interval_mins)) else {
                    log::error!(
                        "Invalid purge_account_events_interval_mins {interval_mins}: minutes to nanoseconds conversion overflow"
                    );
                    break 'purge_account_events;
                };
                let lookback_mins = self.config.purge_account_events_lookback_mins.unwrap_or(0);
                let lookback_secs = mins_to_secs(u64::from(lookback_mins));
                let cache = self.cache.clone();
                let clock = self.clock.clone();

                let callback_fn: Rc<dyn Fn(TimeEvent)> = Rc::new(move |_event| {
                    let ts_now = clock.borrow().timestamp_ns();
                    cache
                        .borrow_mut()
                        .purge_account_events(ts_now, lookback_secs);
                });
                let callback = TimeEventCallback::from(callback_fn);

                log::info!(
                    "Starting purge account events timer at {interval_mins} minute intervals"
                );

                if let Err(e) = self.clock.borrow_mut().set_timer_ns(
                    TIMER_PURGE_ACCOUNT_EVENTS,
                    interval_ns,
                    None,
                    None,
                    Some(callback),
                    None,
                    None,
                ) {
                    log::error!("Failed to set {TIMER_PURGE_ACCOUNT_EVENTS} timer: {e}");
                }
            }
        }
    }

    /// Stops the purge timers if running.
    pub fn stop_purge_timers(&mut self) {
        let timer_names: Vec<String> = self
            .clock
            .borrow()
            .timer_names()
            .into_iter()
            .map(String::from)
            .collect();

        if timer_names.iter().any(|n| n == TIMER_PURGE_CLOSED_ORDERS) {
            log::info!("Canceling purge closed orders timer");
            self.clock
                .borrow_mut()
                .cancel_timer(TIMER_PURGE_CLOSED_ORDERS);
        }

        if timer_names
            .iter()
            .any(|n| n == TIMER_PURGE_CLOSED_POSITIONS)
        {
            log::info!("Canceling purge closed positions timer");
            self.clock
                .borrow_mut()
                .cancel_timer(TIMER_PURGE_CLOSED_POSITIONS);
        }

        if timer_names.iter().any(|n| n == TIMER_PURGE_ACCOUNT_EVENTS) {
            log::info!("Canceling purge account events timer");
            self.clock
                .borrow_mut()
                .cancel_timer(TIMER_PURGE_ACCOUNT_EVENTS);
        }
    }

    /// Creates snapshots of all open positions.
    pub fn snapshot_open_position_states(&self) {
        Self::snapshot_open_positions(&self.clock, &self.cache, self.config.debug);
    }

    fn snapshot_open_positions(
        clock: &Rc<RefCell<dyn Clock>>,
        cache: &Rc<RefCell<Cache>>,
        debug: bool,
    ) {
        let positions: Vec<Position> = cache
            .borrow()
            .positions_open(None, None, None, None, None)
            .into_iter()
            .map(|p| p.cloned())
            .collect();

        for position in positions {
            Self::publish_position_state_snapshot(clock, cache, debug, &position, true);
        }
    }

    #[expect(clippy::await_holding_refcell_ref)]
    /// Loads persistent state into cache and rebuilds indices.
    ///
    /// # Errors
    ///
    /// Returns an error if any cache operation fails.
    pub async fn load_cache(&mut self) -> anyhow::Result<()> {
        let ts = SystemTime::now(); // dst-ok: init-time log timing, not on DST state path

        {
            let mut cache = self.cache.borrow_mut();
            cache.clear_index();
            cache.cache_general()?;
        }

        self.cache.borrow_mut().cache_all().await?;

        // Snapshot before iterating: `get_or_init_own_order_book` re-enters `self.cache.borrow_mut()`.
        let own_book_entries: Vec<(InstrumentId, OwnBookOrder)> = {
            let mut cache = self.cache.borrow_mut();
            cache.build_index();
            let _ = cache.check_integrity();

            if self.config.manage_own_order_books {
                cache
                    .orders(None, None, None, None, None)
                    .into_iter()
                    .filter(|o| !o.is_closed() && should_handle_own_book_order(o))
                    .map(|o| (o.instrument_id(), o.to_own_book_order()))
                    .collect()
            } else {
                Vec::new()
            }
        };

        for (instrument_id, own_order) in own_book_entries {
            let mut own_book = self.get_or_init_own_order_book(&instrument_id);
            own_book.add(own_order);
        }

        self.set_position_id_counts();

        log::info!(
            "Loaded cache in {}ms",
            SystemTime::now() // dst-ok: init-time log timing, not on DST state path
                .duration_since(ts)
                .map_err(|e| anyhow::anyhow!("Failed to calculate duration: {e}"))?
                .as_millis()
        );

        Ok(())
    }

    /// Flushes the database to persist all cached data.
    pub fn flush_db(&self) {
        self.cache.borrow_mut().flush_db();
    }

    /// Reconciles an execution report.
    pub fn reconcile_execution_report(&mut self, report: &ExecutionReport) {
        if !matches!(report, ExecutionReport::MassStatus(_)) {
            self.report_count += 1;
        }

        match report {
            ExecutionReport::Order(order_report) => {
                self.reconcile_order_status_report(order_report);
            }
            ExecutionReport::Fill(fill_report) => {
                self.reconcile_fill_report(fill_report);
            }
            ExecutionReport::OrderWithFills(order_report, fills) => {
                self.reconcile_order_with_fills(order_report, fills);
            }
            ExecutionReport::Position(position_report) => {
                self.reconcile_position_report(position_report);
            }
            ExecutionReport::MassStatus(mass_status) => {
                self.reconcile_execution_mass_status(mass_status);
            }
        }
    }

    /// Reconciles an order status report received at runtime.
    ///
    /// Handles order status transitions by generating appropriate events when the venue
    /// reports a different status than our local state. Supports all order states including
    /// fills with inferred fill generation when instruments are available.
    ///
    /// When the order is not found in cache, creates an external order from the report.
    /// This handles exchange-generated orders (liquidation, ADL, settlement) that were
    /// not submitted locally.
    pub fn reconcile_order_status_report(&mut self, report: &OrderStatusReport) {
        self.handle_order_status_report(report, false);
    }

    fn handle_order_status_report(&mut self, report: &OrderStatusReport, is_snapshot: bool) {
        msgbus::publish_any(
            MessagingSwitchboard::reconciliation_raw_order_status_report_topic(),
            report,
        );

        let cache = self.cache.borrow();

        let order = report
            .client_order_id
            .and_then(|id| cache.order(&id).map(|o| o.clone()))
            .or_else(|| {
                cache
                    .client_order_id(&report.venue_order_id)
                    .and_then(|cid| cache.order(cid).map(|o| o.clone()))
            });

        let instrument = cache.instrument(&report.instrument_id).cloned();

        drop(cache);

        if let Some(order) = order {
            let ts_now = self.clock.borrow().timestamp_ns();

            let events = if is_snapshot {
                generate_reconciliation_order_snapshot_events(
                    &order,
                    report,
                    instrument.as_ref(),
                    ts_now,
                )
            } else {
                generate_reconciliation_order_events(&order, report, instrument.as_ref(), ts_now)
            };

            for event in &events {
                self.handle_event(event);
            }
        } else {
            self.create_external_order(report, instrument.as_ref());
        }
    }

    fn create_external_order(
        &mut self,
        report: &OrderStatusReport,
        instrument: Option<&InstrumentAny>,
    ) {
        let Some(instrument) = instrument else {
            log::warn!(
                "Cannot create external order for venue_order_id={}: instrument {} not found",
                report.venue_order_id,
                report.instrument_id
            );
            return;
        };

        let Some(order) = self.materialize_external_order_from_status(report) else {
            return;
        };

        let ts_now = self.clock.borrow().timestamp_ns();
        let events = generate_external_order_status_events(
            &order,
            report,
            &report.account_id,
            instrument,
            ts_now,
        );

        for event in &events {
            self.handle_event(event);
        }
    }

    /// Builds and registers an external order from an [`OrderStatusReport`] without
    /// emitting status events. Returns the registered order.
    fn materialize_external_order_from_status(
        &mut self,
        report: &OrderStatusReport,
    ) -> Option<OrderAny> {
        let strategy_id = self.resolve_external_strategy(&report.instrument_id);
        if self.should_filter_unclaimed_external_order(strategy_id) {
            self.filtered_unclaimed_external_order_count += 1;

            if self.filtered_unclaimed_external_order_count == 1 {
                let external_order_id = report
                    .client_order_id
                    .map_or_else(|| report.venue_order_id.to_string(), |id| id.to_string());
                log::info!(
                    "Filtering unclaimed external orders; first filtered order {} ({}) for {}",
                    external_order_id,
                    report.venue_order_id,
                    report.instrument_id,
                );
            } else {
                let external_order_id = report
                    .client_order_id
                    .map_or_else(|| report.venue_order_id.to_string(), |id| id.to_string());
                log::debug!(
                    "Filtered unclaimed external order {} ({}) for {}",
                    external_order_id,
                    report.venue_order_id,
                    report.instrument_id,
                );
            }

            return None;
        }

        self.materialize_external_order_from_status_with_strategy(report, strategy_id)
    }

    fn materialize_external_order_from_status_with_strategy(
        &self,
        report: &OrderStatusReport,
        strategy_id: StrategyId,
    ) -> Option<OrderAny> {
        let client_order_id = report
            .client_order_id
            .unwrap_or_else(|| ClientOrderId::from(report.venue_order_id.as_str()));

        let trader_id = get_message_bus().borrow().trader_id;
        let ts_now = self.clock.borrow().timestamp_ns();
        let Some(order_side) = report.order_side else {
            log::error!(
                "Skipping external order {} ({}) for {}: order side is not specified",
                client_order_id,
                report.venue_order_id,
                report.instrument_id,
            );
            return None;
        };

        let initialized = match OrderInitialized::new_checked(
            trader_id,
            strategy_id,
            report.instrument_id,
            client_order_id,
            order_side,
            report.order_type,
            report.quantity,
            report.time_in_force,
            report.post_only,
            report.reduce_only,
            false, // quote_quantity
            true,  // reconciliation
            UUID4::new(),
            ts_now,
            ts_now,
            report.price,
            report.activation_price,
            report.trigger_price,
            report.trigger_type,
            report.limit_offset,
            report.trailing_offset,
            report.trailing_offset_type,
            report.expire_time,
            report.display_qty,
            None, // emulation_trigger
            None, // trigger_instrument_id
            report.contingency_type,
            report.order_list_id,
            report.linked_order_ids.clone(),
            report.parent_order_id,
            None, // exec_algorithm_id
            None, // exec_algorithm_params
            None, // exec_spawn_id
            None, // tags
        ) {
            Ok(initialized) => initialized,
            Err(e) => {
                log::error!("Failed to create external order from report: {e}");
                return None;
            }
        };

        self.materialize_external_order(
            initialized,
            client_order_id,
            report.venue_order_id,
            report.instrument_id,
            strategy_id,
            ts_now,
            Some(report.order_status),
            self.source_client_id_for_account(report.account_id, &report.instrument_id),
        )
    }

    /// Builds and registers an external order from a [`FillReport`] when no matching
    /// order exists in cache. The order is created with `OrderType::Market` and a
    /// quantity equal to the fill's `last_qty`, so the fill consumes the entire
    /// order on application.
    ///
    /// This handles venue-initiated fills (most commonly Hyperliquid liquidations)
    /// where the venue does not surface a user-level order on its order channel.
    fn materialize_external_order_from_fill(&mut self, report: &FillReport) -> Option<OrderAny> {
        let strategy_id = self.resolve_external_strategy(&report.instrument_id);
        if self.should_filter_unclaimed_external_order(strategy_id) {
            self.filtered_unclaimed_external_order_count += 1;

            let external_order_id = report
                .client_order_id
                .map_or_else(|| report.venue_order_id.to_string(), |id| id.to_string());

            if self.filtered_unclaimed_external_order_count == 1 {
                log::info!(
                    "Filtering unclaimed external orders; first filtered fill {} ({}) for {}",
                    external_order_id,
                    report.venue_order_id,
                    report.instrument_id,
                );
            } else {
                log::debug!(
                    "Filtered unclaimed external fill {} ({}) for {}",
                    external_order_id,
                    report.venue_order_id,
                    report.instrument_id,
                );
            }

            return None;
        }

        let client_order_id = report
            .client_order_id
            .unwrap_or_else(|| ClientOrderId::from(report.venue_order_id.as_str()));

        let trader_id = get_message_bus().borrow().trader_id;
        let ts_now = self.clock.borrow().timestamp_ns();

        let initialized = OrderInitialized::new(
            trader_id,
            strategy_id,
            report.instrument_id,
            client_order_id,
            report.order_side,
            OrderType::Market,
            report.last_qty,
            TimeInForce::Ioc,
            false, // post_only
            true,  // reduce_only: venue-initiated closes always reduce
            false, // quote_quantity
            true,  // reconciliation
            UUID4::new(),
            ts_now,
            ts_now,
            None, // price
            None, // activation_price
            None, // trigger_price
            None, // trigger_type
            None, // limit_offset
            None, // trailing_offset
            None,
            None, // expire_time
            None, // display_qty
            None, // emulation_trigger
            None, // trigger_instrument_id
            None,
            None, // order_list_id
            None, // linked_order_ids
            None, // parent_order_id
            None, // exec_algorithm_id
            None, // exec_algorithm_params
            None, // exec_spawn_id
            None, // tags
        );

        self.materialize_external_order(
            initialized,
            client_order_id,
            report.venue_order_id,
            report.instrument_id,
            strategy_id,
            ts_now,
            None,
            self.source_client_id_for_account(report.account_id, &report.instrument_id),
        )
    }

    fn resolve_external_strategy(&self, instrument_id: &InstrumentId) -> StrategyId {
        self.cache
            .borrow()
            .external_order_claim(instrument_id)
            .unwrap_or_else(StrategyId::external)
    }

    fn should_filter_unclaimed_external_order(&self, strategy_id: StrategyId) -> bool {
        self.config.filter_unclaimed_external_orders && strategy_id.is_external()
    }

    /// Adds an external order to the cache and registers it for adapter routing.
    /// Returns the registered order on success.
    #[allow(
        clippy::too_many_arguments,
        reason = "external order materialization threads several ids and a timestamp"
    )]
    fn materialize_external_order(
        &self,
        initialized: OrderInitialized,
        client_order_id: ClientOrderId,
        venue_order_id: VenueOrderId,
        instrument_id: InstrumentId,
        strategy_id: StrategyId,
        ts_now: UnixNanos,
        order_status: Option<OrderStatus>,
        source_client_id: Option<ClientId>,
    ) -> Option<OrderAny> {
        let initialized = OrderEventAny::Initialized(initialized);
        let order = match OrderAny::from_events(vec![initialized.clone()]) {
            Ok(order) => order,
            Err(e) => {
                log::error!("Failed to create external order from report: {e}");
                return None;
            }
        };

        {
            let mut cache = self.cache.borrow_mut();
            if let Err(e) = cache.add_venue_order_id(&client_order_id, &venue_order_id, false) {
                log::warn!("Failed to claim venue order ID for external order: {e}");
                return None;
            }

            if let Err(e) = cache.add_order(order.clone(), None, source_client_id, false) {
                log::error!("Failed to add external order to cache: {e}");
                return None;
            }
        }

        self.publish_order_event(&initialized);

        match order_status {
            Some(status) => log::info!(
                "Created external order {client_order_id} ({venue_order_id}) for {instrument_id} [{status}]",
            ),
            None => log::info!(
                "Created external order {client_order_id} ({venue_order_id}) for {instrument_id}",
            ),
        }

        self.register_external_order(
            client_order_id,
            venue_order_id,
            instrument_id,
            strategy_id,
            ts_now,
        );

        Some(order)
    }

    /// Resolves the execution client origin for a live-stream report by matching
    /// the report account against registered clients. A unique match stamps the
    /// materialized order's client origin; no match or an ambiguous match keeps
    /// the order origin-free.
    fn source_client_id_for_account(
        &self,
        account_id: AccountId,
        instrument_id: &InstrumentId,
    ) -> Option<ClientId> {
        let mut matches = self
            .clients
            .values()
            .filter(|adapter| {
                adapter.account_id == account_id && adapter.handles_order_venue(instrument_id.venue)
            })
            .map(|adapter| adapter.client_id);

        let first = matches.next()?;

        matches.next().is_none().then_some(first)
    }

    /// Reconciles a fill report received at runtime.
    ///
    /// Finds the associated order, validates the fill, and generates an `OrderFilled` event
    /// if the fill is not a duplicate and won't cause an overfill. When the order is not
    /// in cache, an external order is bootstrapped from the fill so that venue-initiated
    /// closures (e.g. Hyperliquid liquidations) that arrive without a companion order
    /// status report still update the local position.
    pub fn reconcile_fill_report(&mut self, report: &FillReport) {
        msgbus::publish_any(
            MessagingSwitchboard::reconciliation_raw_fill_report_topic(),
            report,
        );

        if report.last_qty.is_zero() {
            log::warn!("Skipping zero-quantity fill report: {report}");
            return;
        }

        let cache = self.cache.borrow();

        let order = report
            .client_order_id
            .and_then(|id| cache.order(&id).map(|o| o.clone()))
            .or_else(|| {
                cache
                    .client_order_id(&report.venue_order_id)
                    .and_then(|cid| cache.order(cid).map(|o| o.clone()))
            });

        let instrument = cache.instrument(&report.instrument_id).cloned();

        drop(cache);

        let Some(instrument) = instrument else {
            log::debug!(
                "Cannot reconcile fill report for venue_order_id={}: instrument {} not found",
                report.venue_order_id,
                report.instrument_id
            );
            return;
        };

        let order = match order {
            Some(order) => order,
            None => {
                let Some(order) = self.materialize_external_order_from_fill(report) else {
                    return;
                };
                let ts_now = self.clock.borrow().timestamp_ns();
                let accepted = OrderAccepted::new(
                    order.trader_id(),
                    order.strategy_id(),
                    order.instrument_id(),
                    order.client_order_id(),
                    report.venue_order_id,
                    report.account_id,
                    UUID4::new(),
                    report.ts_event,
                    ts_now,
                    true, // reconciliation
                );
                self.handle_event(&OrderEventAny::Accepted(accepted));
                self.cache
                    .borrow()
                    .order(&order.client_order_id())
                    .map(|o| o.clone())
                    .unwrap_or(order)
            }
        };

        let ts_now = self.clock.borrow().timestamp_ns();

        if let Some(event) = reconcile_fill(
            &order,
            report,
            &instrument,
            ts_now,
            self.config.allow_overfills,
        ) {
            self.handle_event(&event);
        }
    }

    /// Reconciles an [`OrderStatusReport`] paired with companion [`FillReport`]s
    /// for the same venue event.
    ///
    /// Real fills supplied by the adapter are applied first so their `trade_id` and
    /// `commission` are preserved; any residual quantity not covered by the fills is
    /// then synthesized as an inferred fill from the status report's `avg_px`.
    /// Adapters use this to emit ADL / liquidation / settlement events without
    /// losing real fill metadata.
    pub fn reconcile_order_with_fills(&mut self, report: &OrderStatusReport, fills: &[FillReport]) {
        msgbus::publish_any(
            MessagingSwitchboard::reconciliation_raw_order_status_report_topic(),
            report,
        );

        let fill_report_topic = MessagingSwitchboard::reconciliation_raw_fill_report_topic();
        for fill in fills {
            msgbus::publish_any(fill_report_topic, fill);
        }

        let cache = self.cache.borrow();
        let order = report
            .client_order_id
            .and_then(|id| cache.order(&id).map(|o| o.clone()))
            .or_else(|| {
                cache
                    .client_order_id(&report.venue_order_id)
                    .and_then(|cid| cache.order(cid).map(|o| o.clone()))
            });
        let instrument = cache.instrument(&report.instrument_id).cloned();
        drop(cache);

        let Some(instrument) = instrument else {
            log::debug!(
                "Cannot reconcile bundled report for venue_order_id={}: instrument {} not found",
                report.venue_order_id,
                report.instrument_id,
            );

            if fills.is_empty()
                && let Some(order) = order
            {
                let ts_now = self.clock.borrow().timestamp_ns();
                let events =
                    generate_reconciliation_order_snapshot_events(&order, report, None, ts_now);

                for event in &events {
                    self.handle_event(event);
                }
            }
            return;
        };

        // Bootstrap the external order with only OrderAccepted; defer fill events to
        // the per-fill loop so real fill metadata is preserved.
        let mut order = match order {
            Some(order) => {
                let ts_now = self.clock.borrow().timestamp_ns();
                let events = generate_reconciliation_order_pre_fill_events(&order, report, ts_now);
                for event in &events {
                    self.handle_event(event);
                }
                self.cache
                    .borrow()
                    .order(&order.client_order_id())
                    .map(|o| o.clone())
                    .unwrap_or(order)
            }
            None => {
                let Some(order) = self.materialize_external_order_from_status(report) else {
                    return;
                };
                let ts_now = self.clock.borrow().timestamp_ns();
                let accepted = OrderAccepted::new(
                    order.trader_id(),
                    order.strategy_id(),
                    order.instrument_id(),
                    order.client_order_id(),
                    report.venue_order_id,
                    report.account_id,
                    UUID4::new(),
                    report.ts_accepted,
                    ts_now,
                    true, // reconciliation
                );
                self.handle_event(&OrderEventAny::Accepted(accepted));
                self.cache
                    .borrow()
                    .order(&order.client_order_id())
                    .map(|o| o.clone())
                    .unwrap_or(order)
            }
        };

        let client_order_id = order.client_order_id();

        for fill in fills {
            let ts_now = self.clock.borrow().timestamp_ns();

            if let Some(event) = reconcile_fill(
                &order,
                fill,
                &instrument,
                ts_now,
                self.config.allow_overfills,
            ) {
                self.handle_event(&event);
            }

            // Refresh order after fill to keep filled_qty accurate for the next iteration.
            if let Some(refreshed) = self
                .cache
                .borrow()
                .order(&client_order_id)
                .map(|o| o.clone())
            {
                order = refreshed;
            }
        }

        let ts_now = self.clock.borrow().timestamp_ns();
        let events = generate_reconciliation_order_snapshot_events(
            &order,
            report,
            Some(&instrument),
            ts_now,
        );

        for event in &events {
            self.handle_event(event);
        }
    }

    /// Reconciles a position status report received at runtime.
    ///
    /// Compares the venue-reported position with cached positions and logs any discrepancies.
    /// Handles both hedging (with `venue_position_id`) and netting (without) modes.
    pub fn reconcile_position_report(&mut self, report: &PositionStatusReport) {
        msgbus::publish_any(
            MessagingSwitchboard::reconciliation_raw_position_status_report_topic(),
            report,
        );

        let cache = self.cache.borrow();

        let size_precision = cache
            .instrument(&report.instrument_id)
            .map(InstrumentAny::size_precision);

        if report.venue_position_id.is_some() {
            self.reconcile_position_report_hedging(report, &cache);
        } else {
            self.reconcile_position_report_netting(report, &cache, size_precision);
        }
    }

    fn reconcile_position_report_hedging(&self, report: &PositionStatusReport, cache: &Cache) {
        let venue_position_id = report.venue_position_id.as_ref().unwrap();

        log::debug!(
            "Reconciling HEDGE position for {}, venue_position_id={}",
            report.instrument_id,
            venue_position_id
        );

        let Some(position) = cache.position(venue_position_id) else {
            if report.signed_decimal_qty == Decimal::ZERO {
                return;
            }

            log::error!("Cannot reconcile position: {venue_position_id} not found in cache");
            return;
        };

        let cached_signed_qty = match position.side {
            PositionSide::Long => position.quantity.as_decimal(),
            PositionSide::Short => -position.quantity.as_decimal(),
            _ => Decimal::ZERO,
        };
        let venue_signed_qty = report.signed_decimal_qty;

        if cached_signed_qty != venue_signed_qty {
            log::error!(
                "Position mismatch for {} {}: cached={}, venue={}",
                report.instrument_id,
                venue_position_id,
                cached_signed_qty,
                venue_signed_qty
            );
        }
    }

    fn reconcile_position_report_netting(
        &self,
        report: &PositionStatusReport,
        cache: &Cache,
        size_precision: Option<u8>,
    ) {
        log::debug!("Reconciling NET position for {}", report.instrument_id);

        let positions_open = Self::netting_positions_open_for_report(cache, report);

        let position_refs = positions_open
            .iter()
            .map(|position| &**position)
            .collect::<Vec<_>>();

        if let Some(message) =
            Self::netting_split_position_ownership_message(report, &position_refs)
        {
            log::warn!("{message}");
        }

        // Sum up cached position quantities using domain types to avoid f64 precision loss
        let cached_signed_qty: Decimal = positions_open
            .iter()
            .map(|position| Self::position_signed_decimal_qty(position))
            .sum();

        log::debug!(
            "Position report: venue_signed_qty={}, cached_signed_qty={}",
            report.signed_decimal_qty,
            cached_signed_qty
        );

        let _ = check_position_reconciliation(report, cached_signed_qty, size_precision);
    }

    fn netting_positions_open_for_report<'a>(
        cache: &'a Cache,
        report: &PositionStatusReport,
    ) -> Vec<PositionRef<'a>> {
        cache.positions_open(
            None,
            Some(&report.instrument_id),
            None,
            Some(&report.account_id),
            None,
        )
    }

    fn netting_split_position_ownership_message(
        report: &PositionStatusReport,
        positions_open: &[&Position],
    ) -> Option<String> {
        let mut strategy_ids = positions_open
            .iter()
            .map(|position| position.strategy_id.to_string())
            .collect::<Vec<_>>();
        strategy_ids.sort();
        strategy_ids.dedup();

        if strategy_ids.len() <= 1 {
            return None;
        }

        let position_details = Self::position_details(positions_open.iter().copied());

        Some(format!(
            "NETTING reconciliation found split ownership for account_id={}, instrument_id={}: \
             strategies=[{}], positions=[{}]",
            report.account_id,
            report.instrument_id,
            strategy_ids.join(", "),
            position_details
        ))
    }

    /// Reconciles an execution mass status report.
    ///
    /// Processes all order reports, fill reports, and position reports contained
    /// in the mass status. Order reports are paired with their companion fills so
    /// real trade IDs and commissions are applied before any residual inferred fill.
    /// Filled-quantity decreases generate fill voids even when no companion fills are present.
    /// Order snapshots are skipped when cached fill activity is initialized at or after
    /// collection starts (`mass_status.ts_init`); companion trades still reconcile.
    pub fn reconcile_execution_mass_status(&mut self, mass_status: &ExecutionMassStatus) {
        self.report_count += 1;

        log::info!(
            "Reconciling mass status for client={}, account={}, venue={}",
            mass_status.client_id,
            mass_status.account_id,
            mass_status.venue
        );

        let order_reports = mass_status.order_reports();
        let fill_reports = mass_status.fill_reports();
        let mut paired_venue_ids = AHashSet::new();

        for order_report in order_reports.values() {
            if self.is_order_snapshot_stale(order_report, mass_status.ts_init) {
                msgbus::publish_any(
                    MessagingSwitchboard::reconciliation_raw_order_status_report_topic(),
                    order_report,
                );

                log::debug!(
                    "Skipping snapshot for {} after concurrent fill activity",
                    order_report.venue_order_id,
                );
                continue;
            }

            if let Some(fills) = fill_reports.get(&order_report.venue_order_id)
                && !fills.is_empty()
            {
                self.reconcile_order_with_fills(order_report, fills);
                paired_venue_ids.insert(order_report.venue_order_id);
            } else {
                self.handle_order_status_report(order_report, true);
            }
        }

        for fill_reports in fill_reports.values() {
            for fill_report in fill_reports {
                if paired_venue_ids.contains(&fill_report.venue_order_id) {
                    continue;
                }

                self.reconcile_fill_report(fill_report);
            }
        }

        for position_reports in mass_status.position_reports().values() {
            for position_report in position_reports {
                self.reconcile_position_report(position_report);
            }
        }

        log::info!(
            "Mass status reconciliation complete: {} orders, {} fills, {} positions",
            mass_status.order_reports().len(),
            mass_status
                .fill_reports()
                .values()
                .map(Vec::len)
                .sum::<usize>(),
            mass_status
                .position_reports()
                .values()
                .map(Vec::len)
                .sum::<usize>()
        );
    }

    fn is_order_snapshot_stale(&self, report: &OrderStatusReport, ts_snapshot: UnixNanos) -> bool {
        let cache = self.cache.borrow();

        let order = report
            .client_order_id
            .and_then(|id| cache.order(&id))
            .or_else(|| {
                cache
                    .client_order_id(&report.venue_order_id)
                    .and_then(|id| cache.order(id))
            });

        order.is_some_and(|order| {
            order.events().into_iter().any(|event| match event {
                OrderEventAny::Filled(fill) => fill.ts_init >= ts_snapshot,
                OrderEventAny::FillVoided(void) => void.ts_init >= ts_snapshot,
                _ => false,
            })
        })
    }

    /// Executes a trading command by routing it to the appropriate execution client.
    pub fn execute(&self, command: TradingCommand) {
        self.execute_command(command);
    }

    /// Processes an order event, updating internal state and routing as needed.
    pub fn process(&mut self, event: &OrderEventAny) {
        self.handle_event(event);
    }

    /// Projects a reconciled fill onto its order without applying position or portfolio economics.
    pub fn project_reconciliation_fill(&mut self, fill: &OrderFilled) {
        self.handle_event_with_position_application(&OrderEventAny::Filled(fill.clone()), false);
    }

    /// Starts the execution engine and all registered execution clients.
    pub fn start(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.start() {
                log::error!("{e}");
            }
        }

        self.start_snapshot_timer();
        self.start_purge_timers();

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

    /// Stops the execution engine and all registered execution clients.
    ///
    /// Adapters are expected to be idempotent on repeated `stop()` calls
    /// (e.g. via an internal `is_stopped` guard); the backtest teardown
    /// sequence calls `stop()` more than once per run.
    pub fn stop(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.stop() {
                log::error!("{e}");
            }
        }

        self.stop_snapshot_timer();
        self.stop_purge_timers();

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

    /// Stops all registered execution clients without stopping the engine itself.
    pub fn stop_clients(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.stop() {
                log::error!("{e}");
            }
        }
    }

    /// Resets the execution engine and all registered execution clients to initial state.
    ///
    /// Cancels engine-owned timers (snapshot, purge) but leaves timers owned by
    /// other components on the shared clock untouched.
    pub fn reset(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.reset() {
                log::error!("{e}");
            }
        }

        self.cache.borrow_mut().reset();
        self.pos_id_generator.reset();
        self.orders_dispatched.borrow_mut().clear();

        self.stop_snapshot_timer();
        self.stop_purge_timers();

        self.command_count.set(0);
        self.event_count = 0;
        self.report_count = 0;
        self.filtered_unclaimed_external_order_count = 0;
        log::info!("Reset");
    }

    /// Disposes of the execution engine, releasing resources from all clients and timers.
    ///
    /// Cancels engine-owned timers (snapshot, purge) but leaves timers owned by
    /// other components on the shared clock untouched.
    pub fn dispose(&mut self) {
        for client in self.get_clients_mut() {
            if let Err(e) = client.dispose() {
                log::error!("{e}");
            }
        }

        self.stop_snapshot_timer();
        self.stop_purge_timers();

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

    fn execute_command(&self, command: TradingCommand) {
        self.command_count.set(self.command_count.get() + 1);

        if self.config.debug {
            log::debug!("{RECV}{CMD} {command}");
        }

        match self.validate_submission(&command) {
            SubmissionValidationResult::Valid => {}
            SubmissionValidationResult::StaleOrder {
                client_order_id,
                status,
            } => {
                log::warn!(
                    "Skipping stale submit command for {client_order_id} in status {status}"
                );
                return;
            }
            SubmissionValidationResult::Dispatched { client_order_id } => {
                log::warn!(
                    "Skipping duplicate submit command for {client_order_id} already dispatched to an execution client"
                );
                return;
            }
            SubmissionValidationResult::Deny(reason) => {
                self.deny_submission(&command, &reason);
                return;
            }
        }

        if let Some(cid) = command.client_id()
            && self.external_clients.contains(&cid)
        {
            let topic = format!("commands.trading.{cid}");
            msgbus::publish_any(topic.into(), &command);

            // The external client now owns the submitted orders, exactly like a registered client
            match &command {
                TradingCommand::SubmitOrder(cmd) => {
                    self.orders_dispatched
                        .borrow_mut()
                        .insert(cmd.client_order_id);
                }
                TradingCommand::SubmitOrderList(cmd) => {
                    self.orders_dispatched
                        .borrow_mut()
                        .extend(cmd.order_list.client_order_ids.iter().copied());
                }
                _ => {}
            }

            if self.config.debug {
                log::debug!("Skipping execution command for external client {cid}: {command}");
            }
            return;
        }

        let client = if let Some(adapter) = self.find_client_for_command(&command) {
            adapter.client.as_ref()
        } else {
            let routing_context = Self::routing_context_for_command(&command);

            log::error!(
                "No execution client found for command: client_id={:?}, {routing_context}, command={command}",
                command.client_id(),
            );

            let reason = OrderDeniedReason::NoExecutionClient {
                client_id: command.client_id(),
                routing_context,
            }
            .to_string();

            match command {
                TradingCommand::SubmitOrder(cmd) => {
                    let order = self
                        .cache
                        .borrow()
                        .order(&cmd.client_order_id)
                        .map(|o| o.clone());

                    if let Some(order) = order {
                        self.deny_order(&order, &reason);
                    }
                }
                TradingCommand::SubmitOrderList(cmd) => {
                    let orders: Vec<OrderAny> = self
                        .cache
                        .borrow()
                        .orders_for_ids(&cmd.order_list.client_order_ids, &cmd);

                    for order in &orders {
                        self.deny_order(order, &reason);
                    }
                }
                _ => {}
            }

            return;
        };

        match command {
            TradingCommand::SubmitOrder(cmd) => self.handle_submit_order(client, cmd),
            TradingCommand::SubmitOrderList(cmd) => self.handle_submit_order_list(client, cmd),
            TradingCommand::ModifyOrder(cmd) => self.handle_modify_order(client, cmd),
            TradingCommand::ModifyOrders(cmd) => self.handle_batch_modify_orders(client, cmd),
            TradingCommand::CancelOrder(cmd) => self.handle_cancel_order(client, cmd),
            TradingCommand::CancelOrders(cmd) => self.handle_batch_cancel_orders(client, cmd),
            TradingCommand::CancelAllOrders(cmd) => self.handle_cancel_all_orders(client, &cmd),
            TradingCommand::QueryOrder(cmd) => self.handle_query_order(client, cmd),
            TradingCommand::QueryAccount(cmd) => self.handle_query_account(client, cmd),
        }
    }

    fn validate_submission(&self, command: &TradingCommand) -> SubmissionValidationResult {
        match command {
            TradingCommand::SubmitOrder(cmd) => {
                let client_order_id = cmd.client_order_id;
                let cache = self.cache.borrow();

                if let Some(order) = cache.order(&client_order_id)
                    && !Self::has_submittable_status(&order)
                {
                    return SubmissionValidationResult::StaleOrder {
                        client_order_id,
                        status: order.status(),
                    };
                }

                if self.is_dispatched(client_order_id) {
                    return SubmissionValidationResult::Dispatched { client_order_id };
                }

                SubmissionValidationResult::Valid
            }
            TradingCommand::SubmitOrderList(cmd) => {
                let cache = self.cache.borrow();
                let client_order_ids = &cmd.order_list.client_order_ids;
                let has_ineligible_order = client_order_ids.iter().any(|client_order_id| {
                    self.is_dispatched(*client_order_id)
                        || cache
                            .order(client_order_id)
                            .is_some_and(|order| !Self::has_submittable_status(&order))
                });

                if !has_ineligible_order {
                    return SubmissionValidationResult::Valid;
                }

                SubmissionValidationResult::Deny(OrderDeniedReason::OrderListDenied {
                    order_list_id: cmd.order_list.id,
                })
            }
            _ => SubmissionValidationResult::Valid,
        }
    }

    fn has_submittable_status(order: &OrderAny) -> bool {
        matches!(
            order.status(),
            OrderStatus::Initialized | OrderStatus::Released
        )
    }

    // The dispatch record is checked independently of the cache because the command is dispatched
    // to an external client without the engine caching its orders (see the type-level documentation).
    fn is_dispatched(&self, client_order_id: ClientOrderId) -> bool {
        self.orders_dispatched.borrow().contains(&client_order_id)
    }

    // A cached order accepts a submit command only while it has a submittable status and has not
    // already been dispatched to an execution client.
    fn is_eligible_for_submission(&self, order: &OrderAny) -> bool {
        Self::has_submittable_status(order) && !self.is_dispatched(order.client_order_id())
    }

    fn deny_submission(&self, command: &TradingCommand, reason: &OrderDeniedReason) {
        let TradingCommand::SubmitOrderList(cmd) = command else {
            return;
        };

        let cache = self.cache.borrow();
        let mut orders: Vec<OrderAny> = cmd
            .order_list
            .client_order_ids
            .iter()
            .filter_map(|client_order_id| cache.order_owned(client_order_id))
            .collect();
        drop(cache);

        for client_order_id in &cmd.order_list.client_order_ids {
            if orders
                .iter()
                .any(|order| order.client_order_id() == *client_order_id)
            {
                continue;
            }

            let Some(order_init) = cmd
                .order_inits
                .iter()
                .find(|init| init.client_order_id == *client_order_id)
            else {
                continue;
            };

            if let Some(order) = self.add_order_from_init(order_init, cmd.position_id, cmd) {
                orders.push(order);
            }
        }

        let mut eligible_orders = Vec::with_capacity(orders.len());

        for order in &orders {
            if self.is_eligible_for_submission(order) {
                eligible_orders.push(order);
            } else if Self::has_submittable_status(order) {
                log::warn!(
                    "Preserving {} already dispatched to an execution client, not denying for order list {}",
                    order.client_order_id(),
                    cmd.order_list.id,
                );
            }
        }

        if eligible_orders.is_empty() {
            log::warn!(
                "Skipping stale submit command for order list {}",
                cmd.order_list.id
            );
            return;
        }

        let reason = reason.to_string();
        for order in eligible_orders {
            self.deny_order(order, &reason);
        }
    }

    fn routing_context_for_command(command: &TradingCommand) -> String {
        match command {
            TradingCommand::SubmitOrder(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::SubmitOrderList(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::ModifyOrder(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::ModifyOrders(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::CancelOrder(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::CancelOrders(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::CancelAllOrders(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::QueryOrder(cmd) => format!("venue={}", cmd.instrument_id.venue),
            TradingCommand::QueryAccount(cmd) => {
                let issuer = cmd.account_id.get_issuer();
                format!("account_id={}, issuer={issuer}", cmd.account_id)
            }
        }
    }

    fn find_client_for_command(&self, command: &TradingCommand) -> Option<&ExecutionClientAdapter> {
        if let Some(client_id) = command.client_id()
            && let Some(adapter) = self.clients.get(&client_id)
        {
            return Some(adapter);
        }

        if let Some(account_id) = self.account_id_for_command(command) {
            let issuer = account_id.get_issuer();
            let issuer_client_id = ClientId::from(issuer.as_str());

            if let Some(adapter) = self.clients.get(&issuer_client_id) {
                return Some(adapter);
            }

            if let Some(client_id) = self.routing_map.get(&issuer)
                && let Some(adapter) = self.clients.get(client_id)
            {
                return Some(adapter);
            }
        }

        if let Some(instrument_id) = Self::instrument_id_for_command(command)
            && let Some(client_id) = self.routing_map.get(&instrument_id.venue)
            && let Some(adapter) = self.clients.get(client_id)
        {
            return Some(adapter);
        }

        self.default_client_id.and_then(|id| self.clients.get(&id))
    }

    fn account_id_for_command(&self, command: &TradingCommand) -> Option<AccountId> {
        match command {
            TradingCommand::QueryAccount(cmd) => Some(cmd.account_id),
            TradingCommand::SubmitOrder(cmd) => self
                .cache
                .borrow()
                .order(&cmd.client_order_id)
                .and_then(|order| order.account_id()),
            TradingCommand::ModifyOrder(cmd) => self
                .cache
                .borrow()
                .order(&cmd.client_order_id)
                .and_then(|order| order.account_id()),
            TradingCommand::CancelOrder(cmd) => self
                .cache
                .borrow()
                .order(&cmd.client_order_id)
                .and_then(|order| order.account_id()),
            TradingCommand::SubmitOrderList(_)
            | TradingCommand::ModifyOrders(_)
            | TradingCommand::CancelOrders(_)
            | TradingCommand::CancelAllOrders(_)
            | TradingCommand::QueryOrder(_) => None,
        }
    }

    const fn instrument_id_for_command(command: &TradingCommand) -> Option<InstrumentId> {
        match command {
            TradingCommand::SubmitOrder(cmd) => Some(cmd.instrument_id),
            TradingCommand::SubmitOrderList(cmd) => Some(cmd.instrument_id),
            TradingCommand::ModifyOrder(cmd) => Some(cmd.instrument_id),
            TradingCommand::ModifyOrders(cmd) => Some(cmd.instrument_id),
            TradingCommand::CancelOrder(cmd) => Some(cmd.instrument_id),
            TradingCommand::CancelOrders(cmd) => Some(cmd.instrument_id),
            TradingCommand::CancelAllOrders(cmd) => Some(cmd.instrument_id),
            TradingCommand::QueryOrder(cmd) => Some(cmd.instrument_id),
            TradingCommand::QueryAccount(_) => None,
        }
    }

    fn handle_submit_order(&self, client: &dyn ExecutionClient, cmd: SubmitOrder) {
        let client_order_id = cmd.client_order_id;
        let cached_order = { self.cache.borrow().order_owned(&client_order_id) };

        let (order, added_to_cache) = match cached_order {
            Some(order) => (order, false),
            None => {
                let Some(order) = self.add_order_from_init(&cmd.order_init, cmd.position_id, &cmd)
                else {
                    return;
                };

                (order, true)
            }
        };

        if added_to_cache && self.config.snapshot_orders {
            self.create_order_state_snapshot(&order);
        }

        let order_venue = order.instrument_id().venue;
        let client_venue = client.venue();
        if !client.handles_order_venue(order_venue) {
            let client_id = client.client_id();
            let reason = OrderDeniedReason::ClientVenueMismatch {
                client_id,
                order_venue,
                client_venue,
            }
            .to_string();
            self.deny_order(&order, &reason);
            return;
        }

        if let Some(reason) = self.check_position_id_against_oms(
            cmd.instrument_id,
            cmd.strategy_id,
            cmd.position_id,
            client,
        ) {
            self.deny_order(&order, &reason.to_string());
            return;
        }

        let instrument_id = order.instrument_id();

        if !added_to_cache && self.config.snapshot_orders {
            self.create_order_state_snapshot(&order);
        }

        {
            let cache = self.cache.borrow();
            if cache.instrument(&instrument_id).is_none() {
                log::error!(
                    "Cannot handle submit order: no instrument found for {instrument_id}, {cmd}",
                );
                return;
            }
        }

        let client_id = client.client_id();
        let claim_result = self
            .cache
            .borrow_mut()
            .claim_order_clients(&[(client_order_id, client_id)]);

        if let Err(e) = claim_result {
            self.deny_order(
                &order,
                &OrderDeniedReason::ValidationFailed {
                    detail: format!(
                        "Failed to claim execution client {client_id} for {client_order_id}: {e}"
                    ),
                }
                .to_string(),
            );
            return;
        }

        if self.config.manage_own_order_books && should_handle_own_book_order(&order) {
            let mut own_book = self.get_or_init_own_order_book(&order.instrument_id());
            own_book.add(order.to_own_book_order());
        }

        log_info!("Submit {order}", color = LogColor::Blue);

        if let Err(e) = client.submit_order(cmd) {
            self.deny_order(
                &order,
                &OrderDeniedReason::SubmitFailed {
                    detail: e.to_string(),
                }
                .to_string(),
            );
            return;
        }

        self.orders_dispatched.borrow_mut().insert(client_order_id);
    }

    fn handle_submit_order_list(&self, client: &dyn ExecutionClient, cmd: SubmitOrderList) {
        let mut orders = Vec::with_capacity(cmd.order_list.client_order_ids.len());
        let mut added_client_order_ids = AHashSet::new();

        for client_order_id in &cmd.order_list.client_order_ids {
            let cached_order = { self.cache.borrow().order_owned(client_order_id) };

            if let Some(order) = cached_order {
                orders.push(order);
                continue;
            }

            let Some(order_init) = cmd
                .order_inits
                .iter()
                .find(|init| init.client_order_id == *client_order_id)
            else {
                log::error!(
                    "Cannot handle submit order list: order not found in cache and no initialization event for {client_order_id}, {cmd}"
                );
                continue;
            };

            let Some(order) = self.add_order_from_init(order_init, cmd.position_id, &cmd) else {
                continue;
            };

            added_client_order_ids.insert(order.client_order_id());
            orders.push(order);
        }

        if self.config.snapshot_orders {
            for order in &orders {
                if added_client_order_ids.contains(&order.client_order_id()) {
                    self.create_order_state_snapshot(order);
                }
            }
        }

        if orders.len() != cmd.order_list.client_order_ids.len() {
            let reason = OrderDeniedReason::OrderListIncomplete {
                order_list_id: cmd.order_list.id,
            }
            .to_string();

            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        let order_list_venue = cmd.instrument_id.venue;
        let client_venue = client.venue();
        if !client.handles_order_venue(order_list_venue) {
            let client_id = client.client_id();
            let reason = OrderDeniedReason::ClientVenueMismatch {
                client_id,
                order_venue: order_list_venue,
                client_venue,
            }
            .to_string();

            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        let is_uniform_instrument = orders
            .iter()
            .all(|o| o.instrument_id() == cmd.instrument_id);

        if let Some(position_id) = cmd.position_id
            && !is_uniform_instrument
        {
            let reason = OrderDeniedReason::InvalidPositionId {
                position_id,
                detail: "not valid for a mixed-instrument order list; a position belongs to a single instrument"
                    .to_string(),
            }
            .to_string();

            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        if let Some(reason) = self.check_position_id_against_oms(
            cmd.instrument_id,
            cmd.strategy_id,
            cmd.position_id,
            client,
        ) {
            let reason = reason.to_string();
            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        if self.config.snapshot_orders {
            for order in &orders {
                if !added_client_order_ids.contains(&order.client_order_id()) {
                    self.create_order_state_snapshot(order);
                }
            }
        }

        {
            let cache = self.cache.borrow();
            if cache.instrument(&cmd.instrument_id).is_none() {
                log::error!(
                    "Cannot handle submit order list: no instrument found for {}, {cmd}",
                    cmd.instrument_id,
                );
                return;
            }
        }

        let client_id = client.client_id();
        let claims = orders
            .iter()
            .map(|order| (order.client_order_id(), client_id))
            .collect::<Vec<_>>();
        let claim_result = self.cache.borrow_mut().claim_order_clients(&claims);
        if let Err(e) = claim_result {
            let reason = OrderDeniedReason::ValidationFailed {
                detail: format!(
                    "Failed to claim execution client {client_id} for order list {}: {e}",
                    cmd.order_list.id,
                ),
            }
            .to_string();

            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        if self.config.manage_own_order_books {
            for order in &orders {
                if should_handle_own_book_order(order) {
                    let mut own_book = self.get_or_init_own_order_book(&order.instrument_id());
                    own_book.add(order.to_own_book_order());
                }
            }
        }

        log_info!("Submit {}", cmd.order_list, color = LogColor::Blue);

        if let Err(e) = client.submit_order_list(cmd) {
            log::error!("Error submitting order list to client: {e}");
            let reason = OrderDeniedReason::SubmitFailed {
                detail: e.to_string(),
            }
            .to_string();

            for order in &orders {
                self.deny_order(order, &reason);
            }
            return;
        }

        self.orders_dispatched
            .borrow_mut()
            .extend(orders.iter().map(Order::client_order_id));
    }

    fn add_order_from_init(
        &self,
        order_init: &OrderInitialized,
        position_id: Option<PositionId>,
        context: &dyn Display,
    ) -> Option<OrderAny> {
        let client_order_id = order_init.client_order_id;
        let order = match OrderAny::from_events(vec![OrderEventAny::Initialized(
            order_init.clone(),
        )]) {
            Ok(order) => order,
            Err(e) => {
                log::error!(
                    "Cannot reconstruct order from initialization event for {client_order_id}: {e}, {context}"
                );
                return None;
            }
        };

        if let Err(e) = self
            .cache
            .borrow_mut()
            .add_order(order.clone(), position_id, None, true)
        {
            log::error!(
                "Cannot add reconstructed order to cache for {client_order_id}: {e}, {context}"
            );
            return None;
        }

        Some(order)
    }

    fn handle_modify_order(&self, client: &dyn ExecutionClient, cmd: ModifyOrder) {
        let venue_str = cmd
            .venue_order_id
            .map_or_else(String::new, |venue_order_id| format!(" {venue_order_id}"));

        log_info!(
            "Modify {}{venue_str}",
            cmd.client_order_id,
            color = LogColor::Blue
        );

        if let Err(e) = client.modify_order(cmd) {
            log::error!("Error modifying order: {e}");
        }
    }

    fn handle_batch_modify_orders(&self, client: &dyn ExecutionClient, cmd: BatchModifyOrders) {
        if let Err(e) = client.batch_modify_orders(cmd) {
            log::error!("Error batch modifying orders: {e}");
        }
    }

    fn handle_cancel_order(&self, client: &dyn ExecutionClient, cmd: CancelOrder) {
        let venue_str = cmd
            .venue_order_id
            .map_or_else(String::new, |venue_order_id| format!(" {venue_order_id}"));

        log_info!(
            "Cancel {}{venue_str}",
            cmd.client_order_id,
            color = LogColor::Blue
        );

        if let Err(e) = client.cancel_order(cmd) {
            log::error!("Error canceling order: {e}");
        }
    }

    fn handle_cancel_all_orders(&self, client: &dyn ExecutionClient, command: &CancelAllOrders) {
        let client_id = client.client_id();
        let account_id = client.account_id();
        let algorithm_commands = self.plan_cancel_all_orders(command, client_id, account_id);
        let venue_command = Self::create_cancel_all_child(command, client_id);
        let emulator_command = Self::create_cancel_all_child(command, client_id);
        let side_str = command
            .order_side
            .map_or_else(|| " ".to_string(), |order_side| format!(" {order_side} "));

        log_info!("Cancel all{side_str}orders", color = LogColor::Blue);

        if let Err(e) = client.cancel_all_orders(venue_command) {
            log::error!("Error canceling all orders: {e}");
        }

        msgbus::send_trading_command(
            MessagingSwitchboard::order_emulator_execute(),
            TradingCommand::CancelAllOrders(emulator_command),
        );

        for (exec_algorithm_id, algorithm_command) in algorithm_commands {
            let endpoint = format!("{exec_algorithm_id}.execute");
            msgbus::send_any(
                endpoint.into(),
                &TradingCommand::CancelOrder(algorithm_command),
            );
        }
    }

    fn plan_cancel_all_orders(
        &self,
        command: &CancelAllOrders,
        client_id: ClientId,
        account_id: AccountId,
    ) -> Vec<(ExecAlgorithmId, CancelOrder)> {
        let order_side = command.order_side;
        let candidates: Vec<(OrderAny, bool)> = {
            let cache = self.cache.borrow();
            cache
                .orders_active_local_refs(
                    None,
                    Some(&command.instrument_id),
                    None,
                    None,
                    order_side,
                )
                .into_iter()
                .filter_map(|order| {
                    if order
                        .account_id()
                        .is_some_and(|order_account_id| order_account_id != account_id)
                    {
                        return None;
                    }

                    let cached_client_id = cache.client_id(&order.client_order_id()).copied();
                    let matches_client = match cached_client_id {
                        Some(order_client_id) => order_client_id == client_id,
                        None => command.client_id.is_none(),
                    };

                    if !matches_client {
                        return None;
                    }

                    let is_emulated = order.is_emulated() || order.emulation_trigger().is_some();
                    if !is_emulated && order.exec_algorithm_id().is_none() {
                        return None;
                    }

                    Some((order.cloned(), cached_client_id.is_none()))
                })
                .collect()
        };

        let claims: Vec<_> = candidates
            .iter()
            .filter_map(|(order, needs_claim)| {
                needs_claim.then_some((order.client_order_id(), client_id))
            })
            .collect();
        let claims_succeeded = claims.is_empty()
            || match self.cache.borrow_mut().claim_order_clients(&claims) {
                Ok(()) => true,
                Err(e) => {
                    log::error!(
                        "Cannot scope local cancel-all orders to execution client {client_id}: {e}"
                    );
                    false
                }
            };
        let correlation_id = command.correlation_id.or(Some(command.command_id));
        let mut algorithm_commands = Vec::new();

        for (order, needs_claim) in candidates {
            if needs_claim && !claims_succeeded {
                continue;
            }

            let is_emulated = order.is_emulated() || order.emulation_trigger().is_some();
            if is_emulated {
                continue;
            }

            if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
                let mut child = CancelOrder::new(
                    command.trader_id,
                    Some(client_id),
                    order.strategy_id(),
                    order.instrument_id(),
                    order.client_order_id(),
                    order.venue_order_id(),
                    UUID4::new(),
                    command.ts_init,
                    command.params.clone(),
                    correlation_id,
                );
                child.causation_id = Some(command.command_id);
                algorithm_commands.push((exec_algorithm_id, child));
            }
        }

        algorithm_commands.sort_by_key(|(exec_algorithm_id, command)| {
            (*exec_algorithm_id, command.client_order_id)
        });
        algorithm_commands.dedup_by_key(|(_, command)| command.client_order_id);

        algorithm_commands
    }

    fn create_cancel_all_child(command: &CancelAllOrders, client_id: ClientId) -> CancelAllOrders {
        let mut child = CancelAllOrders::new(
            command.trader_id,
            Some(client_id),
            command.strategy_id,
            command.instrument_id,
            command.order_side,
            UUID4::new(),
            command.ts_init,
            command.params.clone(),
            command.correlation_id.or(Some(command.command_id)),
        );
        child.causation_id = Some(command.command_id);
        child
    }

    fn handle_batch_cancel_orders(&self, client: &dyn ExecutionClient, cmd: BatchCancelOrders) {
        let client_order_ids: Vec<ClientOrderId> = cmd
            .cancels
            .iter()
            .map(|cancel| cancel.client_order_id)
            .collect();

        log_info!(
            "Batch cancel orders {client_order_ids:?}",
            color = LogColor::Blue
        );

        if let Err(e) = client.batch_cancel_orders(cmd) {
            log::error!("Error batch canceling orders: {e}");
        }
    }

    fn handle_query_account(&self, client: &dyn ExecutionClient, cmd: QueryAccount) {
        log_info!("Query {}", cmd.account_id, color = LogColor::Blue);

        if let Err(e) = client.query_account(cmd) {
            log::warn!("Error querying account: {e}");
        }
    }

    fn handle_query_order(&self, client: &dyn ExecutionClient, cmd: QueryOrder) {
        log_info!("Query {}", cmd.client_order_id, color = LogColor::Blue);

        if let Err(e) = client.query_order(cmd) {
            log::warn!("Error querying order: {e}");
        }
    }

    fn create_order_state_snapshot(&self, order: &OrderAny) {
        if self.config.debug {
            log::debug!("Creating order state snapshot for {order}");
        }

        if self.cache.borrow().has_backing()
            && let Err(e) = self.cache.borrow().snapshot_order_state(order)
        {
            log::warn!("Failed to snapshot order state: {e}");
        }
    }

    fn create_position_state_snapshot(&self, position: &Position, open_only: bool) {
        Self::publish_position_state_snapshot(
            &self.clock,
            &self.cache,
            self.config.debug,
            position,
            open_only,
        );
    }

    fn publish_position_state_snapshot(
        clock: &Rc<RefCell<dyn Clock>>,
        cache: &Rc<RefCell<Cache>>,
        debug: bool,
        position: &Position,
        open_only: bool,
    ) {
        if debug {
            log::debug!("Creating position state snapshot for {position}");
        }

        let ts_snapshot = clock.borrow().timestamp_ns();
        let unrealized_pnl = cache.borrow().calculate_unrealized_pnl(position);

        let snapshot = PositionStateSnapshot {
            position: position.clone(),
            unrealized_pnl,
            ts_snapshot,
        };

        let topic = switchboard::get_snapshot_position_topic(position.id);
        msgbus::publish_any(topic, &snapshot);

        let has_backing = cache.borrow().has_backing();
        if has_backing
            && let Err(e) = cache.borrow_mut().snapshot_position_state(
                position,
                ts_snapshot,
                unrealized_pnl,
                Some(open_only),
            )
        {
            log::warn!("Failed to snapshot position state: {e}");
        }
    }

    fn handle_event(&mut self, event: &OrderEventAny) {
        self.handle_event_with_position_application(event, true);
    }

    fn handle_event_with_position_application(
        &mut self,
        event: &OrderEventAny,
        apply_position: bool,
    ) {
        if let OrderEventAny::Filled(fill) = event
            && fill.last_qty.is_zero()
        {
            log::warn!("Skipping zero-quantity fill event: {fill}");
            return;
        }

        self.event_count += 1;

        if self.config.debug {
            log::debug!("{RECV}{EVT} {event}");
        }

        let event_client_order_id = event.client_order_id();
        let cache = self.cache.borrow();
        let client_order_id = if cache.order_exists(&event_client_order_id) {
            event_client_order_id
        } else {
            let is_leg_fill =
                matches!(event, OrderEventAny::Filled(fill) if self.is_leg_fill(fill));
            if !is_leg_fill {
                log::warn!(
                    "Order with {} not found in the cache to apply {}",
                    event.client_order_id(),
                    event
                );
            }

            // Try to find order by venue order ID if available
            let venue_order_id = if let Some(id) = event.venue_order_id() {
                id
            } else {
                log::error!(
                    "Cannot apply event to any order: {} not found in the cache with no VenueOrderId",
                    event.client_order_id()
                );
                return;
            };

            // Look up client order ID from venue order ID
            let client_order_id = if let Some(id) = cache.client_order_id(&venue_order_id) {
                *id
            } else {
                if let OrderEventAny::Filled(fill) = event
                    && is_leg_fill
                {
                    log::info!(
                        "Processing leg fill without corresponding order: {} for instrument {}",
                        fill.client_order_id,
                        fill.instrument_id
                    );
                    drop(cache);
                    self.handle_leg_fill_without_order(fill.clone());
                    return;
                }

                log::error!(
                    "Cannot apply event to any order: {} and {venue_order_id} not found in the cache",
                    event.client_order_id(),
                );
                return;
            };

            // Get order using found client order ID
            if cache.order_exists(&client_order_id) {
                log::info!("Order with {client_order_id} was found in the cache");
                client_order_id
            } else {
                if let OrderEventAny::Filled(fill) = event
                    && is_leg_fill
                {
                    log::info!(
                        "Processing leg fill without corresponding order: {} for instrument {}",
                        fill.client_order_id,
                        fill.instrument_id
                    );
                    drop(cache);
                    self.handle_leg_fill_without_order(fill.clone());
                    return;
                }

                log::error!(
                    "Cannot apply event to any order: {client_order_id} and {venue_order_id} not found in cache",
                );
                return;
            }
        };
        let order_before_fill = if matches!(event, OrderEventAny::Filled(_)) {
            cache.order(&client_order_id).map(|o| o.clone())
        } else {
            None
        };

        drop(cache);

        let event = if event_client_order_id == client_order_id {
            event.clone()
        } else {
            event.clone().with_client_order_id(client_order_id)
        };

        match &event {
            OrderEventAny::Filled(fill) => {
                let Some(order_before_fill) = order_before_fill else {
                    log::error!(
                        "Cannot apply fill: order {} not found in the cache",
                        fill.client_order_id()
                    );
                    return;
                };
                let configured_oms_type = self.determine_oms_type(fill);
                let Some(position_id) =
                    self.determine_position_id(fill, configured_oms_type, Some(&order_before_fill))
                else {
                    return;
                };
                let oms_type = self
                    .cache
                    .borrow()
                    .oms_type(&position_id)
                    .unwrap_or(configured_oms_type);

                let mut fill = fill.clone();
                fill.position_id = Some(position_id);

                let validation = if apply_position {
                    self.validate_fill_for_order(&order_before_fill, &fill)
                } else {
                    self.validate_fill_for_order_projection(&order_before_fill, &fill)
                };

                if validation.is_ok() {
                    if apply_position
                        && !self.validate_fill_for_external_position(
                            &order_before_fill,
                            &fill,
                            oms_type,
                            position_id,
                        )
                    {
                        return;
                    }

                    let event = OrderEventAny::Filled(fill.clone());
                    let Some(order) =
                        self.update_cached_order(client_order_id, &event, apply_position)
                    else {
                        return;
                    };

                    let position_events = if apply_position {
                        self.handle_order_fill(&order, fill, oms_type)
                    } else {
                        Vec::new()
                    };
                    self.publish_order_event(&event);
                    self.publish_position_events(position_events);
                }
            }
            OrderEventAny::FillVoided(voided) => {
                let mut voided = voided.clone();
                let Some(order_before_void) = self
                    .cache
                    .borrow()
                    .order(&client_order_id)
                    .map(|order| order.clone())
                else {
                    log::error!("Cannot apply fill void: order {client_order_id} not found");
                    return;
                };
                let original_fill = order_before_void
                    .events()
                    .into_iter()
                    .find_map(|candidate| match candidate {
                        OrderEventAny::Filled(fill) if fill.trade_id == voided.trade_id => {
                            Some(fill.clone())
                        }
                        _ => None,
                    });

                if voided.position_id.is_none() {
                    voided.position_id = original_fill.as_ref().and_then(|fill| fill.position_id);
                }
                let event = OrderEventAny::FillVoided(voided.clone());

                let mut validated_order = order_before_void.clone();
                match validated_order.apply(event.clone()) {
                    Ok(()) => {}
                    Err(OrderError::DuplicateFillVoid(trade_id)) => {
                        log::warn!(
                            "Duplicate fill void rejected at order level: trade_id={trade_id}"
                        );
                        return;
                    }
                    Err(e) => {
                        log::error!("Cannot apply fill void to order: {e}");
                        return;
                    }
                }

                let corrected_positions = if apply_position
                    && original_fill
                        .as_ref()
                        .is_some_and(|fill| fill.position_id.is_some())
                {
                    match self.prepare_order_fill_void_positions(&order_before_void, &voided) {
                        Ok(positions) => positions,
                        Err(e) => {
                            log::error!("Cannot apply fill void to positions: {e}");
                            return;
                        }
                    }
                } else {
                    Vec::new()
                };

                let mut position_events = Vec::new();

                for CorrectedPosition {
                    position,
                    corrected_qty,
                    absorbed_prior_cycles,
                    closed_cycles_pnl,
                } in corrected_positions
                {
                    if let Err(e) = self.cache.borrow_mut().update_position(&position) {
                        log::error!("Cannot apply fill void to position {}: {e}", position.id);
                        return;
                    }

                    if absorbed_prior_cycles {
                        log::info!(
                            "Settling archived NETTING cycles rebuilt by fill void {} for position {}: realized={closed_cycles_pnl:?}",
                            voided.trade_id,
                            position.id,
                        );

                        self.cache
                            .borrow_mut()
                            .settle_position_snapshots(&position, closed_cycles_pnl);
                    }

                    if self.config.snapshot_positions {
                        self.create_position_state_snapshot(&position, false);
                    }

                    position_events.push(Self::create_fill_void_position_event(
                        &position,
                        &voided,
                        corrected_qty,
                    ));
                }

                if self
                    .update_cached_order(client_order_id, &event, true)
                    .is_none()
                {
                    return;
                }

                if original_fill.is_some() {
                    let portfolio_endpoint = MessagingSwitchboard::portfolio_update_order();
                    msgbus::send_order_event(portfolio_endpoint, event.clone());
                }
                self.publish_order_event(&event);
                self.publish_position_events(position_events);
            }
            _ => {
                if self
                    .update_cached_order(client_order_id, &event, true)
                    .is_some()
                {
                    self.publish_order_event(&event);
                }
            }
        }
    }

    fn handle_leg_fill_without_order(&mut self, mut fill: OrderFilled) {
        let instrument =
            if let Some(instrument) = self.cache.borrow().instrument(&fill.instrument_id) {
                instrument.clone()
            } else {
                log::error!(
                    "Cannot handle leg fill: no instrument found for {}, {fill}",
                    fill.instrument_id,
                );
                return;
            };

        if let Err(e) = self.cache.borrow().try_account(&fill.account_id) {
            log::error!("Cannot handle leg fill: {e}, {fill}");
            return;
        }

        let oms_type = self.determine_oms_type(&fill);
        let position_id = self.determine_leg_fill_position_id(&fill, oms_type);
        fill.position_id = Some(position_id);

        if !self.validate_fill_for_position(position_id, &fill) {
            return;
        }

        let duplicate_position_fill = self.position_contains_trade_id(position_id, fill.trade_id);

        let event = OrderEventAny::Filled(fill.clone());

        if duplicate_position_fill {
            log::warn!(
                "Duplicate leg fill: {} trade_id={} already applied to position {}, skipping",
                fill.client_order_id,
                fill.trade_id,
                position_id
            );
            return;
        }

        let portfolio_endpoint = MessagingSwitchboard::portfolio_update_order();
        msgbus::send_order_event(portfolio_endpoint, event.clone());
        let position_events = self.handle_position_update(&instrument, fill, oms_type);
        self.publish_order_event(&event);
        self.publish_position_events(position_events);
    }

    fn determine_leg_fill_position_id(
        &mut self,
        fill: &OrderFilled,
        oms_type: OmsType,
    ) -> PositionId {
        let cache = self.cache.borrow();
        let cached_position_id = cache.position_id(&fill.client_order_id()).copied();
        drop(cache);

        if let Some(position_id) = cached_position_id {
            if let Some(fill_position_id) = fill.position_id
                && fill_position_id != position_id
            {
                log::warn!(
                    "Incorrect position ID assigned to leg fill: \
                     cached={position_id}, assigned={fill_position_id}; \
                     re-assigning from cache",
                );
            }

            return position_id;
        }

        match oms_type {
            OmsType::Hedging => self
                .orderless_hedging_leg_position_id(fill)
                .or(fill.position_id)
                .unwrap_or_else(|| self.pos_id_generator.generate(fill.strategy_id, false)),
            OmsType::Netting => self.determine_netting_position_id(fill, None),
            _ => self.determine_netting_position_id(fill, None),
        }
    }

    fn orderless_hedging_leg_position_id(&self, fill: &OrderFilled) -> Option<PositionId> {
        if !self.is_leg_fill(fill) {
            return None;
        }

        let cache = self.cache.borrow();
        if cache.order_exists(&fill.client_order_id()) {
            return None;
        }

        let matching_positions: Vec<PositionId> = cache
            .positions_open(
                Some(&fill.instrument_id.venue),
                Some(&fill.instrument_id),
                Some(&fill.strategy_id),
                Some(&fill.account_id),
                None,
            )
            .iter()
            .filter(|position| position.opening_order_id == fill.client_order_id)
            .map(|position| position.id)
            .collect();

        match matching_positions.as_slice() {
            [position_id] => Some(*position_id),
            [] => None,
            _ => {
                log::warn!(
                    "Cannot uniquely correlate HEDGING leg fill {} to an orderless position: \
                     found {} positions with opening_order_id={}",
                    fill.trade_id,
                    matching_positions.len(),
                    fill.client_order_id,
                );
                None
            }
        }
    }

    fn is_leg_fill(&self, fill: &OrderFilled) -> bool {
        if !fill.client_order_id.as_str().contains("-LEG-")
            && !fill.venue_order_id.as_str().contains("-LEG-")
        {
            return false;
        }

        self.cache
            .borrow()
            .instrument(&fill.instrument_id)
            .is_some_and(|instrument| !instrument.is_spread())
    }

    fn determine_oms_type(&self, fill: &OrderFilled) -> OmsType {
        let client_id = self
            .cache
            .borrow()
            .client_id(&fill.client_order_id)
            .copied();

        let client = client_id.and_then(|id| self.get_client(&id)).or_else(|| {
            self.source_client_id_for_account(fill.account_id, &fill.instrument_id)
                .and_then(|id| self.get_client(&id))
        });

        self.resolve_oms_type_for_client(fill.strategy_id, client)
    }

    fn resolve_oms_type_for_client(
        &self,
        strategy_id: StrategyId,
        client: Option<&dyn ExecutionClient>,
    ) -> OmsType {
        if let Some(oms_type) = self.oms_overrides.get(&strategy_id)
            && *oms_type != OmsType::Unspecified
        {
            return *oms_type;
        }

        // Missing or ambiguous ownership retains the origin-free NETTING fallback
        client.map_or(OmsType::Netting, ExecutionClient::oms_type)
    }

    fn check_position_id_against_oms(
        &self,
        instrument_id: InstrumentId,
        strategy_id: StrategyId,
        position_id: Option<PositionId>,
        client: &dyn ExecutionClient,
    ) -> Option<OrderDeniedReason> {
        let position_id = position_id?;

        if self.resolve_oms_type_for_client(strategy_id, Some(client)) != OmsType::Netting {
            return None;
        }

        let expected = format!("{instrument_id}-{strategy_id}");
        if position_id.as_str() == expected {
            return None;
        }

        Some(OrderDeniedReason::InvalidPositionId {
            position_id,
            detail: format!(
                "not valid for NETTING OMS; expected '{expected}' (use HEDGING for custom position IDs)"
            ),
        })
    }

    fn determine_position_id(
        &mut self,
        fill: &OrderFilled,
        oms_type: OmsType,
        order: Option<&OrderAny>,
    ) -> Option<PositionId> {
        let cache = self.cache.borrow();
        let cached_position_id = cache.position_id(&fill.client_order_id()).copied();
        drop(cache);

        if self.config.debug {
            log::debug!(
                "Determining position ID for {}, position_id={:?}",
                fill.client_order_id(),
                cached_position_id,
            );
        }

        if let Some(cached_position_id) = cached_position_id
            && let Some(fill_position_id) = fill.position_id
            && cached_position_id != fill_position_id
        {
            if oms_type == OmsType::Hedging {
                if !self.is_flip_remainder(fill, cached_position_id, fill_position_id) {
                    log::error!(
                        "Cannot apply hedging fill {} for {}: venue position ID {fill_position_id} conflicts with cached position ID {cached_position_id}",
                        fill.trade_id,
                        fill.client_order_id(),
                    );

                    return None;
                }
            } else {
                log::warn!(
                    "Incorrect position ID assigned to fill: \
                     cached={cached_position_id}, assigned={fill_position_id}; \
                     re-assigning from cache",
                );
            }
        }

        if let Some(position_id) = cached_position_id {
            if self.config.debug {
                log::debug!("Assigned {position_id} to {}", fill.client_order_id());
            }

            if !self.validate_fill_for_position(position_id, fill) {
                return None;
            }

            return Some(position_id);
        }

        let position_id = match (oms_type, fill.position_id) {
            (OmsType::Hedging, Some(position_id)) => position_id,
            (OmsType::Hedging, None) => self.determine_hedging_position_id(fill, order),
            (OmsType::Netting, _) => self.determine_netting_position_id(fill, order),
            _ => self.determine_netting_position_id(fill, order),
        };

        if !self.validate_fill_for_position(position_id, fill) {
            return None;
        }

        let order = if let Some(o) = order {
            o.clone()
        } else {
            let cache = self.cache.borrow();
            cache.order(&fill.client_order_id()).map_or_else(
                || {
                    panic!(
                        "Order for {} not found to determine position ID",
                        fill.client_order_id()
                    )
                },
                |o| o.clone(),
            )
        };

        if order.exec_algorithm_id().is_some()
            && let Some(exec_spawn_id) = order.exec_spawn_id()
        {
            let cache = self.cache.borrow();
            let primary = if let Some(p) = cache.order(&exec_spawn_id) {
                p.clone()
            } else {
                log::warn!(
                    "Primary exec spawn order {exec_spawn_id} not found, \
                     skipping position ID propagation"
                );
                return Some(position_id);
            };
            let primary_already_indexed = cache.position_id(&primary.client_order_id()).is_some();
            drop(cache);

            if primary.position_id().is_none() && !primary_already_indexed {
                if let Some(mut primary_mut) = self.cache.borrow_mut().order_mut(&exec_spawn_id) {
                    primary_mut.set_position_id(Some(position_id));
                }
                let _ = self.cache.borrow_mut().add_position_id(
                    &position_id,
                    &primary.instrument_id().venue,
                    &primary.client_order_id(),
                    &primary.strategy_id(),
                );
                log::debug!("Assigned primary order {position_id}");
            }
        }

        Some(position_id)
    }

    /// Returns whether `fill` may be applied to the position assigned to `position_id`.
    ///
    /// Only `instrument_id` is compared. A position's instrument never changes, and a fill
    /// for another instrument would be priced with this position's precision, multiplier,
    /// currencies, and PnL rules.
    ///
    /// `account_id` and `strategy_id` are deliberately NOT compared, because each has a
    /// legitimate mismatch path. Netting position IDs are `{instrument_id}-{strategy_id}`,
    /// so two accounts trading one instrument under one strategy share a position ID.
    /// External order claims can be handed to a successor strategy while the predecessor's
    /// positions stay cached, so a later venue fill can carry the new strategy against them.
    fn validate_fill_for_position(&self, position_id: PositionId, fill: &OrderFilled) -> bool {
        let cache = self.cache.borrow();
        let Some(position) = cache.position_ref(&position_id) else {
            return true;
        };

        if position.instrument_id != fill.instrument_id {
            log::error!(
                "Cannot apply fill {} to position {position_id}: instrument_id mismatch, expected={}, received={}",
                fill.trade_id,
                position.instrument_id,
                fill.instrument_id
            );
            return false;
        }

        true
    }

    fn validate_fill_for_external_position(
        &self,
        order: &OrderAny,
        fill: &OrderFilled,
        oms_type: OmsType,
        position_id: PositionId,
    ) -> bool {
        if oms_type != OmsType::Netting || !order.is_reduce_only() {
            return true;
        }

        let cache = self.cache.borrow();

        let Some(position) = cache.position_ref(&position_id) else {
            return true;
        };

        if position.strategy_id.is_external()
            && position.strategy_id != fill.strategy_id
            && (position.account_id != fill.account_id
                || !position.is_opposite_side(fill.order_side)
                || fill.last_qty > position.quantity)
        {
            log::error!(
                "Cannot apply reduce-only fill {} to external NETTING position {position_id}: \
                 account, side, or quantity does not match the open position",
                fill.trade_id,
            );
            return false;
        }

        true
    }

    /// Returns whether `fill` is a later fill of an order this engine already flipped.
    ///
    /// Flipping under `Hedging` closes the original virtual position with the reversing order
    /// and opens a newly minted virtual position from the same order, moving the order's cache
    /// index onto the new ID. Every later fill of that order still carries the original ID, so
    /// the venue ID and the cached ID disagree for the rest of the order's life.
    ///
    /// The split is recognized from the two positions rather than from the order, because
    /// applying a fill writes the determined ID onto the order and would erase the evidence for
    /// the fill after it. Both halves must still name this order: the cached position was opened
    /// by it, and the position the fill names was closed by it.
    fn is_flip_remainder(
        &self,
        fill: &OrderFilled,
        cached_position_id: PositionId,
        fill_position_id: PositionId,
    ) -> bool {
        if !cached_position_id.is_virtual() || !fill_position_id.is_virtual() {
            return false;
        }

        let cache = self.cache.borrow();
        let client_order_id = fill.client_order_id();

        let opened_by_order = cache
            .position_ref(&cached_position_id)
            .is_some_and(|flipped| flipped.opening_order_id == client_order_id);

        let closed_by_order = cache
            .position_ref(&fill_position_id)
            .is_some_and(|original| {
                original.is_closed() && original.closing_order_id == Some(client_order_id)
            });

        opened_by_order && closed_by_order
    }

    fn determine_hedging_position_id(
        &mut self,
        fill: &OrderFilled,
        order: Option<&OrderAny>,
    ) -> PositionId {
        let cache = self.cache.borrow();

        let cached_order;
        let order: &OrderAny = if let Some(order) = order {
            order
        } else {
            cached_order = cache.order(&fill.client_order_id()).unwrap_or_else(|| {
                panic!(
                    "Order for {} not found to determine position ID",
                    fill.client_order_id()
                )
            });
            &cached_order
        };

        // Check execution spawn orders
        if let Some(spawn_id) = order.exec_spawn_id() {
            let spawn_orders = cache.orders_for_exec_spawn(&spawn_id);
            for spawned_order in spawn_orders {
                if let Some(pos_id) = spawned_order.position_id() {
                    if self.config.debug {
                        log::debug!("Found spawned {} for {}", pos_id, fill.client_order_id());
                    }
                    return pos_id;
                }
            }
        }

        if order.is_reduce_only() {
            let mut candidates = cache
                .positions_open(
                    None,
                    Some(&fill.instrument_id),
                    Some(&fill.strategy_id),
                    Some(&fill.account_id),
                    None,
                )
                .into_iter()
                .filter(|position| position.is_opposite_side(fill.order_side));
            let candidate = candidates.next();

            if let Some(position) = candidate
                && candidates.next().is_none()
                && order.would_reduce_only(position.side, position.quantity)
            {
                if self.config.debug {
                    log::debug!(
                        "Assigned reduce-only fill {} to position {}",
                        fill.client_order_id(),
                        position.id
                    );
                }
                return position.id;
            }
        }

        // Generate new position ID
        let position_id = self.pos_id_generator.generate(fill.strategy_id, false);

        if self.config.debug {
            log::debug!("Generated {} for {}", position_id, fill.client_order_id());
        }
        position_id
    }

    fn determine_netting_position_id(
        &self,
        fill: &OrderFilled,
        order: Option<&OrderAny>,
    ) -> PositionId {
        let position_id = PositionId::new(format!("{}-{}", fill.instrument_id, fill.strategy_id));
        let cache = self.cache.borrow();
        if order.is_none_or(|order| !order.is_reduce_only())
            || cache
                .position_ref(&position_id)
                .is_some_and(|position| position.is_open())
        {
            return position_id;
        }

        let mut candidates = cache
            .positions_open(
                None,
                Some(&fill.instrument_id),
                Some(&StrategyId::external()),
                Some(&fill.account_id),
                None,
            )
            .into_iter()
            .filter(|position| {
                position.is_opposite_side(fill.order_side)
                    && cache.oms_type(&position.id) == Some(OmsType::Netting)
            });

        let candidate = candidates.next();

        if let Some(position) = candidate
            && candidates.next().is_none()
            && fill.last_qty <= position.quantity
        {
            return position.id;
        }

        position_id
    }

    fn validate_fill_for_order(&self, order: &OrderAny, fill: &OrderFilled) -> anyhow::Result<()> {
        if order.is_duplicate_fill(fill) {
            log::warn!(
                "Duplicate fill: {} trade_id={} already applied, skipping",
                order.client_order_id(),
                fill.trade_id
            );
            anyhow::bail!("Duplicate fill");
        }

        if let Some(position_id) = fill.position_id
            && self.position_contains_trade_id(position_id, fill.trade_id)
        {
            log::warn!(
                "Duplicate fill: {} trade_id={} already applied to position {}, skipping",
                order.client_order_id(),
                fill.trade_id,
                position_id
            );
            anyhow::bail!("Duplicate position fill");
        }

        self.check_overfill(order, fill)
    }

    fn validate_fill_for_order_projection(
        &self,
        order: &OrderAny,
        fill: &OrderFilled,
    ) -> anyhow::Result<()> {
        if order.is_duplicate_fill(fill) {
            anyhow::bail!("Duplicate fill");
        }

        self.check_overfill(order, fill)
    }

    fn position_contains_trade_id(&self, position_id: PositionId, trade_id: TradeId) -> bool {
        self.cache
            .borrow()
            .position(&position_id)
            .is_some_and(|position| position.trade_ids.contains(&trade_id))
    }

    fn update_cached_order(
        &self,
        client_order_id: ClientOrderId,
        event: &OrderEventAny,
        send_portfolio_update: bool,
    ) -> Option<OrderAny> {
        let result = { self.cache.borrow_mut().update_order(event) };

        let order = match result {
            Ok(order) => order,
            Err(e) => {
                if matches!(
                    e.downcast_ref::<OrderError>(),
                    Some(OrderError::InvalidStateTransition)
                ) {
                    // A non-fill event that fails to apply to an already-closed order is an
                    // expected venue race (e.g. a place reject then a stream cancel for the same
                    // order), not an anomaly. A dropped fill stays at warn even on a closed order,
                    // since it represents real, possibly lost, execution.
                    let already_closed = self
                        .cache
                        .borrow()
                        .order(&client_order_id)
                        .is_some_and(|o| o.is_closed());

                    if already_closed && !matches!(event, OrderEventAny::Filled(_)) {
                        log::debug!("InvalidStateTrigger: {e}, did not apply {event}");
                    } else {
                        log::warn!("InvalidStateTrigger: {e}, did not apply {event}");
                    }
                    return None;
                }

                if let Some(OrderError::DuplicateFill(trade_id)) = e.downcast_ref::<OrderError>() {
                    log::warn!(
                        "Duplicate fill rejected at order level: trade_id={trade_id}, did not apply {event}"
                    );
                    return None;
                }

                if let Some(OrderError::DuplicateFillVoid(trade_id)) =
                    e.downcast_ref::<OrderError>()
                {
                    log::warn!(
                        "Duplicate fill void rejected at order level: trade_id={trade_id}, did not apply {event}"
                    );
                    return None;
                }

                log::error!("Error applying event: {e}, did not apply {event}");

                if matches!(
                    event,
                    OrderEventAny::Denied(_)
                        | OrderEventAny::Rejected(_)
                        | OrderEventAny::Canceled(_)
                        | OrderEventAny::Expired(_)
                ) {
                    log::warn!(
                        "Terminal event {event} failed to apply to {client_order_id}, forcing cleanup from own book"
                    );
                    self.cache
                        .borrow_mut()
                        .force_remove_from_own_order_book(&client_order_id);
                } else {
                    let order = self
                        .cache
                        .borrow()
                        .order(&client_order_id)
                        .map(|o| o.clone());

                    if let Some(order) = order {
                        let should_update_own_book = {
                            let cache = self.cache.borrow();
                            let own_book = cache.own_order_book(&order.instrument_id());
                            (own_book.is_some() && order.is_closed())
                                || should_handle_own_book_order(&order)
                        };

                        if should_update_own_book {
                            self.cache.borrow_mut().update_own_order_book(&order);
                        }
                    }
                }
                return None;
            }
        };

        // The client's first status transition closes the dispatch window
        if !Self::has_submittable_status(&order) {
            self.orders_dispatched.borrow_mut().remove(&client_order_id);
        }

        if self.config.manage_own_order_books && should_handle_own_book_order(&order) {
            let needs_own_book = {
                self.cache
                    .borrow()
                    .own_order_book(&order.instrument_id())
                    .is_none()
            };

            if needs_own_book {
                self.cache.borrow_mut().update_own_order_book(&order);
            }
        }

        if self.config.debug {
            log::debug!("{SEND}{EVT} {event}");
        }

        if self.config.snapshot_orders {
            self.create_order_state_snapshot(&order);
        }

        if send_portfolio_update {
            self.send_order_update_to_portfolio(event);
        }

        Some(order)
    }

    fn send_order_update_to_portfolio(&self, event: &OrderEventAny) {
        let is_wallet = event.account_id().is_some_and(|account_id| {
            self.cache
                .borrow()
                .account(&account_id)
                .is_some_and(|account| account.account_type() == AccountType::Wallet)
        });
        let send_to_portfolio = match event {
            OrderEventAny::Filled(fill) => self
                .cache
                .borrow()
                .account(&fill.account_id)
                .is_none_or(|account| !account.is_margin_account()),
            OrderEventAny::Accepted(_)
            | OrderEventAny::Canceled(_)
            | OrderEventAny::Expired(_)
            | OrderEventAny::Rejected(_)
            | OrderEventAny::Updated(_) => true,
            OrderEventAny::Submitted(_)
            | OrderEventAny::Triggered(_)
            | OrderEventAny::PendingUpdate(_)
            | OrderEventAny::PendingCancel(_)
            | OrderEventAny::ModifyRejected(_)
            | OrderEventAny::CancelRejected(_)
            | OrderEventAny::FillVoided(_) => is_wallet,
            _ => false,
        };

        if send_to_portfolio {
            let portfolio_endpoint = MessagingSwitchboard::portfolio_update_order();
            msgbus::send_order_event(portfolio_endpoint, event.clone());
        }
    }

    fn publish_order_event(&self, event: &OrderEventAny) {
        let topic = switchboard::get_event_order_topic(event.strategy_id());
        msgbus::publish_order_event(topic, event);

        #[rustfmt::skip]
        let topic = match event {
            OrderEventAny::Submitted(_) => switchboard::get_order_submitted_topic(event.instrument_id()),
            OrderEventAny::Rejected(_) => switchboard::get_order_rejected_topic(event.instrument_id()),
            OrderEventAny::PendingUpdate(_) => switchboard::get_order_pending_update_topic(event.instrument_id()),
            OrderEventAny::PendingCancel(_) => switchboard::get_order_pending_cancel_topic(event.instrument_id()),
            OrderEventAny::ModifyRejected(_) => switchboard::get_order_modify_rejected_topic(event.instrument_id()),
            OrderEventAny::CancelRejected(_) => switchboard::get_order_cancel_rejected_topic(event.instrument_id()),
            OrderEventAny::Canceled(_) => switchboard::get_order_canceled_topic(event.instrument_id()),
            OrderEventAny::FillVoided(_) => switchboard::get_order_fill_voided_topic(event.instrument_id()),
            // Keep Filled out of this generic fanout: handle_order_fill publishes the instrument
            // topic, while leg fills stay on the strategy topic.
            _ => return,
        };

        msgbus::publish_order_event(topic, event);
    }

    fn publish_position_events(&self, events: Vec<PositionEvent>) {
        for event in events {
            let strategy_id = match &event {
                PositionEvent::PositionOpened(event) => event.strategy_id,
                PositionEvent::PositionChanged(event) => event.strategy_id,
                PositionEvent::PositionClosed(event) => event.strategy_id,
                PositionEvent::PositionAdjusted(event) => event.strategy_id,
            };
            let topic = switchboard::get_event_position_topic(strategy_id);
            msgbus::publish_position_event(topic, &event);
        }
    }

    fn check_overfill(&self, order: &OrderAny, fill: &OrderFilled) -> anyhow::Result<()> {
        let potential_overfill = order.calculate_overfill(fill.last_qty);

        if potential_overfill.is_positive() {
            if self.config.allow_overfills {
                log::warn!(
                    "Order overfill detected: {} potential_overfill={}, current_filled={}, last_qty={}, quantity={}",
                    order.client_order_id(),
                    potential_overfill,
                    order.filled_qty(),
                    fill.last_qty,
                    order.quantity()
                );
            } else {
                let msg = format!(
                    "Order overfill rejected: {} potential_overfill={}, current_filled={}, last_qty={}, quantity={}. \
                Set `allow_overfills=true` in ExecutionEngineConfig to allow overfills.",
                    order.client_order_id(),
                    potential_overfill,
                    order.filled_qty(),
                    fill.last_qty,
                    order.quantity()
                );
                anyhow::bail!("{msg}");
            }
        }

        Ok(())
    }

    fn handle_order_fill(
        &mut self,
        order: &OrderAny,
        fill: OrderFilled,
        oms_type: OmsType,
    ) -> Vec<PositionEvent> {
        let instrument =
            if let Some(instrument) = self.cache.borrow().instrument(&fill.instrument_id) {
                instrument.clone()
            } else {
                log::error!(
                    "Cannot handle order fill: no instrument found for {}, {fill}",
                    fill.instrument_id,
                );
                return Vec::new();
            };

        let is_margin_account = {
            let cache = self.cache.borrow();
            let account = match cache.try_account(&fill.account_id) {
                Ok(account) => account,
                Err(e) => {
                    log::error!("Cannot handle order fill: {e}, {fill}");
                    return Vec::new();
                }
            };

            account.is_margin_account()
        };

        // Skip portfolio position updates for combo fills (spread instruments)
        // Combo fills are only used for order management, not portfolio updates
        if !instrument.is_spread() && is_margin_account {
            let portfolio_endpoint = MessagingSwitchboard::portfolio_update_order();
            msgbus::send_order_event(portfolio_endpoint, OrderEventAny::Filled(fill.clone()));
        }

        let (position, position_events) = if instrument.is_spread() {
            (None, Vec::new())
        } else {
            let position_events = self.handle_position_update(&instrument, fill.clone(), oms_type);
            let position_id = fill.position_id.unwrap();
            (
                self.cache
                    .borrow()
                    .position(&position_id)
                    .map(|position| position.clone_without_events()),
                position_events,
            )
        };

        if !position_events.is_empty() {
            self.index_external_position_reduction(order, &fill, oms_type, position.as_ref());
        }

        // Handle contingent orders for both spread and non-spread instruments
        // For spread instruments, contingent orders work without position linkage
        if matches!(order.contingency_type(), Some(ContingencyType::Oto)) {
            // For non-spread instruments, link to position if available
            if !instrument.is_spread()
                && let Some(ref pos) = position
                && pos.is_open()
            {
                let position_id = pos.id;

                for client_order_id in order.linked_order_ids().unwrap_or_default() {
                    // Take a scoped write borrow on the contingent's cell. The borrow drops at
                    // the end of `and_then` so the subsequent `add_position_id` on the cache is
                    // free to take `&mut Cache`.
                    let link = self.cache.borrow_mut().order_mut(client_order_id).and_then(
                        |mut contingent_order| {
                            if contingent_order.position_id().is_none() {
                                contingent_order.set_position_id(Some(position_id));
                                Some((
                                    contingent_order.instrument_id().venue,
                                    contingent_order.client_order_id(),
                                    contingent_order.strategy_id(),
                                ))
                            } else {
                                None
                            }
                        },
                    );

                    if let Some((venue, contingent_id, strategy_id)) = link
                        && let Err(e) = self.cache.borrow_mut().add_position_id(
                            &position_id,
                            &venue,
                            &contingent_id,
                            &strategy_id,
                        )
                    {
                        log::error!("Failed to add position ID: {e}");
                    }
                }
            }
            // For spread instruments, contingent orders can still be triggered
            // but without position linkage (since no position is created for spreads)
        }

        let topic = switchboard::get_order_filled_topic(fill.instrument_id);
        let event = OrderEventAny::Filled(fill);
        msgbus::publish_order_event(topic, &event);

        position_events
    }

    fn index_external_position_reduction(
        &self,
        order: &OrderAny,
        fill: &OrderFilled,
        oms_type: OmsType,
        position: Option<&Position>,
    ) {
        if oms_type == OmsType::Netting
            && order.is_reduce_only()
            && let Some(position) = position
            && position.strategy_id.is_external()
            && let Err(e) = self.cache.borrow_mut().add_position_id(
                &position.id,
                &fill.instrument_id.venue,
                &fill.client_order_id,
                &position.strategy_id,
            )
        {
            log::error!("Failed to index external position for reducing order: {e}");
        }
    }

    fn prepare_order_fill_void_positions(
        &self,
        order: &OrderAny,
        event: &OrderFillVoided,
    ) -> anyhow::Result<Vec<CorrectedPosition>> {
        let source_event_id = order
            .events()
            .into_iter()
            .find_map(|order_event| match order_event {
                OrderEventAny::Filled(fill) if fill.trade_id == event.trade_id => {
                    Some(fill.event_id)
                }
                _ => None,
            })
            .ok_or_else(|| anyhow::anyhow!("fill {} is not in order history", event.trade_id))?;

        let positions: Vec<Position> = {
            let cache = self.cache.borrow();

            let strategy_id = event
                .position_id
                .and_then(|id| cache.position_ref(&id))
                .filter(|position| {
                    order.is_reduce_only()
                        && position.strategy_id.is_external()
                        && cache.oms_type(&position.id) == Some(OmsType::Netting)
                })
                .map_or(event.strategy_id, |position| position.strategy_id);

            cache
                .positions(
                    None,
                    Some(&event.instrument_id),
                    Some(&strategy_id),
                    Some(&event.account_id),
                    None,
                )
                .into_iter()
                .map(|position| position.cloned())
                .collect()
        };
        let mut fragments = Vec::new();

        for position in &positions {
            for replay_event in &position.replay_events {
                let PositionReplayEvent::Filled(fill) = replay_event else {
                    continue;
                };

                if fill.client_order_id != event.client_order_id || fill.trade_id != event.trade_id
                {
                    continue;
                }
                let split_rank = if fill.event_id == source_event_id {
                    0
                } else if fill.causation_id == Some(source_event_id) {
                    1
                } else {
                    continue;
                };
                fragments.push((position.id, split_rank, fill.last_qty, fill.commission));
            }
        }
        anyhow::ensure!(
            !fragments.is_empty(),
            "no position fragments found for fill {}",
            event.trade_id
        );
        fragments.sort_by_key(|(_, split_rank, _, _)| *split_rank);

        let mut allocations = IndexMap::<PositionId, (Quantity, Option<Money>)>::new();
        let mut remaining_qty = event.voided_qty;
        for (position_id, _, quantity, _) in fragments.iter().rev() {
            if remaining_qty.is_zero() {
                break;
            }
            let removed = remaining_qty.min(*quantity);
            allocations
                .entry(*position_id)
                .and_modify(|allocation| allocation.0 = allocation.0 + removed)
                .or_insert((removed, None));
            remaining_qty = remaining_qty - removed;
        }
        anyhow::ensure!(
            remaining_qty.is_zero(),
            "position fragments do not cover voided quantity for fill {}",
            event.trade_id
        );

        if let Some(mut remaining_commission) = event.commission_voided {
            for (position_id, _, _, commission) in fragments.iter().rev() {
                if remaining_commission.is_zero() {
                    break;
                }
                let Some(commission) = commission else {
                    continue;
                };
                anyhow::ensure!(
                    commission.currency == remaining_commission.currency,
                    "position commission currency differs for fill {}",
                    event.trade_id
                );
                let magnitude = remaining_commission.abs().min(commission.abs());

                let removed = if remaining_commission.is_negative() {
                    -magnitude
                } else {
                    magnitude
                };

                allocations
                    .entry(*position_id)
                    .and_modify(|allocation| {
                        allocation.1 = Some(
                            allocation
                                .1
                                .map_or(removed, |commission| commission + removed),
                        );
                    })
                    .or_insert((Quantity::zero(event.voided_qty.precision), Some(removed)));
                remaining_commission = remaining_commission - removed;
            }
            anyhow::ensure!(
                remaining_commission.is_zero(),
                "position fragments do not cover voided commission for fill {}",
                event.trade_id
            );
        }

        let mut corrected_positions = Vec::new();

        for (position_id, (voided_qty, commission_voided)) in allocations {
            if voided_qty.is_zero() {
                anyhow::bail!(
                    "commission-only position correction requires authoritative reconciliation for fill {}",
                    event.trade_id
                );
            }
            let mut position = self
                .cache
                .borrow()
                .position_owned(&position_id)
                .ok_or_else(|| anyhow::anyhow!("position {position_id} is not cached"))?;
            let previous = position
                .fill_voids
                .iter()
                .rev()
                .find(|record| {
                    record.event.client_order_id == event.client_order_id
                        && record.event.trade_id == event.trade_id
                })
                .map(|record| (record.voided_qty, record.commission_voided));
            if previous == Some((voided_qty, commission_voided)) {
                continue;
            }
            let corrected_qty = previous.map_or(voided_qty, |(prior_qty, _)| {
                voided_qty.saturating_sub(prior_qty)
            });

            // `events` holds the fills since the position was last flat, because `apply_fill`
            // clears it when reopening from flat. A NETTING flip splits one fill across the
            // closing and reopening cycles under the same trade, so compare quantities rather
            // than presence: the correction reaches an earlier cycle once it exceeds what the
            // current cycle originally held. Earlier corrections have already shrunk the
            // fragments in `events` while `voided_qty` stays cumulative, so add back what this
            // position already voided. Read this before `apply_fill_void`, whose rebuild
            // re-derives `events` and can move that boundary.
            let previously_voided = previous
                .map_or(Quantity::zero(position.size_precision), |(prior_qty, _)| {
                    prior_qty
                });
            let current_cycle_qty = position
                .events
                .iter()
                .filter(|fill| {
                    fill.client_order_id == event.client_order_id && fill.trade_id == event.trade_id
                })
                .fold(previously_voided, |total, fill| total + fill.last_qty);
            let absorbed_prior_cycles = voided_qty > current_cycle_qty;
            let closed_cycles_pnl =
                position.apply_fill_void(event.clone(), voided_qty, commission_voided)?;
            corrected_positions.push(CorrectedPosition {
                position,
                corrected_qty,
                absorbed_prior_cycles,
                closed_cycles_pnl,
            });
        }
        Ok(corrected_positions)
    }

    fn create_fill_void_position_event(
        position: &Position,
        fill_voided: &OrderFillVoided,
        corrected_qty: Quantity,
    ) -> PositionEvent {
        let event_id = UUID4::new();
        let ts_init = fill_voided.ts_init;

        if position.is_closed() {
            PositionEvent::PositionClosed(PositionClosed {
                trader_id: position.trader_id,
                strategy_id: position.strategy_id,
                instrument_id: position.instrument_id,
                position_id: position.id,
                account_id: position.account_id,
                opening_order_id: position.opening_order_id,
                closing_order_id: position.closing_order_id,
                entry: position.entry,
                side: position.side,
                signed_qty: position.signed_qty,
                quantity: position.quantity,
                peak_quantity: position.peak_qty,
                last_qty: corrected_qty,
                last_px: fill_voided.last_px,
                currency: position.quote_currency,
                avg_px_open: position.avg_px_open,
                avg_px_close: position.avg_px_close,
                realized_return: position.realized_return,
                realized_pnl: position.realized_pnl,
                unrealized_pnl: Money::zero(position.quote_currency),
                duration: position.duration_ns,
                event_id,
                ts_opened: position.ts_opened,
                ts_closed: position.ts_closed,
                ts_event: fill_voided.ts_event,
                ts_init,
            })
        } else {
            PositionEvent::PositionChanged(PositionChanged {
                trader_id: position.trader_id,
                strategy_id: position.strategy_id,
                instrument_id: position.instrument_id,
                position_id: position.id,
                account_id: position.account_id,
                opening_order_id: position.opening_order_id,
                entry: position.entry,
                side: position.side,
                signed_qty: position.signed_qty,
                quantity: position.quantity,
                peak_quantity: position.peak_qty,
                last_qty: corrected_qty,
                last_px: fill_voided.last_px,
                currency: position.quote_currency,
                avg_px_open: position.avg_px_open,
                avg_px_close: position.avg_px_close,
                realized_return: position.realized_return,
                realized_pnl: position.realized_pnl,
                unrealized_pnl: Money::zero(position.quote_currency),
                event_id,
                ts_opened: position.ts_opened,
                ts_event: fill_voided.ts_event,
                ts_init,
            })
        }
    }

    /// Handle position creation or update for a fill.
    ///
    /// This function mirrors the Python `_handle_position_update` method.
    fn handle_position_update(
        &mut self,
        instrument: &InstrumentAny,
        fill: OrderFilled,
        oms_type: OmsType,
    ) -> Vec<PositionEvent> {
        enum Action {
            Open,
            Reopen,
            Flip(Box<Position>),
            Update,
        }

        let position_id = if let Some(position_id) = fill.position_id {
            position_id
        } else {
            log::error!("Cannot handle position update: no position ID found for fill {fill}");
            return Vec::new();
        };

        let action = {
            let cache = self.cache.borrow();

            match cache.position(&position_id) {
                None => Action::Open,
                Some(position) if position.is_closed() => Action::Reopen,
                Some(position) if self.will_flip_position(&position, &fill) => {
                    Action::Flip(Box::new(position.clone()))
                }
                Some(_) => Action::Update,
            }
        };

        match action {
            Action::Open => {
                if self.reject_reduce_only_position_open(&fill, oms_type) {
                    return Vec::new();
                }

                self.open_position(instrument, None, true, fill, oms_type)
                    .unwrap_or_default()
            }
            Action::Reopen => {
                if self.reject_reduce_only_position_open(&fill, oms_type) {
                    return Vec::new();
                }

                self.open_position(instrument, Some(position_id), true, fill, oms_type)
                    .unwrap_or_default()
            }
            Action::Flip(mut position) => {
                self.flip_position(instrument, &mut position, &fill, oms_type)
            }
            Action::Update => self
                .update_position_from_fill(position_id, &fill)
                .into_iter()
                .collect(),
        }
    }

    fn reject_reduce_only_position_open(&self, fill: &OrderFilled, oms_type: OmsType) -> bool {
        let cache = self.cache.borrow();
        let Some(order) = cache.order_owned(&fill.client_order_id) else {
            return false;
        };

        if !order.is_reduce_only() {
            return false;
        }

        let positions_open = cache.positions_open(
            None,
            Some(&fill.instrument_id),
            None,
            Some(&fill.account_id),
            None,
        );
        let position_id = fill
            .position_id
            .map_or_else(|| "None".to_string(), |position_id| position_id.to_string());
        let matching_position_details = Self::position_details(
            positions_open
                .iter()
                .filter(|position| position.is_opposite_side(fill.order_side))
                .map(|position| &**position),
        );
        let open_position_details =
            Self::position_details(positions_open.iter().map(|position| &**position));

        log::error!(
            "Cannot open {oms_type} position {position_id} from reduce-only fill {} for {}; \
             matching_reduce_positions=[{}], open_positions=[{}]",
            fill.trade_id,
            fill.instrument_id,
            matching_position_details,
            open_position_details
        );

        true
    }

    #[allow(
        clippy::needless_pass_by_value,
        reason = "takes the opening fill by value to seed the new position"
    )]
    fn open_position(
        &self,
        instrument: &InstrumentAny,
        prior_position_id: Option<PositionId>,
        archive_prior: bool,
        fill: OrderFilled,
        oms_type: OmsType,
    ) -> anyhow::Result<Vec<PositionEvent>> {
        if let Some(position_id) = prior_position_id {
            let prior_snapshot = {
                let cache = self.cache.borrow();
                let position = cache
                    .position(&position_id)
                    .ok_or_else(|| anyhow::anyhow!("position {position_id} is not cached"))?;

                if archive_prior && position.has_replay_trade_id(fill.trade_id) {
                    log::warn!(
                        "Ignoring duplicate fill {} for closed position {}; no position reopened (side={:?}, qty={}, px={})",
                        fill.trade_id,
                        position.id,
                        fill.order_side,
                        fill.last_qty,
                        fill.last_px
                    );
                    return Ok(Vec::new());
                }

                archive_prior.then(|| position.clone_for_snapshot())
            };

            if let Some(position) = prior_snapshot {
                self.reopen_position(&position, oms_type)?;
            }
        }

        let position = Position::new(instrument, fill.clone());
        let is_orderless_leg = self.is_leg_fill(&fill)
            && !self.cache.borrow().order_exists(&position.opening_order_id);
        if let Some(position_id) = prior_position_id {
            debug_assert_eq!(position_id, position.id);

            self.cache.borrow_mut().replace_position(
                &position,
                oms_type,
                !is_orderless_leg,
                self.config.carry_replay_events_on_reopen,
            )?;
        } else if is_orderless_leg {
            self.cache
                .borrow_mut()
                .add_position_without_order(&position, oms_type)?;
        } else {
            self.cache.borrow_mut().add_position(&position, oms_type)?;
        }

        let (position, snapshot_position) = {
            let cache = self.cache.borrow();
            let position = cache
                .position(&fill.position_id.expect("Opening fill has no position ID"))
                .expect("Opened position is no longer cached");
            (
                position.clone_for_snapshot(),
                self.config.snapshot_positions.then(|| position.clone()),
            )
        };

        if let Some(snapshot_position) = snapshot_position {
            self.create_position_state_snapshot(&snapshot_position, true);
        }

        let ts_init = self.clock.borrow().timestamp_ns();
        let event = PositionOpened::create(&position, &fill, UUID4::new(), ts_init);

        Ok(vec![PositionEvent::PositionOpened(event)])
    }

    fn reopen_position(&self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> {
        if oms_type == OmsType::Netting {
            if position.is_open() {
                anyhow::bail!(
                    "Cannot reopen position {} (oms_type=NETTING): reopening is only valid for closed positions in NETTING mode",
                    position.id
                );
            }
        } else {
            // HEDGING mode
            log::warn!(
                "Received fill for closed position {} in HEDGING mode; archiving closed cycle and creating new position",
                position.id
            );
        }

        // Snapshot the closed cycle before its ID is reused: `add_position` replaces the
        // cached position, and realized PnL totals read closed cycles from the snapshots
        self.snapshot_position(position)?;

        Ok(())
    }

    /// Archives the closed `position` and anchors the frame when an anchorer is installed.
    ///
    /// An installed anchorer needs the encoded frame, so this takes the eager path. Without one
    /// the cache defers the encode unless a backing database has to persist the frame.
    fn snapshot_position(&self, position: &Position) -> anyhow::Result<()> {
        let mut cache = self.cache.borrow_mut();

        let Some(anchorer) = &self.snapshot_anchorer else {
            return cache.snapshot_position(position);
        };

        let snapshot_ref = cache.snapshot_position_encoded(position)?;
        drop(cache);

        if let Err(e) = anchorer(snapshot_ref) {
            log::warn!("Failed to record cache snapshot anchor: {e}");
        }

        Ok(())
    }

    fn update_position(
        &self,
        position: &mut Position,
        fill: &OrderFilled,
    ) -> Option<PositionEvent> {
        // Apply the fill to the position
        position.apply(fill);

        // Check if position is closed after applying the fill
        let is_closed = position.is_closed();

        // Update position in cache - this should handle the closed state tracking
        if let Err(e) = self.cache.borrow_mut().update_position(position) {
            log::error!("Failed to update position: {e:?}");
            return None;
        }

        // Verify cache state after update
        let cache = self.cache.borrow();

        drop(cache);

        // Create position state snapshot if enabled
        if self.config.snapshot_positions {
            self.create_position_state_snapshot(position, false);
        }

        let ts_init = self.clock.borrow().timestamp_ns();

        if is_closed {
            let event = PositionClosed::create(position, fill, UUID4::new(), ts_init);
            Some(PositionEvent::PositionClosed(event))
        } else {
            let event = PositionChanged::create(position, fill, UUID4::new(), ts_init);
            Some(PositionEvent::PositionChanged(event))
        }
    }

    fn update_position_from_fill(
        &self,
        position_id: PositionId,
        fill: &OrderFilled,
    ) -> Option<PositionEvent> {
        let position = match self
            .cache
            .borrow_mut()
            .update_position_from_fill(position_id, fill)
        {
            Ok(position) => position,
            Err(e) => {
                log::error!("Failed to update position: {e:?}");
                return None;
            }
        };

        if self.config.snapshot_positions {
            let position = self
                .cache
                .borrow()
                .position_owned(&position_id)
                .expect("Updated position is no longer cached");
            self.create_position_state_snapshot(&position, false);
        }

        let ts_init = self.clock.borrow().timestamp_ns();

        if position.is_closed() {
            let event = PositionClosed::create(&position, fill, UUID4::new(), ts_init);
            Some(PositionEvent::PositionClosed(event))
        } else {
            let event = PositionChanged::create(&position, fill, UUID4::new(), ts_init);
            Some(PositionEvent::PositionChanged(event))
        }
    }

    fn will_flip_position(&self, position: &Position, fill: &OrderFilled) -> bool {
        position.is_opposite_side(fill.order_side) && (fill.last_qty > position.quantity)
    }

    fn position_signed_decimal_qty(position: &Position) -> Decimal {
        match position.side {
            PositionSide::Long => position.quantity.as_decimal(),
            PositionSide::Short => -position.quantity.as_decimal(),
            _ => Decimal::ZERO,
        }
    }

    fn position_details<'a>(positions: impl IntoIterator<Item = &'a Position>) -> String {
        positions
            .into_iter()
            .map(|position| {
                format!(
                    "{} strategy_id={} signed_qty={}",
                    position.id,
                    position.strategy_id,
                    Self::position_signed_decimal_qty(position)
                )
            })
            .collect::<Vec<_>>()
            .join(", ")
    }

    fn flip_position(
        &mut self,
        instrument: &InstrumentAny,
        position: &mut Position,
        fill: &OrderFilled,
        oms_type: OmsType,
    ) -> Vec<PositionEvent> {
        let mut position_events = Vec::new();

        if fill.commission.is_none() {
            log::warn!(
                "Commission is not available for position flip, splitting with no commission"
            );
        }

        let position_id_flip = if oms_type == OmsType::Hedging
            && let Some(position_id) = fill.position_id
            && position_id.is_virtual()
        {
            // Generate new position ID for flipped virtual position (Hedging OMS only)
            Some(self.pos_id_generator.generate(fill.strategy_id, true))
        } else {
            // Default: use the same position ID as the fill (Python behavior)
            fill.position_id
        };

        let (fill_split1, fill_split2) = fill
            .split_for_position_flip(position.quantity, position_id_flip, UUID4::new())
            .expect("Invalid position flip split");

        if let Some(position_event) = self.update_position(position, &fill_split1) {
            position_events.push(position_event);
        }

        // Snapshot closed position before reusing ID (NETTING mode)
        if oms_type == OmsType::Netting
            && let Err(e) = self.snapshot_position(position)
        {
            log::warn!("Failed to snapshot position during flip: {e:?}");
        }

        if oms_type == OmsType::Hedging
            && let Some(position_id) = fill.position_id
            && position_id.is_virtual()
        {
            log::warn!("Closing position {fill_split1}");
            log::warn!("Flipping position {fill_split2}");
        }

        // Open flipped position
        let prior_position_id =
            (fill_split2.position_id == Some(position.id)).then_some(position.id);

        match self.open_position(instrument, prior_position_id, false, fill_split2, oms_type) {
            Ok(opened_events) => position_events.extend(opened_events),
            Err(e) => log::error!("Failed to open flipped position: {e:?}"),
        }

        position_events
    }

    /// Sets the internal position ID generator counts based on existing cached positions.
    pub fn set_position_id_counts(&mut self) {
        let cache = self.cache.borrow();
        let positions = cache.positions(None, None, None, None, None);

        // Count positions per instrument_id using a HashMap
        let mut counts: HashMap<StrategyId, usize> = HashMap::new();

        for position in positions {
            *counts.entry(position.strategy_id).or_insert(0) += 1;
        }

        self.pos_id_generator.reset();

        for (strategy_id, count) in counts {
            self.pos_id_generator.set_count(count, strategy_id);
            log::info!("Set PositionId count for {strategy_id} to {count}");
        }
    }

    fn deny_order(&self, order: &OrderAny, reason: &str) {
        let denied = OrderDenied::new(
            order.trader_id(),
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            reason.into(),
            UUID4::new(),
            self.clock.borrow().timestamp_ns(),
            self.clock.borrow().timestamp_ns(),
        );

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

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

        if self.config.snapshot_orders {
            self.create_order_state_snapshot(&order);
        }
    }

    fn get_or_init_own_order_book(&self, instrument_id: &InstrumentId) -> RefMut<'_, OwnOrderBook> {
        let mut cache = self.cache.borrow_mut();
        if cache.own_order_book_mut(instrument_id).is_none() {
            let own_book = OwnOrderBook::new(*instrument_id);
            cache.add_own_order_book(own_book).unwrap();
        }

        RefMut::map(cache, |c| c.own_order_book_mut(instrument_id).unwrap())
    }
}

enum SubmissionValidationResult {
    Valid,
    StaleOrder {
        client_order_id: ClientOrderId,
        status: OrderStatus,
    },
    Dispatched {
        client_order_id: ClientOrderId,
    },
    Deny(OrderDeniedReason),
}

#[cfg(test)]
mod tests {
    use nautilus_common::{
        clock::VirtualClock,
        msgbus::{MessageBus, set_message_bus},
    };
    use nautilus_model::{
        enums::{LiquiditySide, OrderSide, OrderType, PositionSide},
        events::{OrderFillVoided, order::spec::OrderFilledSpec},
        identifiers::{AccountId, ClientOrderId, TradeId, VenueOrderId},
        instruments::{InstrumentAny, stubs::audusd_sim},
        orders::builder::OrderTestBuilder,
        position::PositionFillVoid,
        types::Price,
    };
    use rstest::*;

    use super::*;

    #[rstest]
    fn netting_positions_open_for_report_scopes_positions_by_account() {
        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
        let account1_id = AccountId::from("SIM-001");
        let account2_id = AccountId::from("SIM-002");
        let position1 = position_for_account(
            &instrument,
            account1_id,
            StrategyId::from("S-001"),
            PositionId::from("P-ACC-1"),
            OrderSide::Buy,
            Quantity::from(1_000),
        );
        let position2 = position_for_account(
            &instrument,
            account2_id,
            StrategyId::from("S-002"),
            PositionId::from("P-ACC-2"),
            OrderSide::Buy,
            Quantity::from(2_000),
        );
        let mut cache = Cache::default();
        cache.add_position(&position1, OmsType::Netting).unwrap();
        cache.add_position(&position2, OmsType::Netting).unwrap();

        let report = PositionStatusReport::new(
            account1_id,
            instrument.id(),
            PositionSide::Long,
            Quantity::from(1_000),
            UnixNanos::from(1_000_000),
            UnixNanos::from(1_000_000),
            None,
            None,
            None,
        );

        let positions_open = ExecutionEngine::netting_positions_open_for_report(&cache, &report);
        let signed_qty: Decimal = positions_open
            .iter()
            .map(|position| ExecutionEngine::position_signed_decimal_qty(position))
            .sum();

        assert_eq!(positions_open.len(), 1);
        assert_eq!(positions_open[0].id, position1.id);
        assert_eq!(signed_qty, Decimal::from(1_000));
    }

    #[rstest]
    fn netting_split_position_ownership_message_reports_only_split_ownership() {
        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
        let account_id = AccountId::from("SIM-001");
        let external_position = position_for_account(
            &instrument,
            account_id,
            StrategyId::from("EXTERNAL"),
            PositionId::from("P-EXTERNAL"),
            OrderSide::Buy,
            Quantity::from(1_000),
        );
        let strategy_position = position_for_account(
            &instrument,
            account_id,
            StrategyId::from("S-001"),
            PositionId::from("P-STRATEGY"),
            OrderSide::Buy,
            Quantity::from(500),
        );
        let same_strategy_position = position_for_account(
            &instrument,
            account_id,
            StrategyId::from("EXTERNAL"),
            PositionId::from("P-EXTERNAL-2"),
            OrderSide::Buy,
            Quantity::from(250),
        );
        let report = PositionStatusReport::new(
            account_id,
            instrument.id(),
            PositionSide::Long,
            Quantity::from(1_500),
            UnixNanos::from(1_000_000),
            UnixNanos::from(1_000_000),
            None,
            None,
            None,
        );

        let message = ExecutionEngine::netting_split_position_ownership_message(
            &report,
            &[&external_position, &strategy_position],
        )
        .expect("split ownership should produce a warning message");

        assert!(message.contains("account_id=SIM-001"));
        assert!(message.contains(&format!("instrument_id={}", instrument.id())));
        assert!(message.contains("EXTERNAL"));
        assert!(message.contains("S-001"));
        assert!(message.contains("P-EXTERNAL"));
        assert!(message.contains("P-STRATEGY"));
        assert!(message.contains("signed_qty=1000"));
        assert!(message.contains("signed_qty=500"));
        assert!(
            ExecutionEngine::netting_split_position_ownership_message(
                &report,
                &[&external_position, &same_strategy_position],
            )
            .is_none()
        );
    }

    #[rstest]
    fn materialize_external_order_rejects_venue_id_owned_by_another_order() {
        let cache = Rc::new(RefCell::new(Cache::default()));
        let venue_order_id = VenueOrderId::from("V-SHARED");
        let owner_id = ClientOrderId::from("O-OWNER");
        cache
            .borrow_mut()
            .add_venue_order_id(&owner_id, &venue_order_id, false)
            .unwrap();
        let engine = ExecutionEngine::new(
            Rc::new(RefCell::new(VirtualClock::new())),
            Rc::clone(&cache),
            None,
        );
        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
        let claimant_id = ClientOrderId::from("O-CLAIMANT");
        let order = OrderTestBuilder::new(OrderType::Limit)
            .instrument_id(instrument.id())
            .client_order_id(claimant_id)
            .side(OrderSide::Buy)
            .quantity(Quantity::from(100_000))
            .price(Price::from("1.00000"))
            .build();
        let OrderEventAny::Initialized(initialized) = order.last_event().clone() else {
            panic!("Expected initialized order");
        };

        let result = engine.materialize_external_order(
            initialized,
            claimant_id,
            venue_order_id,
            instrument.id(),
            order.strategy_id(),
            UnixNanos::default(),
            None,
            None,
        );

        assert!(result.is_none());
        assert!(!cache.borrow().order_exists(&claimant_id));
        assert_eq!(
            cache.borrow().client_order_id(&venue_order_id),
            Some(&owner_id)
        );
        assert_eq!(cache.borrow().venue_order_id(&claimant_id), None);
    }

    #[rstest]
    fn carry_replay_reopen_transfers_history_allocations() {
        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
        let position_id = PositionId::from("P-REOPEN-TRANSFER");
        let opening = position_fill(
            &instrument,
            position_id,
            "O-OPEN",
            "T-OPEN",
            OrderSide::Buy,
            10,
            1,
        );
        let closing = position_fill(
            &instrument,
            position_id,
            "O-CLOSE",
            "T-CLOSE",
            OrderSide::Sell,
            10,
            2,
        );
        let reopening = position_fill(
            &instrument,
            position_id,
            "O-REOPEN",
            "T-REOPEN",
            OrderSide::Buy,
            10,
            3,
        );
        let mut position = Position::new(&instrument, opening.clone());
        position.apply(&closing);

        let fill_voided = OrderFillVoided::new(
            opening.trader_id,
            opening.strategy_id,
            opening.instrument_id,
            opening.client_order_id,
            opening.venue_order_id,
            opening.account_id,
            "C-TRANSFER".into(),
            opening.trade_id,
            Quantity::from(1),
            None,
            opening.order_side,
            opening.order_type,
            opening.last_px,
            opening.currency,
            opening.liquidity_side,
            opening.position_id,
            None,
            None,
            UUID4::new(),
            UnixNanos::from(4),
            UnixNanos::from(4),
            false,
            false,
        );
        position.fill_voids.push(PositionFillVoid {
            event: fill_voided,
            voided_qty: Quantity::from(1),
            commission_voided: None,
        });

        let cache = Rc::new(RefCell::new(Cache::default()));
        cache
            .borrow_mut()
            .add_position_without_order(&position, OmsType::Netting)
            .unwrap();
        {
            let mut cached = cache.borrow_mut();
            let mut position = cached.position_mut(&position_id).unwrap();
            position.replay_events.reserve(8);
            position.fill_voids.reserve(4);
        }
        let (replay_ptr, void_ptr, replay_len, void_len, replay_before, voids_before) = {
            let cache = cache.borrow();
            let position = cache.position(&position_id).unwrap();
            (
                position.replay_events.as_ptr(),
                position.fill_voids.as_ptr(),
                position.replay_events.len(),
                position.fill_voids.len(),
                serde_json::to_value(&position.replay_events).unwrap(),
                serde_json::to_value(&position.fill_voids).unwrap(),
            )
        };
        let config = ExecutionEngineConfig::builder()
            .carry_replay_events_on_reopen(true)
            .build()
            .unwrap();
        let mut engine = ExecutionEngine::new(
            Rc::new(RefCell::new(VirtualClock::new())),
            Rc::clone(&cache),
            Some(config),
        );

        let events = engine.handle_position_update(&instrument, reopening, OmsType::Netting);

        assert_eq!(events.len(), 1);
        let cache = cache.borrow();
        let position = cache.position(&position_id).unwrap();
        assert_eq!(position.replay_events.as_ptr(), replay_ptr);
        assert_eq!(position.fill_voids.as_ptr(), void_ptr);
        assert_eq!(position.replay_events.len(), replay_len + 1);
        assert_eq!(position.fill_voids.len(), void_len);
        assert_eq!(
            serde_json::to_value(&position.replay_events[..replay_len]).unwrap(),
            replay_before,
        );
        assert_eq!(
            serde_json::to_value(&position.fill_voids).unwrap(),
            voids_before
        );
        assert!(matches!(
            &position.replay_events[0],
            PositionReplayEvent::Filled(fill) if fill.event_id == opening.event_id
        ));
    }

    #[rstest]
    fn position_opened_uses_state_captured_before_snapshot_subscriber() {
        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
        let position_id = PositionId::from("P-SNAPSHOT-SUBSCRIBER");
        let fill = position_fill(
            &instrument,
            position_id,
            "O-SNAPSHOT-SUBSCRIBER",
            "T-SNAPSHOT-SUBSCRIBER",
            OrderSide::Buy,
            10,
            1,
        );
        let cache = Rc::new(RefCell::new(Cache::default()));
        let config = ExecutionEngineConfig::builder()
            .snapshot_positions(true)
            .build()
            .unwrap();
        let engine = ExecutionEngine::new(
            Rc::new(RefCell::new(VirtualClock::new())),
            Rc::clone(&cache),
            Some(config),
        );
        set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
        let subscriber_cache = Rc::clone(&cache);
        let topic = switchboard::get_snapshot_position_topic(position_id);
        msgbus::subscribe_any(
            topic.as_str().into(),
            TypedHandler::from_typed::<PositionStateSnapshot, _>(move |_| {
                subscriber_cache
                    .borrow_mut()
                    .position_mut(&position_id)
                    .unwrap()
                    .quantity = Quantity::from(99);
            }),
            None,
        );

        let events = engine
            .open_position(&instrument, None, true, fill.clone(), OmsType::Netting)
            .unwrap();

        assert_eq!(
            cache.borrow().position(&position_id).unwrap().quantity,
            Quantity::from(99)
        );
        let [PositionEvent::PositionOpened(opened)] = events.as_slice() else {
            panic!("Expected one position-opened event");
        };
        assert_eq!(opened.position_id, position_id);
        assert_eq!(opened.opening_order_id, fill.client_order_id);
        assert_eq!(opened.quantity, fill.last_qty);
        assert_eq!(opened.last_px, fill.last_px);
    }

    fn position_fill(
        instrument: &InstrumentAny,
        position_id: PositionId,
        order_id: &str,
        trade_id: &str,
        order_side: OrderSide,
        quantity: u32,
        ts_event: u64,
    ) -> OrderFilled {
        OrderFilledSpec::builder()
            .instrument_id(instrument.id())
            .client_order_id(ClientOrderId::from(order_id))
            .venue_order_id(VenueOrderId::from(order_id))
            .trade_id(TradeId::from(trade_id))
            .order_side(order_side)
            .last_qty(Quantity::from(quantity))
            .last_px(Price::from("1.00000"))
            .currency(instrument.quote_currency())
            .liquidity_side(LiquiditySide::Maker)
            .position_id(position_id)
            .ts_event(UnixNanos::from(ts_event))
            .build()
    }

    fn position_for_account(
        instrument: &InstrumentAny,
        account_id: AccountId,
        strategy_id: StrategyId,
        position_id: PositionId,
        order_side: OrderSide,
        quantity: Quantity,
    ) -> Position {
        let client_order_id = ClientOrderId::from(format!("O-{position_id}"));
        let fill = OrderFilledSpec::builder()
            .strategy_id(strategy_id)
            .instrument_id(instrument.id())
            .client_order_id(client_order_id)
            .venue_order_id(VenueOrderId::from(format!("V-{position_id}")))
            .account_id(account_id)
            .trade_id(TradeId::new(format!("T-{position_id}")))
            .order_side(order_side)
            .last_qty(quantity)
            .last_px(Price::from("1.0"))
            .currency(instrument.quote_currency())
            .liquidity_side(LiquiditySide::Maker)
            .position_id(position_id)
            .commission(Money::from("2 USD"))
            .build();

        Position::new(instrument, fill)
    }
}