nautilus_common/msgbus/

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

//! Public API functions for interacting with the message bus.
//!
//! This module provides free-standing functions that wrap the thread-local
//! message bus, providing a convenient API for:
//!
//! - Registering endpoint handlers (point-to-point messaging).
//! - Subscribing to topics (pub/sub messaging).
//! - Publishing messages to subscribers.
//! - Sending messages to endpoints.

use std::{any::Any, cell::RefCell, thread::LocalKey};

use nautilus_core::UUID4;
#[cfg(feature = "defi")]
use nautilus_model::defi::{
    Block, DefiData, Pool, PoolFeeCollect, PoolFlash, PoolLiquidityUpdate, PoolSwap,
};
use nautilus_model::{
    data::{
        Bar, Data, FundingRateUpdate, GreeksData, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
        option_chain::{OptionChainSlice, OptionGreeks},
    },
    events::{AccountState, OrderEventAny, PortfolioSnapshot, PositionEvent},
    instruments::InstrumentAny,
    orderbook::OrderBook,
    orders::OrderAny,
    position::Position,
};
use smallvec::SmallVec;
use ustr::Ustr;

pub use super::external::{process_external_typed_message, republish_external_message};
use super::{
    ACCOUNT_STATE_HANDLERS, ANY_HANDLERS, BAR_HANDLERS, BOOK_HANDLERS, BusPayloadType,
    DELTAS_HANDLERS, DEPTH_HANDLERS, FUNDING_RATE_HANDLERS, GREEKS_HANDLERS, HANDLER_BUFFER_CAP,
    INDEX_PRICE_HANDLERS, INSTRUMENT_HANDLERS, MARK_PRICE_HANDLERS, OPTION_CHAIN_HANDLERS,
    OPTION_GREEKS_HANDLERS, ORDER_EVENT_HANDLERS, PORTFOLIO_SNAPSHOT_HANDLERS,
    POSITION_EVENT_HANDLERS, QUOTE_HANDLERS, TRADE_HANDLERS,
    core::{MessageBus, Subscription},
    dispatch_tap_publish, dispatch_tap_response, dispatch_tap_send,
    external::{forward_any_to_external_egress, forward_to_external_egress},
    get_message_bus,
    matching::is_matching_backtracking,
    mstr::{Endpoint, MStr, Pattern, Topic},
    try_get_message_bus,
    typed_handler::{ShareableMessageHandler, TypedHandler, TypedIntoHandler},
};
#[cfg(feature = "defi")]
use super::{
    DEFI_BLOCK_HANDLERS, DEFI_COLLECT_HANDLERS, DEFI_FLASH_HANDLERS, DEFI_LIQUIDITY_HANDLERS,
    DEFI_POOL_HANDLERS, DEFI_SWAP_HANDLERS,
};
use crate::{
    actor::PublicationScope,
    messages::{
        data::{DataCommand, DataResponse, DataResponseEnvelope},
        execution::{ExecutionReport, TradingCommand},
    },
};

/// Registers a handler for an endpoint using runtime type dispatch (Any).
pub fn register_any(endpoint: MStr<Endpoint>, handler: ShareableMessageHandler) {
    log::debug!(
        "Registering endpoint '{endpoint}' with handler ID {}",
        handler.0.id(),
    );
    get_message_bus()
        .borrow_mut()
        .endpoints
        .insert(endpoint, handler);
}

/// Registers a response handler for a correlation ID.
pub fn register_response_handler(
    correlation_id: &UUID4,
    handler: TypedHandler<DataResponseEnvelope>,
) {
    if let Err(e) = get_message_bus()
        .borrow_mut()
        .register_response_handler(correlation_id, handler)
    {
        log::error!("Failed to register request handler: {e}");
    }
}

/// Registers a quote tick handler at an endpoint.
pub fn register_quote_endpoint(endpoint: MStr<Endpoint>, handler: TypedHandler<QuoteTick>) {
    get_message_bus()
        .borrow_mut()
        .endpoints_quotes
        .register(endpoint, handler);
}

/// Returns whether a quote tick handler is registered for the given endpoint.
#[must_use]
pub fn has_quote_endpoint(endpoint: MStr<Endpoint>) -> bool {
    get_message_bus()
        .borrow()
        .endpoints_quotes
        .is_registered(endpoint)
}

/// Registers a trade tick handler at an endpoint.
pub fn register_trade_endpoint(endpoint: MStr<Endpoint>, handler: TypedHandler<TradeTick>) {
    get_message_bus()
        .borrow_mut()
        .endpoints_trades
        .register(endpoint, handler);
}

/// Registers a bar handler at an endpoint.
pub fn register_bar_endpoint(endpoint: MStr<Endpoint>, handler: TypedHandler<Bar>) {
    get_message_bus()
        .borrow_mut()
        .endpoints_bars
        .register(endpoint, handler);
}

/// Registers an order event handler at an endpoint (ownership-based).
pub fn register_order_event_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedIntoHandler<OrderEventAny>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_order_events
        .register(endpoint, handler);
}

/// Registers an account state handler at an endpoint.
pub fn register_account_state_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedHandler<AccountState>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_account_state
        .register(endpoint, handler);
}

/// Registers a trading command handler at an endpoint (ownership-based).
pub fn register_trading_command_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedIntoHandler<TradingCommand>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_trading_commands
        .register(endpoint, handler);
}

/// Registers a data command handler at an endpoint (ownership-based).
pub fn register_data_command_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedIntoHandler<DataCommand>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_data_commands
        .register(endpoint, handler);
}

/// Registers a data response handler at an endpoint (ownership-based).
pub fn register_data_response_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedIntoHandler<DataResponse>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_data_responses
        .register(endpoint, handler);
}

/// Registers an execution report handler at an endpoint (ownership-based).
pub fn register_execution_report_endpoint(
    endpoint: MStr<Endpoint>,
    handler: TypedIntoHandler<ExecutionReport>,
) {
    get_message_bus()
        .borrow_mut()
        .endpoints_exec_reports
        .register(endpoint, handler);
}

/// Registers a data handler at an endpoint (ownership-based).
pub fn register_data_endpoint(endpoint: MStr<Endpoint>, handler: TypedIntoHandler<Data>) {
    get_message_bus()
        .borrow_mut()
        .endpoints_data
        .register(endpoint, handler);
}

/// Registers a DeFi data handler at an endpoint (ownership-based).
#[cfg(feature = "defi")]
pub fn register_defi_data_endpoint(endpoint: MStr<Endpoint>, handler: TypedIntoHandler<DefiData>) {
    get_message_bus()
        .borrow_mut()
        .endpoints_defi_data
        .register(endpoint, handler);
}

/// Deregisters the handler for an endpoint (Any-based).
pub fn deregister_any(endpoint: MStr<Endpoint>) {
    log::debug!("Deregistering endpoint '{endpoint}'");
    get_message_bus()
        .borrow_mut()
        .endpoints
        .shift_remove(&endpoint);
}

/// Returns whether an endpoint handler is registered for the given endpoint name.
///
/// An invalid endpoint name returns `false`, because registration rejects such names.
#[must_use]
pub fn has_endpoint(endpoint: &str) -> bool {
    let Ok(key) = MStr::<Endpoint>::endpoint(endpoint) else {
        return false;
    };

    get_message_bus().borrow().get_endpoint(key).is_some()
}

/// Subscribes a handler to a pattern using runtime type dispatch (Any).
///
/// # Warnings
///
/// Assigning priority handling is an advanced feature which *shouldn't
/// normally be needed by most users*. **Only assign a higher priority to the
/// subscription if you are certain of what you're doing**. If an inappropriate
/// priority is assigned then the handler may receive messages before core
/// system components have been able to process necessary calculations and
/// produce potential side effects for logically sound behavior.
pub fn subscribe_any(
    pattern: MStr<Pattern>,
    handler: ShareableMessageHandler,
    priority: Option<u32>,
) {
    let msgbus = get_message_bus();
    let mut msgbus_ref_mut = msgbus.borrow_mut();
    let sub = Subscription::new(pattern, handler, priority);

    log::debug!(
        "Subscribing {:?} for pattern '{}'",
        sub.handler,
        sub.pattern
    );

    if msgbus_ref_mut.subscriptions.contains(&sub) {
        log::warn!("{sub:?} already exists");
        return;
    }

    for (topic, subs) in &mut msgbus_ref_mut.topics {
        if is_matching_backtracking(*topic, sub.pattern) {
            subs.push(sub.clone());
            subs.sort_by(Subscription::delivery_order);
            log::debug!("Added subscription for '{topic}'");
        }
    }

    msgbus_ref_mut.subscriptions.insert(sub);
}

/// Subscribes a handler to instrument messages matching a pattern.
pub fn subscribe_instruments(
    pattern: MStr<Pattern>,
    handler: TypedHandler<InstrumentAny>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_instruments.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to instrument close messages matching a pattern.
pub fn subscribe_instrument_close(
    pattern: MStr<Pattern>,
    handler: ShareableMessageHandler,
    priority: Option<u32>,
) {
    subscribe_any(pattern, handler, priority);
}

/// Subscribes a handler to order book deltas matching a pattern.
pub fn subscribe_book_deltas(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OrderBookDeltas>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_deltas
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to order book depth snapshots matching a pattern.
pub fn subscribe_book_depth(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OrderBookDepth>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_depth
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to order book snapshots matching a pattern.
pub fn subscribe_book_snapshots(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OrderBook>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_book_snapshots
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to quote ticks matching a pattern.
pub fn subscribe_quotes(
    pattern: MStr<Pattern>,
    handler: TypedHandler<QuoteTick>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_quotes
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to trade ticks matching a pattern.
pub fn subscribe_trades(
    pattern: MStr<Pattern>,
    handler: TypedHandler<TradeTick>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_trades
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to bars matching a pattern.
pub fn subscribe_bars(pattern: MStr<Pattern>, handler: TypedHandler<Bar>, priority: Option<u32>) {
    get_message_bus()
        .borrow_mut()
        .router_bars
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to mark price updates matching a pattern.
pub fn subscribe_mark_prices(
    pattern: MStr<Pattern>,
    handler: TypedHandler<MarkPriceUpdate>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_mark_prices.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to index price updates matching a pattern.
pub fn subscribe_index_prices(
    pattern: MStr<Pattern>,
    handler: TypedHandler<IndexPriceUpdate>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_index_prices
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to funding rate updates matching a pattern.
pub fn subscribe_funding_rates(
    pattern: MStr<Pattern>,
    handler: TypedHandler<FundingRateUpdate>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_funding_rates
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to greeks data matching a pattern.
pub fn subscribe_greeks(
    pattern: MStr<Pattern>,
    handler: TypedHandler<GreeksData>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_greeks
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to option greeks updates matching a pattern.
pub fn subscribe_option_greeks(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OptionGreeks>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_option_greeks
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to option chain slice updates matching a pattern.
pub fn subscribe_option_chain(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OptionChainSlice>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_option_chain
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to order events matching a pattern.
pub fn subscribe_order_events(
    pattern: MStr<Pattern>,
    handler: TypedHandler<OrderEventAny>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_order_events
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to position events matching a pattern.
pub fn subscribe_position_events(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PositionEvent>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_position_events
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to positions matching a pattern.
pub fn subscribe_positions(
    pattern: MStr<Pattern>,
    handler: TypedHandler<Position>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_positions.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to account state updates matching a pattern.
pub fn subscribe_account_state(
    pattern: MStr<Pattern>,
    handler: TypedHandler<AccountState>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_account_state
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to portfolio snapshots matching a pattern.
pub fn subscribe_portfolio_snapshot(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PortfolioSnapshot>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_portfolio.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to DeFi blocks matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_blocks(
    pattern: MStr<Pattern>,
    handler: TypedHandler<Block>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_defi_blocks.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to DeFi pools matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_pools(
    pattern: MStr<Pattern>,
    handler: TypedHandler<Pool>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_defi_pools.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to DeFi pool swaps matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_swaps(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PoolSwap>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_defi_swaps.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Subscribes a handler to DeFi liquidity updates matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_liquidity(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PoolLiquidityUpdate>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_defi_liquidity
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to DeFi fee collects matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_collects(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PoolFeeCollect>,
    priority: Option<u32>,
) {
    get_message_bus()
        .borrow_mut()
        .router_defi_collects
        .subscribe(pattern, handler, priority.unwrap_or(0));
}

/// Subscribes a handler to DeFi flash loans matching a pattern.
#[cfg(feature = "defi")]
pub fn subscribe_defi_flash(
    pattern: MStr<Pattern>,
    handler: TypedHandler<PoolFlash>,
    priority: Option<u32>,
) {
    get_message_bus().borrow_mut().router_defi_flash.subscribe(
        pattern,
        handler,
        priority.unwrap_or(0),
    );
}

/// Unsubscribes a handler from instrument messages.
pub fn unsubscribe_instruments(pattern: MStr<Pattern>, handler: &TypedHandler<InstrumentAny>) {
    get_message_bus()
        .borrow_mut()
        .router_instruments
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from instrument close messages.
pub fn unsubscribe_instrument_close(pattern: MStr<Pattern>, handler: &ShareableMessageHandler) {
    unsubscribe_any(pattern, handler);
}

/// Unsubscribes a handler from order book deltas.
pub fn unsubscribe_book_deltas(pattern: MStr<Pattern>, handler: &TypedHandler<OrderBookDeltas>) {
    get_message_bus()
        .borrow_mut()
        .router_deltas
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from order book depth snapshots.
pub fn unsubscribe_book_depth(pattern: MStr<Pattern>, handler: &TypedHandler<OrderBookDepth>) {
    get_message_bus()
        .borrow_mut()
        .router_depth
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from order book snapshots.
pub fn unsubscribe_book_snapshots(pattern: MStr<Pattern>, handler: &TypedHandler<OrderBook>) {
    get_message_bus()
        .borrow_mut()
        .router_book_snapshots
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from quote ticks.
pub fn unsubscribe_quotes(pattern: MStr<Pattern>, handler: &TypedHandler<QuoteTick>) {
    get_message_bus()
        .borrow_mut()
        .router_quotes
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from trade ticks.
pub fn unsubscribe_trades(pattern: MStr<Pattern>, handler: &TypedHandler<TradeTick>) {
    get_message_bus()
        .borrow_mut()
        .router_trades
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from bars.
pub fn unsubscribe_bars(pattern: MStr<Pattern>, handler: &TypedHandler<Bar>) {
    get_message_bus()
        .borrow_mut()
        .router_bars
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from mark price updates.
pub fn unsubscribe_mark_prices(pattern: MStr<Pattern>, handler: &TypedHandler<MarkPriceUpdate>) {
    get_message_bus()
        .borrow_mut()
        .router_mark_prices
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from index price updates.
pub fn unsubscribe_index_prices(pattern: MStr<Pattern>, handler: &TypedHandler<IndexPriceUpdate>) {
    get_message_bus()
        .borrow_mut()
        .router_index_prices
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from funding rate updates.
pub fn unsubscribe_funding_rates(
    pattern: MStr<Pattern>,
    handler: &TypedHandler<FundingRateUpdate>,
) {
    get_message_bus()
        .borrow_mut()
        .router_funding_rates
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from account state updates.
pub fn unsubscribe_account_state(pattern: MStr<Pattern>, handler: &TypedHandler<AccountState>) {
    get_message_bus()
        .borrow_mut()
        .router_account_state
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from portfolio snapshots.
pub fn unsubscribe_portfolio_snapshot(
    pattern: MStr<Pattern>,
    handler: &TypedHandler<PortfolioSnapshot>,
) {
    get_message_bus()
        .borrow_mut()
        .router_portfolio
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from order events.
pub fn unsubscribe_order_events(pattern: MStr<Pattern>, handler: &TypedHandler<OrderEventAny>) {
    get_message_bus()
        .borrow_mut()
        .router_order_events
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from position events.
pub fn unsubscribe_position_events(pattern: MStr<Pattern>, handler: &TypedHandler<PositionEvent>) {
    get_message_bus()
        .borrow_mut()
        .router_position_events
        .unsubscribe(pattern, handler);
}

/// Removes a specific order event handler by pattern and handler ID.
pub fn remove_order_event_handler(pattern: MStr<Pattern>, handler_id: Ustr) {
    get_message_bus()
        .borrow_mut()
        .router_order_events
        .remove_handler(pattern, handler_id);
}

/// Removes a specific position event handler by pattern and handler ID.
pub fn remove_position_event_handler(pattern: MStr<Pattern>, handler_id: Ustr) {
    get_message_bus()
        .borrow_mut()
        .router_position_events
        .remove_handler(pattern, handler_id);
}

/// Unsubscribes a handler from orders.
pub fn unsubscribe_orders(pattern: MStr<Pattern>, handler: &TypedHandler<OrderAny>) {
    get_message_bus()
        .borrow_mut()
        .router_orders
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from positions.
pub fn unsubscribe_positions(pattern: MStr<Pattern>, handler: &TypedHandler<Position>) {
    get_message_bus()
        .borrow_mut()
        .router_positions
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from greeks data.
pub fn unsubscribe_greeks(pattern: MStr<Pattern>, handler: &TypedHandler<GreeksData>) {
    get_message_bus()
        .borrow_mut()
        .router_greeks
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from option greeks updates.
pub fn unsubscribe_option_greeks(pattern: MStr<Pattern>, handler: &TypedHandler<OptionGreeks>) {
    get_message_bus()
        .borrow_mut()
        .router_option_greeks
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from option chain slice updates.
pub fn unsubscribe_option_chain(pattern: MStr<Pattern>, handler: &TypedHandler<OptionChainSlice>) {
    get_message_bus()
        .borrow_mut()
        .router_option_chain
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi blocks.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_blocks(pattern: MStr<Pattern>, handler: &TypedHandler<Block>) {
    get_message_bus()
        .borrow_mut()
        .router_defi_blocks
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi pools.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_pools(pattern: MStr<Pattern>, handler: &TypedHandler<Pool>) {
    get_message_bus()
        .borrow_mut()
        .router_defi_pools
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi pool swaps.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_swaps(pattern: MStr<Pattern>, handler: &TypedHandler<PoolSwap>) {
    get_message_bus()
        .borrow_mut()
        .router_defi_swaps
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi liquidity updates.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_liquidity(
    pattern: MStr<Pattern>,
    handler: &TypedHandler<PoolLiquidityUpdate>,
) {
    get_message_bus()
        .borrow_mut()
        .router_defi_liquidity
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi fee collects.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_collects(pattern: MStr<Pattern>, handler: &TypedHandler<PoolFeeCollect>) {
    get_message_bus()
        .borrow_mut()
        .router_defi_collects
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from DeFi flash loans.
#[cfg(feature = "defi")]
pub fn unsubscribe_defi_flash(pattern: MStr<Pattern>, handler: &TypedHandler<PoolFlash>) {
    get_message_bus()
        .borrow_mut()
        .router_defi_flash
        .unsubscribe(pattern, handler);
}

/// Unsubscribes a handler from a pattern (Any-based).
pub fn unsubscribe_any(pattern: MStr<Pattern>, handler: &ShareableMessageHandler) {
    get_message_bus()
        .borrow_mut()
        .unsubscribe_any(pattern, handler);
}

/// Checks if a handler is subscribed to a pattern (Any-based).
pub fn is_subscribed_any<T: AsRef<str>>(pattern: T, handler: ShareableMessageHandler) -> bool {
    let pattern = MStr::from(pattern.as_ref());
    let sub = Subscription::new(pattern, handler, None);
    get_message_bus().borrow().subscriptions.contains(&sub)
}

/// Returns the count of Any-based subscriptions for a topic.
///
/// # Errors
///
/// Returns an error if the `topic` is not a valid topic string.
pub fn subscriptions_count_any<S: AsRef<str>>(topic: S) -> anyhow::Result<usize> {
    get_message_bus().borrow().subscriptions_count(topic)
}

/// Returns the subscriber count for order book deltas on a topic.
pub fn subscriber_count_deltas(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_deltas
        .subscriber_count(topic)
}

/// Returns the subscriber count for order book depth on a topic.
pub fn subscriber_count_depth(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_depth
        .subscriber_count(topic)
}

/// Returns the subscriber count for order book snapshots on a topic.
pub fn subscriber_count_book_snapshots(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_book_snapshots
        .subscriber_count(topic)
}

/// Returns the exact subscriber count for quotes on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_quotes(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_quotes
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for trades on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_trades(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_trades
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for mark prices on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_mark_prices(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_mark_prices
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for index prices on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_index_prices(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_index_prices
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for funding rates on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_funding_rates(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_funding_rates
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for option greeks on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_option_greeks(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_option_greeks
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for option chains on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_option_chain(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_option_chain
        .exact_subscriber_count(topic)
}

/// Returns the exact subscriber count for bars on a topic,
/// excluding wildcard pattern subscriptions.
pub fn exact_subscriber_count_bars(topic: MStr<Topic>) -> usize {
    get_message_bus()
        .borrow()
        .router_bars
        .exact_subscriber_count(topic)
}

/// Publishes a message to the topic using runtime type dispatch (Any).
pub fn publish_any(topic: MStr<Topic>, message: &dyn Any) {
    let _publication = PublicationScope::enter();
    dispatch_tap_publish(topic, message);

    // Take buffer (re-entrancy safe)
    let mut handlers = ANY_HANDLERS.with_borrow_mut(std::mem::take);

    {
        let bus_rc = get_message_bus();
        let mut bus = bus_rc.borrow_mut();
        bus.fill_matching_any_handlers(topic, &mut handlers);
        bus.increment_pub_count();
    }

    for handler in &handlers {
        handler.0.handle(message);
    }

    handlers.clear(); // Release refs before restore
    ANY_HANDLERS.with_borrow_mut(|buf| *buf = handlers);

    forward_any_to_external_egress(topic, message);
}

/// Tries to publish a message to the current thread's registered message bus.
///
/// Returns `false` when the thread has no bus or the bus is already borrowed.
pub fn try_publish_any(topic: MStr<Topic>, message: &dyn Any) -> bool {
    let Some(bus_rc) = try_get_message_bus() else {
        return false;
    };

    if bus_rc.try_borrow_mut().is_err() {
        return false;
    }

    let _publication = PublicationScope::enter();
    dispatch_tap_publish(topic, message);

    let Ok(mut bus) = bus_rc.try_borrow_mut() else {
        return false;
    };

    // Take buffer (re-entrancy safe)
    let mut handlers = ANY_HANDLERS.with_borrow_mut(std::mem::take);

    bus.fill_matching_any_handlers(topic, &mut handlers);
    bus.increment_pub_count();
    drop(bus);

    for handler in &handlers {
        handler.0.handle(message);
    }

    handlers.clear(); // Release refs before restore
    ANY_HANDLERS.with_borrow_mut(|buf| *buf = handlers);
    true
}

/// Publishes an instrument to subscribers on a topic.
pub fn publish_instrument(topic: MStr<Topic>, instrument: &InstrumentAny) {
    publish_typed(
        topic,
        &INSTRUMENT_HANDLERS,
        |bus, h| bus.router_instruments.fill_matching_handlers(topic, h),
        instrument,
    );

    forward_to_external_egress(topic, BusPayloadType::Instrument, instrument);
}

/// Publishes order book deltas to subscribers on a topic.
pub fn publish_deltas(topic: MStr<Topic>, deltas: &OrderBookDeltas) {
    publish_typed(
        topic,
        &DELTAS_HANDLERS,
        |bus, h| bus.router_deltas.fill_matching_handlers(topic, h),
        deltas,
    );

    forward_to_external_egress(topic, BusPayloadType::OrderBookDeltas, deltas);
}

/// Publishes order book depth to subscribers on a topic.
pub fn publish_depth(topic: MStr<Topic>, depth: &OrderBookDepth) {
    publish_typed(
        topic,
        &DEPTH_HANDLERS,
        |bus, h| bus.router_depth.fill_matching_handlers(topic, h),
        depth,
    );

    forward_to_external_egress(topic, BusPayloadType::OrderBookDepth, depth);
}

/// Publishes an order book snapshot to subscribers on a topic.
pub fn publish_book(topic: MStr<Topic>, book: &OrderBook) {
    publish_typed(
        topic,
        &BOOK_HANDLERS,
        |bus, h| bus.router_book_snapshots.fill_matching_handlers(topic, h),
        book,
    );
}

/// Publishes a quote tick to subscribers on a topic.
pub fn publish_quote(topic: MStr<Topic>, quote: &QuoteTick) {
    publish_typed(
        topic,
        &QUOTE_HANDLERS,
        |bus, h| bus.router_quotes.fill_matching_handlers(topic, h),
        quote,
    );

    forward_to_external_egress(topic, BusPayloadType::QuoteTick, quote);
}

/// Publishes a trade tick to subscribers on a topic.
pub fn publish_trade(topic: MStr<Topic>, trade: &TradeTick) {
    publish_typed(
        topic,
        &TRADE_HANDLERS,
        |bus, h| bus.router_trades.fill_matching_handlers(topic, h),
        trade,
    );

    forward_to_external_egress(topic, BusPayloadType::TradeTick, trade);
}

/// Publishes a bar to subscribers on a topic.
pub fn publish_bar(topic: MStr<Topic>, bar: &Bar) {
    publish_typed(
        topic,
        &BAR_HANDLERS,
        |bus, h| bus.router_bars.fill_matching_handlers(topic, h),
        bar,
    );

    forward_to_external_egress(topic, BusPayloadType::Bar, bar);
}

/// Publishes a mark price update to subscribers on a topic.
pub fn publish_mark_price(topic: MStr<Topic>, mark_price: &MarkPriceUpdate) {
    publish_typed(
        topic,
        &MARK_PRICE_HANDLERS,
        |bus, h| bus.router_mark_prices.fill_matching_handlers(topic, h),
        mark_price,
    );

    forward_to_external_egress(topic, BusPayloadType::MarkPriceUpdate, mark_price);
}

/// Publishes an index price update to subscribers on a topic.
pub fn publish_index_price(topic: MStr<Topic>, index_price: &IndexPriceUpdate) {
    publish_typed(
        topic,
        &INDEX_PRICE_HANDLERS,
        |bus, h| bus.router_index_prices.fill_matching_handlers(topic, h),
        index_price,
    );

    forward_to_external_egress(topic, BusPayloadType::IndexPriceUpdate, index_price);
}

/// Publishes a funding rate update to subscribers on a topic.
pub fn publish_funding_rate(topic: MStr<Topic>, funding_rate: &FundingRateUpdate) {
    publish_typed(
        topic,
        &FUNDING_RATE_HANDLERS,
        |bus, h| bus.router_funding_rates.fill_matching_handlers(topic, h),
        funding_rate,
    );

    forward_to_external_egress(topic, BusPayloadType::FundingRateUpdate, funding_rate);
}

/// Publishes greeks data to subscribers on a topic.
pub fn publish_greeks(topic: MStr<Topic>, greeks: &GreeksData) {
    publish_typed(
        topic,
        &GREEKS_HANDLERS,
        |bus, h| bus.router_greeks.fill_matching_handlers(topic, h),
        greeks,
    );
}

/// Publishes option greeks to subscribers on a topic.
pub fn publish_option_greeks(topic: MStr<Topic>, option_greeks: &OptionGreeks) {
    publish_typed(
        topic,
        &OPTION_GREEKS_HANDLERS,
        |bus, h| bus.router_option_greeks.fill_matching_handlers(topic, h),
        option_greeks,
    );

    forward_to_external_egress(topic, BusPayloadType::OptionGreeks, option_greeks);
}

/// Publishes an option chain slice to subscribers on a topic.
pub fn publish_option_chain(topic: MStr<Topic>, slice: &OptionChainSlice) {
    publish_typed(
        topic,
        &OPTION_CHAIN_HANDLERS,
        |bus, h| bus.router_option_chain.fill_matching_handlers(topic, h),
        slice,
    );
}

/// Publishes an account state to subscribers on a topic.
pub fn publish_account_state(topic: MStr<Topic>, state: &AccountState) {
    publish_typed(
        topic,
        &ACCOUNT_STATE_HANDLERS,
        |bus, h| bus.router_account_state.fill_matching_handlers(topic, h),
        state,
    );

    forward_to_external_egress(topic, BusPayloadType::AccountState, state);
}

/// Publishes a portfolio snapshot to subscribers on a topic.
pub fn publish_portfolio_snapshot(topic: MStr<Topic>, snapshot: &PortfolioSnapshot) {
    publish_typed(
        topic,
        &PORTFOLIO_SNAPSHOT_HANDLERS,
        |bus, h| {
            bus.router_portfolio.fill_matching_handlers(topic, h);
        },
        snapshot,
    );

    forward_to_external_egress(topic, BusPayloadType::PortfolioSnapshot, snapshot);
}

/// Publishes an order event to subscribers on a topic.
pub fn publish_order_event(topic: MStr<Topic>, event: &OrderEventAny) {
    publish_typed(
        topic,
        &ORDER_EVENT_HANDLERS,
        |bus, h| bus.router_order_events.fill_matching_handlers(topic, h),
        event,
    );

    forward_to_external_egress(topic, BusPayloadType::OrderEvent, event);
}

/// Publishes a position event to subscribers on a topic.
pub fn publish_position_event(topic: MStr<Topic>, event: &PositionEvent) {
    publish_typed(
        topic,
        &POSITION_EVENT_HANDLERS,
        |bus, h| bus.router_position_events.fill_matching_handlers(topic, h),
        event,
    );

    forward_to_external_egress(topic, BusPayloadType::PositionEvent, event);
}

/// Publishes a DeFi block to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_block(topic: MStr<Topic>, block: &Block) {
    publish_typed(
        topic,
        &DEFI_BLOCK_HANDLERS,
        |bus, h| bus.router_defi_blocks.fill_matching_handlers(topic, h),
        block,
    );

    forward_to_external_egress(topic, BusPayloadType::Block, block);
}

/// Publishes a DeFi pool to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_pool(topic: MStr<Topic>, pool: &Pool) {
    publish_typed(
        topic,
        &DEFI_POOL_HANDLERS,
        |bus, h| bus.router_defi_pools.fill_matching_handlers(topic, h),
        pool,
    );

    forward_to_external_egress(topic, BusPayloadType::Pool, pool);
}

/// Publishes a DeFi pool swap to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_swap(topic: MStr<Topic>, swap: &PoolSwap) {
    publish_typed(
        topic,
        &DEFI_SWAP_HANDLERS,
        |bus, h| bus.router_defi_swaps.fill_matching_handlers(topic, h),
        swap,
    );
}

/// Publishes a DeFi liquidity update to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_liquidity(topic: MStr<Topic>, update: &PoolLiquidityUpdate) {
    publish_typed(
        topic,
        &DEFI_LIQUIDITY_HANDLERS,
        |bus, h| bus.router_defi_liquidity.fill_matching_handlers(topic, h),
        update,
    );

    forward_to_external_egress(topic, BusPayloadType::PoolLiquidityUpdate, update);
}

/// Publishes a DeFi fee collect to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_collect(topic: MStr<Topic>, collect: &PoolFeeCollect) {
    publish_typed(
        topic,
        &DEFI_COLLECT_HANDLERS,
        |bus, h| bus.router_defi_collects.fill_matching_handlers(topic, h),
        collect,
    );

    forward_to_external_egress(topic, BusPayloadType::PoolFeeCollect, collect);
}

/// Publishes a DeFi flash loan to subscribers on a topic.
#[cfg(feature = "defi")]
pub fn publish_defi_flash(topic: MStr<Topic>, flash: &PoolFlash) {
    publish_typed(
        topic,
        &DEFI_FLASH_HANDLERS,
        |bus, h| bus.router_defi_flash.fill_matching_handlers(topic, h),
        flash,
    );

    forward_to_external_egress(topic, BusPayloadType::PoolFlash, flash);
}

/// Publishes a message to typed handlers using thread-local buffer reuse.
///
/// The `fill_fn` receives a mutable reference to the `MessageBus`, avoiding
/// redundant TLS access and Rc clone/drop overhead per publish.
///
/// Before fanout the registered bus tap (if any) observes the message. Capture must
/// precede subscriber dispatch so the durable record exists before any handler reacts
/// to the message.
///
/// # Invariants
///
/// - `fill_fn` must not call any publish path (would panic from `RefCell` double-borrow).
/// - Handler panics drop the buffer, losing reuse optimization (acceptable as panics are fatal).
#[inline]
fn publish_typed<T: 'static>(
    topic: MStr<Topic>,
    tls: &'static LocalKey<RefCell<SmallVec<[TypedHandler<T>; HANDLER_BUFFER_CAP]>>>,
    fill_fn: impl FnOnce(&mut MessageBus, &mut SmallVec<[TypedHandler<T>; HANDLER_BUFFER_CAP]>),
    message: &T,
) {
    let _publication = PublicationScope::enter();
    dispatch_tap_publish(topic, message);

    // Take buffer (re-entrancy safe)
    let mut handlers = tls.with_borrow_mut(std::mem::take);

    // Borrow scope ends before dispatch to support re-entrant publishes
    let bus_rc = get_message_bus();
    {
        let mut bus = bus_rc.borrow_mut();
        fill_fn(&mut bus, &mut handlers);
        bus.increment_pub_count();
    }

    for handler in &handlers {
        handler.handle(message);
    }

    handlers.clear(); // Release refs before restore
    tls.with_borrow_mut(|buf| *buf = handlers);
}

/// Sends a message to an endpoint handler using runtime type dispatch (Any).
pub fn send_any(endpoint: MStr<Endpoint>, message: &dyn Any) {
    send_any_inner(endpoint, message, "send_any");
}

/// Sends a message to an endpoint, converting to Any (convenience wrapper).
pub fn send_any_value<T: 'static>(endpoint: MStr<Endpoint>, message: &T) {
    send_any_inner(endpoint, message, "send_any_value");
}

#[inline]
fn send_any_inner(endpoint: MStr<Endpoint>, message: &dyn Any, fn_name: &str) {
    dispatch_tap_send(endpoint, message);

    let handler = {
        let bus = get_message_bus();
        let mut bus = bus.borrow_mut();
        let handler = bus.get_endpoint(endpoint).cloned();
        if handler.is_some() {
            bus.increment_sent_count();
        }
        handler
    };

    if let Some(handler) = handler {
        handler.0.handle(message);
    } else {
        log::error!("{fn_name}: no registered endpoint '{endpoint}'");
    }
}

/// Sends the [`DataResponse`] to the registered correlation ID handler.
pub fn send_response(correlation_id: &UUID4, message: &DataResponse) {
    let source = (correlation_id != message.correlation_id()).then_some(*message.correlation_id());
    let envelope = DataResponseEnvelope::new(message.clone(), true, source)
        .with_correlation_id(*correlation_id);
    send_response_envelope(&envelope);
}

/// Delivers an envelope and removes its correlation handler only on completion.
pub fn send_response_envelope(message: &DataResponseEnvelope) {
    let correlation_id = message.correlation_id();
    dispatch_tap_response(correlation_id, message);

    let handler = {
        let bus = get_message_bus();
        let mut bus = bus.borrow_mut();
        let handler = if message.completed() {
            bus.take_response_handler(correlation_id)
        } else {
            bus.get_response_handler(correlation_id).cloned()
        };

        bus.increment_res_count();
        handler
    };

    if let Some(handler) = handler {
        handler.0.handle(message);
    } else {
        log::debug!("Response handler not found for correlation_id '{correlation_id}'");
    }
}

/// Sends a quote tick to an endpoint handler.
pub fn send_quote(endpoint: MStr<Endpoint>, quote: &QuoteTick) {
    send_endpoint_ref(
        endpoint,
        quote,
        |bus| bus.endpoints_quotes.get(endpoint),
        "send_quote",
    );
}

/// Sends a trade tick to an endpoint handler.
pub fn send_trade(endpoint: MStr<Endpoint>, trade: &TradeTick) {
    send_endpoint_ref(
        endpoint,
        trade,
        |bus| bus.endpoints_trades.get(endpoint),
        "send_trade",
    );
}

/// Sends a bar to an endpoint handler.
pub fn send_bar(endpoint: MStr<Endpoint>, bar: &Bar) {
    send_endpoint_ref(
        endpoint,
        bar,
        |bus| bus.endpoints_bars.get(endpoint),
        "send_bar",
    );
}

/// Sends an order event to an endpoint handler, transferring ownership.
pub fn send_order_event(endpoint: MStr<Endpoint>, event: OrderEventAny) {
    send_endpoint_owned(
        endpoint,
        event,
        |bus| bus.endpoints_order_events.get(endpoint),
        "send_order_event",
    );
}

/// Sends an account state to an endpoint handler.
pub fn send_account_state(endpoint: MStr<Endpoint>, state: &AccountState) {
    send_endpoint_ref(
        endpoint,
        state,
        |bus| bus.endpoints_account_state.get(endpoint),
        "send_account_state",
    );
}

/// Sends a trading command to an endpoint handler, transferring ownership.
pub fn send_trading_command(endpoint: MStr<Endpoint>, command: TradingCommand) {
    send_endpoint_owned(
        endpoint,
        command,
        |bus| bus.endpoints_trading_commands.get(endpoint),
        "send_trading_command",
    );
}

/// Sends a data command to an endpoint handler, transferring ownership.
pub fn send_data_command(endpoint: MStr<Endpoint>, command: DataCommand) {
    let is_request = data_command_is_request(&command);
    send_endpoint_owned_counted(
        endpoint,
        command,
        |bus| bus.endpoints_data_commands.get(endpoint),
        "send_data_command",
        is_request,
    );
}

/// Sends a data response to an endpoint handler, transferring ownership.
pub fn send_data_response(endpoint: MStr<Endpoint>, response: DataResponse) {
    send_endpoint_owned(
        endpoint,
        response,
        |bus| bus.endpoints_data_responses.get(endpoint),
        "send_data_response",
    );
}

/// Sends an execution report to an endpoint handler, transferring ownership.
pub fn send_execution_report(endpoint: MStr<Endpoint>, report: ExecutionReport) {
    send_endpoint_owned(
        endpoint,
        report,
        |bus| bus.endpoints_exec_reports.get(endpoint),
        "send_execution_report",
    );
}

/// Sends data to an endpoint handler, transferring ownership.
pub fn send_data(endpoint: MStr<Endpoint>, data: Data) {
    send_endpoint_owned(
        endpoint,
        data,
        |bus| bus.endpoints_data.get(endpoint),
        "send_data",
    );
}

/// Sends DeFi data to an endpoint handler, transferring ownership.
#[cfg(feature = "defi")]
pub fn send_defi_data(endpoint: MStr<Endpoint>, data: DefiData) {
    send_endpoint_owned(
        endpoint,
        data,
        |bus| bus.endpoints_defi_data.get(endpoint),
        "send_defi_data",
    );
}

#[inline]
fn send_endpoint_ref<T: 'static, F>(
    endpoint: MStr<Endpoint>,
    message: &T,
    get_handler: F,
    fn_name: &str,
) where
    F: FnOnce(&MessageBus) -> Option<&TypedHandler<T>>,
{
    dispatch_tap_send(endpoint, message);

    let handler = {
        let bus = get_message_bus();
        let mut bus = bus.borrow_mut();
        let handler = get_handler(&bus).cloned();
        if handler.is_some() {
            bus.increment_sent_count();
        }
        handler
    };

    if let Some(handler) = handler {
        handler.handle(message);
    } else {
        log::error!("{fn_name}: no registered endpoint '{endpoint}'");
    }
}

#[inline]
fn send_endpoint_owned<T: 'static, F>(
    endpoint: MStr<Endpoint>,
    message: T,
    get_handler: F,
    fn_name: &str,
) where
    F: FnOnce(&MessageBus) -> Option<&TypedIntoHandler<T>>,
{
    send_endpoint_owned_counted(endpoint, message, get_handler, fn_name, false);
}

#[inline]
fn send_endpoint_owned_counted<T: 'static, F>(
    endpoint: MStr<Endpoint>,
    message: T,
    get_handler: F,
    fn_name: &str,
    count_request: bool,
) where
    F: FnOnce(&MessageBus) -> Option<&TypedIntoHandler<T>>,
{
    // Capture before the dispatch consumes `message`
    dispatch_tap_send(endpoint, &message);

    let handler = {
        let bus = get_message_bus();
        let mut bus = bus.borrow_mut();
        let handler = get_handler(&bus).cloned();
        if handler.is_some() {
            bus.increment_sent_count();
            if count_request {
                bus.increment_req_count();
            }
        }
        handler
    };

    if let Some(handler) = handler {
        handler.handle(message);
    } else {
        log::error!("{fn_name}: no registered endpoint '{endpoint}'");
    }
}

#[inline]
fn data_command_is_request(command: &DataCommand) -> bool {
    match command {
        DataCommand::Request(_) => true,
        #[cfg(feature = "defi")]
        DataCommand::DefiRequest(_) => true,
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    //! Tests for the message bus API functions.
    //!
    //! Includes re-entrancy tests that verify handlers can call back into the
    //! message bus without causing `RefCell` borrow conflicts. This is the scenario
    //! where `send_*` holds a borrow, calls the handler, and the handler needs to
    //! call `borrow_mut()` for topic getters or other operations.

    use std::{
        cell::{Cell, RefCell},
        rc::Rc,
        sync::Arc,
        thread,
    };

    #[cfg(feature = "defi")]
    use alloy_primitives::{U256, address};
    use bytes::Bytes;
    use nautilus_core::{UUID4, UnixNanos};
    #[cfg(feature = "defi")]
    use nautilus_model::defi::{
        AmmType, Chain, Dex, DexType, PoolIdentifier, PoolLiquidityUpdateType, Token,
    };
    #[cfg(feature = "sbe")]
    use nautilus_model::{data::OptionGreekValues, enums::GreeksConvention};
    use nautilus_model::{
        data::{
            Bar, BarType, CustomData, DataType, FundingRateUpdate, IndexPriceUpdate,
            MarkPriceUpdate, OptionGreeks, OrderBookDelta, OrderBookDeltas, OrderBookDepth,
            QuoteTick, TradeTick,
            stubs::{stub_custom_data, stub_deltas, stub_depth10},
        },
        enums::{
            AccountType, BookType, LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide,
            TimeInForce,
        },
        events::{OrderEventAny, PositionEvent, PositionOpened, order::spec::OrderDeniedSpec},
        identifiers::{
            AccountId, ClientId, ClientOrderId, InstrumentId, OptionSeriesId, PositionId,
            StrategyId, TradeId, TraderId, Venue, VenueOrderId,
        },
        instruments::{InstrumentAny, stubs::audusd_sim},
        orderbook::OrderBook,
        reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
        types::{Currency, Money, Price, Quantity},
    };
    #[cfg(feature = "sbe")]
    use nautilus_serialization::sbe::FromSbe;
    use rstest::rstest;
    use rust_decimal::Decimal;

    use super::*;
    use crate::{
        enums::SerializationEncoding,
        messages::{
            data::{
                BarsResponse, BookDeltasResponse, BookDepthResponse, BookResponse,
                CustomDataResponse, DataCommand, DataResponse, DataResponseEnvelope,
                FundingRatesResponse, InstrumentResponse, InstrumentsResponse,
                OptionChainReferencePriceResponse, QuotesResponse, RequestCommand, RequestQuotes,
                SubscribeCommand, SubscribeQuotes, TradesResponse, UnsubscribeCommand,
                UnsubscribeQuotes,
            },
            execution::{CancelAllOrders, GenerateExecutionMassStatus, TradingCommand},
        },
        msgbus::{
            BusMessage, BusTap, MessageBusConfig, MessageBusExternalEgress, MessagingSwitchboard,
            SuppressExternalGuard, clear_bus_tap, set_bus_tap, set_message_bus,
            stubs::get_call_check_handler,
        },
    };

    #[derive(Debug)]
    struct CapturedEgressMessage {
        topic: String,
        encoding: SerializationEncoding,
        payload_type: BusPayloadType,
        payload: Bytes,
    }

    struct CapturingExternalEgress {
        publications: Rc<RefCell<Vec<CapturedEgressMessage>>>,
        closed: Cell<bool>,
    }

    impl CapturingExternalEgress {
        fn new() -> (Self, Rc<RefCell<Vec<CapturedEgressMessage>>>) {
            let publications = Rc::new(RefCell::new(Vec::new()));
            (
                Self {
                    publications: publications.clone(),
                    closed: Cell::new(false),
                },
                publications,
            )
        }
    }

    impl MessageBusExternalEgress for CapturingExternalEgress {
        fn is_closed(&self) -> bool {
            self.closed.get()
        }

        fn publish(&self, message: BusMessage) {
            self.publications.borrow_mut().push(CapturedEgressMessage {
                topic: message.topic.to_string(),
                encoding: message.encoding,
                payload_type: message.payload_type,
                payload: message.payload,
            });
        }

        fn close(&mut self) {
            self.closed.set(true);
        }
    }

    fn install_capturing_external_egress(
        encoding: SerializationEncoding,
    ) -> Rc<RefCell<Vec<CapturedEgressMessage>>> {
        let msgbus = Rc::new(RefCell::new(MessageBus::default()));
        set_message_bus(msgbus.clone());
        let (external_egress, publications) = CapturingExternalEgress::new();
        msgbus
            .borrow_mut()
            .set_external_egress(Box::new(external_egress), encoding);
        publications
    }

    fn install_capturing_external_egress_config(
        config: &MessageBusConfig,
    ) -> Rc<RefCell<Vec<CapturedEgressMessage>>> {
        let msgbus = Rc::new(RefCell::new(MessageBus::default()));
        set_message_bus(msgbus.clone());
        let (external_egress, publications) = CapturingExternalEgress::new();
        msgbus
            .borrow_mut()
            .set_external_egress_config(Box::new(external_egress), config)
            .expect("message bus config must be valid");
        publications
    }

    fn install_capturing_external_stream_egress(
        encoding: SerializationEncoding,
    ) -> Rc<RefCell<Vec<CapturedEgressMessage>>> {
        install_capturing_external_egress_config(&MessageBusConfig {
            encoding,
            external_streams: Some(vec!["external.test".to_string()]),
            ..Default::default()
        })
    }

    fn reset_message_bus() {
        get_message_bus().borrow_mut().dispose();
        set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
    }

    fn assert_response_handler_is_consumed(response: &DataResponse) {
        reset_message_bus();

        let correlation_id = *response.correlation_id();
        let response_kind = response.kind();
        let calls = Rc::new(Cell::new(0));
        let handler_calls = calls.clone();
        register_response_handler(
            &correlation_id,
            TypedHandler::from(move |_response: &DataResponseEnvelope| {
                handler_calls.set(handler_calls.get() + 1);
            }),
        );

        send_response(&correlation_id, response);
        send_response(&correlation_id, response);

        assert_eq!(
            calls.get(),
            1,
            "{response_kind} response handler must only run once",
        );
        assert!(
            get_message_bus()
                .borrow()
                .get_response_handler(&correlation_id)
                .is_none(),
            "{response_kind} response handler must be removed after dispatch",
        );
        assert_eq!(
            get_message_bus().borrow().res_count(),
            2,
            "{response_kind} duplicate responses must still increment the response count",
        );
    }

    #[rstest]
    #[case("")]
    #[case("   ")]
    #[case("\t\n")]
    #[case("*")]
    #[case("mailbox.*")]
    #[case("mail?ox")]
    fn has_endpoint_returns_false_for_invalid_name(#[case] endpoint: &str) {
        reset_message_bus();

        assert!(!has_endpoint(endpoint));
    }

    #[rstest]
    #[case(SerializationEncoding::MsgPack)]
    #[case(SerializationEncoding::Json)]
    fn publish_quote_forwards_decodable_payload_to_external_egress(
        #[case] encoding: SerializationEncoding,
    ) {
        let publications = install_capturing_external_egress(encoding);
        let quote = QuoteTick::default();

        publish_quote("data.quotes.TEST".into(), &quote);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].topic, "data.quotes.TEST");

        let decoded: QuoteTick = match encoding {
            SerializationEncoding::MsgPack => rmp_serde::from_slice(&publications[0].payload)
                .expect("MsgPack payload must decode as QuoteTick"),
            SerializationEncoding::Json => serde_json::from_slice(&publications[0].payload)
                .expect("JSON payload must decode as QuoteTick"),
            SerializationEncoding::Sbe => {
                unreachable!("schema encodings are tested separately")
            }
        };
        let payload_value: serde_json::Value = match encoding {
            SerializationEncoding::MsgPack => rmp_serde::from_slice(&publications[0].payload)
                .expect("MsgPack payload must decode as a value"),
            SerializationEncoding::Json => serde_json::from_slice(&publications[0].payload)
                .expect("JSON payload must decode as a value"),
            SerializationEncoding::Sbe => {
                unreachable!("schema encodings are tested separately")
            }
        };
        assert_eq!(
            payload_value.get("type").and_then(|value| value.as_str()),
            Some("QuoteTick")
        );
        assert_eq!(decoded, quote);
        drop(publications);
        reset_message_bus();
    }

    fn assert_quote_round_trips(encoding: SerializationEncoding) {
        let publications = install_capturing_external_egress(encoding);
        let quote = QuoteTick::default();

        publish_quote("data.quotes.TEST".into(), &quote);

        let bus_message = {
            let publications = publications.borrow();
            assert_eq!(publications.len(), 1);
            assert_eq!(publications[0].payload_type, BusPayloadType::QuoteTick);
            assert_eq!(publications[0].encoding, encoding);
            BusMessage::with_str_topic(
                publications[0].topic.clone(),
                publications[0].payload_type,
                publications[0].payload.clone(),
                publications[0].encoding,
            )
        };
        publications.borrow_mut().clear();

        let received = Rc::new(RefCell::new(Vec::<QuoteTick>::new()));
        let received_handler = received.clone();
        let handler = TypedHandler::from(move |quote: &QuoteTick| {
            received_handler.borrow_mut().push(*quote);
        });
        subscribe_quotes("data.quotes.*".into(), handler, None);

        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::QuoteTick);
        republish_external_message(&bus_message).unwrap();

        assert_eq!(*received.borrow(), vec![quote]);
        assert!(
            publications.borrow().is_empty(),
            "republished message must not be forwarded back out externally"
        );
        reset_message_bus();
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_round_trips_quote(#[case] encoding: SerializationEncoding) {
        assert_quote_round_trips(encoding);
    }

    #[cfg(feature = "sbe")]
    #[rstest]
    fn republish_external_message_round_trips_quote_sbe() {
        assert_quote_round_trips(SerializationEncoding::Sbe);
    }

    fn assert_typed_external_round_trips<T>(
        encoding: SerializationEncoding,
        payload_type: BusPayloadType,
        topic: &str,
        value: T,
        publish: fn(MStr<Topic>, &T),
        subscribe: fn(MStr<Pattern>, TypedHandler<T>, Option<u32>),
        assert_received: impl Fn(&T, &T),
    ) where
        T: Clone + 'static,
    {
        let publications = install_capturing_external_egress(encoding);

        publish(topic.into(), &value);

        let bus_message = {
            let publications = publications.borrow();
            assert_eq!(publications.len(), 1);
            assert_eq!(publications[0].payload_type, payload_type);
            assert_eq!(publications[0].encoding, encoding);
            BusMessage::with_str_topic(
                publications[0].topic.clone(),
                publications[0].payload_type,
                publications[0].payload.clone(),
                publications[0].encoding,
            )
        };
        publications.borrow_mut().clear();

        let received = Rc::new(RefCell::new(Vec::<T>::new()));
        let received_handler = received.clone();
        let handler = TypedHandler::from(move |message: &T| {
            received_handler.borrow_mut().push(message.clone());
        });
        subscribe(topic.into(), handler, None);

        get_message_bus()
            .borrow_mut()
            .add_streaming_type(payload_type);
        republish_external_message(&bus_message).unwrap();

        let received = received.borrow();
        assert_eq!(received.len(), 1);
        assert_received(&received[0], &value);
        assert!(
            publications.borrow().is_empty(),
            "republished message must not be forwarded back out externally"
        );
        reset_message_bus();
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_account_state_routes_to_portfolio(
        #[case] encoding: SerializationEncoding,
    ) {
        let publications = install_capturing_external_egress(encoding);
        let state = nautilus_model::events::account::stubs::cash_account_state();
        publish_account_state("events.account.SIM-001".into(), &state);
        let captured = publications.borrow_mut().pop().unwrap();
        assert_eq!(captured.payload_type, BusPayloadType::AccountState);
        assert_eq!(captured.encoding, encoding);
        let message = BusMessage::with_str_topic(
            captured.topic,
            captured.payload_type,
            captured.payload,
            captured.encoding,
        );
        let applied = Rc::new(RefCell::new(Vec::new()));
        let handler = TypedHandler::from({
            let applied = Rc::clone(&applied);
            move |event: &AccountState| applied.borrow_mut().push(event.clone())
        });
        register_account_state_endpoint(MessagingSwitchboard::portfolio_update_account(), handler);
        let received = Rc::new(RefCell::new(Vec::new()));
        let subscriber = TypedHandler::from({
            let received = Rc::clone(&received);
            move |event: &AccountState| received.borrow_mut().push(event.clone())
        });
        subscribe_account_state("events.account.*".into(), subscriber, None);
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::AccountState);

        republish_external_message(&message).unwrap();

        assert_eq!(*applied.borrow(), vec![state]);
        assert!(
            received.borrow().is_empty(),
            "Portfolio owns account publication"
        );
        assert!(publications.borrow().is_empty());
        reset_message_bus();
    }

    fn assert_eq_ref<T>(actual: &T, expected: &T)
    where
        T: PartialEq + std::fmt::Debug,
    {
        assert_eq!(actual, expected);
    }

    fn assert_json_value_eq<T>(actual: &T, expected: &T)
    where
        T: serde::Serialize,
    {
        assert_eq!(
            serde_json::to_value(actual).expect("actual value must serialize"),
            serde_json::to_value(expected).expect("expected value must serialize"),
        );
    }

    fn assert_depth_market_eq(actual: &OrderBookDepth, expected: &OrderBookDepth) {
        assert_eq!(actual.instrument_id, expected.instrument_id);
        assert_eq!(actual.bid_counts, expected.bid_counts);
        assert_eq!(actual.ask_counts, expected.ask_counts);
        assert_eq!(actual.flags, expected.flags);
        assert_eq!(actual.sequence, expected.sequence);
        assert_eq!(actual.ts_event, expected.ts_event);
        assert_eq!(actual.ts_init, expected.ts_init);

        for (actual, expected) in actual.bids.iter().zip(expected.bids.iter()) {
            assert_eq!(actual.side, expected.side);
            assert_eq!(actual.price, expected.price);
            assert_eq!(actual.size, expected.size);
        }

        for (actual, expected) in actual.asks.iter().zip(expected.asks.iter()) {
            assert_eq!(actual.side, expected.side);
            assert_eq!(actual.price, expected.price);
            assert_eq!(actual.size, expected.size);
        }
    }

    fn mark_price_update() -> MarkPriceUpdate {
        MarkPriceUpdate::new(
            InstrumentId::from("AUDUSD.SIM"),
            Price::from("1.00010"),
            UnixNanos::from(1),
            UnixNanos::from(2),
        )
    }

    fn index_price_update() -> IndexPriceUpdate {
        IndexPriceUpdate::new(
            InstrumentId::from("AUDUSD.SIM"),
            Price::from("1.00020"),
            UnixNanos::from(3),
            UnixNanos::from(4),
        )
    }

    fn funding_rate_update() -> FundingRateUpdate {
        FundingRateUpdate::new(
            InstrumentId::from("AUDUSD.SIM"),
            Decimal::new(1, 4),
            Some(480),
            Some(UnixNanos::from(5)),
            UnixNanos::from(6),
            UnixNanos::from(7),
        )
    }

    #[cfg(feature = "sbe")]
    fn option_greeks() -> OptionGreeks {
        OptionGreeks {
            instrument_id: InstrumentId::from("BTC-30JUN23-40000-C.DERIBIT"),
            convention: GreeksConvention::PriceAdjusted,
            greeks: OptionGreekValues {
                delta: 0.525,
                gamma: 0.00032,
                vega: 12.25,
                theta: -0.72,
                rho: 0.18,
            },
            mark_iv: Some(0.0),
            bid_iv: None,
            ask_iv: Some(0.54),
            underlying_price: Some(41_500.25),
            open_interest: Some(0.0),
            ts_event: UnixNanos::from(20),
            ts_init: UnixNanos::from(21),
        }
    }

    fn portfolio_snapshot() -> PortfolioSnapshot {
        PortfolioSnapshot::new(
            AccountId::from("SIM-001"),
            AccountType::Cash,
            Some(Currency::USD()),
            vec![],
            vec![],
            vec![],
            vec![],
            vec![],
            None,
            false,
            vec![],
            vec![],
            vec![],
            UUID4::new(),
            UnixNanos::from(8),
            UnixNanos::from(9),
        )
    }

    fn position_event() -> PositionEvent {
        PositionEvent::PositionOpened(PositionOpened {
            trader_id: TraderId::from("TRADER-001"),
            strategy_id: StrategyId::from("S-001"),
            instrument_id: InstrumentId::from("AUDUSD.SIM"),
            position_id: PositionId::from("P-001"),
            account_id: AccountId::from("SIM-001"),
            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
            entry: OrderSide::Buy,
            side: PositionSide::Long,
            signed_qty: 100.into(),
            quantity: Quantity::from("100"),
            last_qty: Quantity::from("100"),
            last_px: Price::from("1.00000"),
            currency: Currency::USD(),
            avg_px_open: 1.into(),
            realized_pnl: None,
            event_id: UUID4::new(),
            ts_event: UnixNanos::from(10),
            ts_init: UnixNanos::from(11),
        })
    }

    #[cfg(feature = "defi")]
    fn defi_chain() -> Arc<Chain> {
        Arc::new(
            Chain::from_chain_id(42161)
                .expect("Arbitrum chain must be registered")
                .clone(),
        )
    }

    #[cfg(feature = "defi")]
    fn defi_dex() -> Arc<Dex> {
        let chain = Chain::from_chain_id(42161)
            .expect("Arbitrum chain must be registered")
            .clone();
        Arc::new(Dex::new(
            chain,
            DexType::UniswapV3,
            "0x1F98431c8aD98523631AE4a59f267346ea31F984",
            0,
            AmmType::CLAMM,
            "PoolCreated",
            "Swap",
            "Mint",
            "Burn",
            "Collect",
        ))
    }

    #[cfg(feature = "defi")]
    fn defi_pool() -> Pool {
        let chain = defi_chain();
        let dex = defi_dex();
        let rain = Token::new(
            chain.clone(),
            address!("0x25118290e6A5f4139381D072181157035864099d"),
            "RAIN".to_string(),
            "RAIN".to_string(),
            18,
        );
        let weth = Token::new(
            chain.clone(),
            address!("0x82aF49447D8a07e3bd95BD0d56f35241523fBab1"),
            "Wrapped Ether".to_string(),
            "WETH".to_string(),
            18,
        );
        let pool_address = address!("0xd13040d4fe917EE704158CfCB3338dCd2838B245");

        Pool::new(
            chain,
            dex,
            pool_address,
            PoolIdentifier::from_address(pool_address),
            0,
            rain,
            weth,
            Some(3000),
            Some(60),
            UnixNanos::from(12),
        )
    }

    #[cfg(feature = "defi")]
    fn defi_block() -> Block {
        Block::new(
            "0x0000000000000000000000000000000000000000000000000000000000000100".to_string(),
            "0x0000000000000000000000000000000000000000000000000000000000000099".to_string(),
            100,
            Ustr::from("0x0000000000000000000000000000000000000001"),
            30_000_000,
            21_000,
            UnixNanos::from(13),
            None,
        )
    }

    #[cfg(feature = "defi")]
    fn defi_transaction_hash() -> String {
        "0x1aa3506e78dd6e7e53986fa310c7ef1b7825042e19693c04eb56b2404067407b".to_string()
    }

    #[cfg(feature = "defi")]
    fn defi_liquidity_update() -> PoolLiquidityUpdate {
        let pool = defi_pool();
        PoolLiquidityUpdate::new(
            pool.chain.clone(),
            pool.dex.clone(),
            pool.instrument_id,
            pool.pool_identifier,
            PoolLiquidityUpdateType::Mint,
            100_000,
            defi_transaction_hash(),
            0,
            1,
            None,
            address!("0x5E325eDA8064b456f4781070C0738d849c824258"),
            100,
            U256::from(10),
            U256::from(20),
            -120,
            120,
            UnixNanos::from(14),
            UnixNanos::from(15),
        )
    }

    #[cfg(feature = "defi")]
    fn defi_collect() -> PoolFeeCollect {
        let pool = defi_pool();
        PoolFeeCollect::new(
            pool.chain.clone(),
            pool.dex.clone(),
            pool.instrument_id,
            pool.pool_identifier,
            100_000,
            defi_transaction_hash(),
            0,
            2,
            address!("0x5E325eDA8064b456f4781070C0738d849c824258"),
            10,
            20,
            -120,
            120,
            UnixNanos::from(16),
            UnixNanos::from(17),
        )
    }

    #[cfg(feature = "defi")]
    fn defi_flash() -> PoolFlash {
        let pool = defi_pool();
        PoolFlash::new(
            pool.chain.clone(),
            pool.dex.clone(),
            pool.instrument_id,
            pool.pool_identifier,
            100_000,
            defi_transaction_hash(),
            0,
            3,
            UnixNanos::from(18),
            UnixNanos::from(19),
            address!("0x1aa3506e78dd6e7e53986fa310c7ef1b7825042e"),
            address!("0x1aa3506e78dd6e7e53986fa310c7ef1b7825042e"),
            U256::from(100),
            U256::from(200),
            U256::from(101),
            U256::from(202),
        )
    }

    fn assert_publishable_json_msgpack_round_trips(encoding: SerializationEncoding) {
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::Instrument,
            "data.instruments.AUDUSD.SIM",
            InstrumentAny::CurrencyPair(audusd_sim()),
            publish_instrument,
            subscribe_instruments,
            assert_json_value_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OrderBookDeltas,
            "data.book.deltas.AAPL.XNAS",
            stub_deltas(),
            publish_deltas,
            subscribe_book_deltas,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OrderBookDepth,
            "data.book.depth.AAPL.XNAS",
            stub_depth10(),
            publish_depth,
            subscribe_book_depth,
            assert_depth_market_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::TradeTick,
            "data.trades.AUDUSD.SIM",
            TradeTick::default(),
            publish_trade,
            subscribe_trades,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::Bar,
            "data.bars.AUDUSD.SIM",
            Bar::default(),
            publish_bar,
            subscribe_bars,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::MarkPriceUpdate,
            "data.mark_prices.AUDUSD.SIM",
            mark_price_update(),
            publish_mark_price,
            subscribe_mark_prices,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::IndexPriceUpdate,
            "data.index_prices.AUDUSD.SIM",
            index_price_update(),
            publish_index_price,
            subscribe_index_prices,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::FundingRateUpdate,
            "data.funding_rates.AUDUSD.SIM",
            funding_rate_update(),
            publish_funding_rate,
            subscribe_funding_rates,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OptionGreeks,
            "data.option_greeks.AUDUSD.SIM",
            OptionGreeks::default(),
            publish_option_greeks,
            subscribe_option_greeks,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::PortfolioSnapshot,
            "events.portfolio.SIM-001",
            portfolio_snapshot(),
            publish_portfolio_snapshot,
            subscribe_portfolio_snapshot,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OrderEvent,
            "events.orders.SIM-001",
            OrderEventAny::Denied(OrderDeniedSpec::builder().build()),
            publish_order_event,
            subscribe_order_events,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::PositionEvent,
            "events.positions.SIM-001",
            position_event(),
            publish_position_event,
            subscribe_position_events,
            assert_json_value_eq,
        );
    }

    #[cfg(feature = "defi")]
    fn assert_publishable_defi_json_msgpack_round_trips(encoding: SerializationEncoding) {
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::Block,
            "data.defi.blocks.ARBITRUM",
            defi_block(),
            publish_defi_block,
            subscribe_defi_blocks,
            assert_json_value_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::Pool,
            "data.defi.pools.RAIN-WETH",
            defi_pool(),
            publish_defi_pool,
            subscribe_defi_pools,
            assert_json_value_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::PoolLiquidityUpdate,
            "data.defi.liquidity.RAIN-WETH",
            defi_liquidity_update(),
            publish_defi_liquidity,
            subscribe_defi_liquidity,
            assert_json_value_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::PoolFeeCollect,
            "data.defi.collects.RAIN-WETH",
            defi_collect(),
            publish_defi_collect,
            subscribe_defi_collects,
            assert_json_value_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::PoolFlash,
            "data.defi.flash.RAIN-WETH",
            defi_flash(),
            publish_defi_flash,
            subscribe_defi_flash,
            assert_json_value_eq,
        );
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_round_trips_publishable_payloads(
        #[case] encoding: SerializationEncoding,
    ) {
        assert_publishable_json_msgpack_round_trips(encoding);
    }

    #[cfg(feature = "defi")]
    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_round_trips_publishable_defi_payloads(
        #[case] encoding: SerializationEncoding,
    ) {
        assert_publishable_defi_json_msgpack_round_trips(encoding);
    }

    fn assert_custom_data_round_trips(encoding: SerializationEncoding) {
        let publications = install_capturing_external_egress(encoding);
        let custom = stub_custom_data(100, 42, None, Some("stub-id".to_string()));

        publish_any("data.custom.StubCustomData".into(), &custom);

        let bus_message = {
            let publications = publications.borrow();
            assert_eq!(publications.len(), 1);
            assert_eq!(
                publications[0].payload_type,
                BusPayloadType::Custom(Ustr::from("StubCustomData"))
            );
            assert_eq!(publications[0].encoding, encoding);
            BusMessage::with_str_topic(
                publications[0].topic.clone(),
                publications[0].payload_type,
                publications[0].payload.clone(),
                publications[0].encoding,
            )
        };
        publications.borrow_mut().clear();

        let received = Rc::new(RefCell::new(Vec::<CustomData>::new()));
        let received_handler = received.clone();
        subscribe_any(
            "data.custom.StubCustomData".into(),
            ShareableMessageHandler::from_typed(move |message: &CustomData| {
                received_handler.borrow_mut().push(message.clone());
            }),
            None,
        );

        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::Custom(Ustr::from("StubCustomData")));
        republish_external_message(&bus_message).unwrap();

        assert_eq!(*received.borrow(), vec![custom]);
        assert!(
            publications.borrow().is_empty(),
            "republished message must not be forwarded back out externally"
        );
        reset_message_bus();
    }

    fn assert_typed_payload_round_trips<T>(
        encoding: SerializationEncoding,
        payload_type: BusPayloadType,
        topic: &str,
        value: T,
    ) where
        T: serde::Serialize + Any + 'static,
    {
        let publications = install_capturing_external_stream_egress(encoding);
        let expected = serde_json::to_value(&value).expect("typed payload must map to JSON");
        let received = Rc::new(RefCell::new(Vec::<serde_json::Value>::new()));
        let received_handler = received.clone();
        subscribe_any(
            topic.into(),
            ShareableMessageHandler::from_typed(move |message: &T| {
                received_handler.borrow_mut().push(
                    serde_json::to_value(message).expect("received payload must map to JSON"),
                );
            }),
            None,
        );

        publish_any(topic.into(), &value);

        let bus_message = {
            let publications = publications.borrow();
            assert_eq!(publications.len(), 1);
            assert_eq!(publications[0].topic, topic);
            assert_eq!(publications[0].payload_type, payload_type);
            assert_eq!(publications[0].encoding, encoding);
            BusMessage::with_str_topic(
                topic,
                publications[0].payload_type,
                publications[0].payload.clone(),
                publications[0].encoding,
            )
        };
        received.borrow_mut().clear();
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(payload_type);

        republish_external_message(&bus_message).expect("typed payload must republish");

        assert_eq!(*received.borrow(), vec![expected]);
        assert_eq!(
            publications.borrow().len(),
            1,
            "republished typed payload must not echo to external egress"
        );
        reset_message_bus();
    }

    fn typed_processor_mapping<T>(
        payload_type: BusPayloadType,
        topic: &str,
        value: &T,
    ) -> serde_json::Value
    where
        T: serde::Serialize,
    {
        let message = BusMessage::with_str_topic(
            topic,
            payload_type,
            Bytes::from(serde_json::to_vec(value).expect("typed payload must serialize")),
            SerializationEncoding::Json,
        );
        let mut received = None;
        let mut processor = |_: &dyn Any, mapping: &serde_json::Value| {
            received = Some(mapping.clone());
            Ok(())
        };

        process_external_typed_message(&message, &mut processor)
            .expect("typed payload must process");

        received.expect("processor must receive the typed mapping")
    }

    #[rstest]
    fn typed_processor_mapping_identifies_subscription_action() {
        let instrument_id = InstrumentId::from("AUD/USD.SIM");
        let client_id = Some(ClientId::from("EXTERNAL"));
        let venue = Some(Venue::from("SIM"));
        let ts_init = UnixNanos::from(1_000_000_000);
        let subscribe = SubscribeCommand::Quotes(SubscribeQuotes::new(
            instrument_id,
            client_id,
            venue,
            UUID4::from("00000000-0000-4000-8000-000000000001"),
            ts_init,
            None,
            None,
        ));
        let unsubscribe = UnsubscribeCommand::Quotes(UnsubscribeQuotes::new(
            instrument_id,
            client_id,
            venue,
            UUID4::from("00000000-0000-4000-8000-000000000002"),
            ts_init,
            None,
            None,
        ));

        let mut subscribe_mapping = typed_processor_mapping(
            BusPayloadType::SubscribeCommand,
            "external.test",
            &subscribe,
        );
        let mut unsubscribe_mapping = typed_processor_mapping(
            BusPayloadType::UnsubscribeCommand,
            "external.test",
            &unsubscribe,
        );

        assert_eq!(
            subscribe_mapping
                .as_object_mut()
                .expect("subscribe payload must map to an object")
                .remove("payload_type"),
            Some(serde_json::json!("SubscribeCommand"))
        );
        assert_eq!(
            unsubscribe_mapping
                .as_object_mut()
                .expect("unsubscribe payload must map to an object")
                .remove("payload_type"),
            Some(serde_json::json!("UnsubscribeCommand"))
        );
        assert_eq!(
            subscribe_mapping,
            serde_json::to_value(subscribe).expect("subscribe command must map to JSON")
        );
        assert_eq!(
            unsubscribe_mapping,
            serde_json::to_value(unsubscribe).expect("unsubscribe command must map to JSON")
        );
        reset_message_bus();
    }

    #[rstest]
    fn typed_processor_publications_do_not_echo_to_external_egress() {
        let publications = install_capturing_external_stream_egress(SerializationEncoding::Json);
        let command = SubscribeCommand::Quotes(SubscribeQuotes::new(
            InstrumentId::from("AUD/USD.SIM"),
            Some(ClientId::from("EXTERNAL")),
            Some(Venue::from("SIM")),
            UUID4::from("00000000-0000-4000-8000-000000000001"),
            UnixNanos::from(1_000_000_000),
            None,
            None,
        ));

        publish_any("external.test".into(), &command);
        let message = {
            let publications = publications.borrow();
            assert_eq!(publications.len(), 1);
            BusMessage::with_str_topic(
                publications[0].topic.clone(),
                publications[0].payload_type,
                publications[0].payload.clone(),
                publications[0].encoding,
            )
        };
        let mut processor = |value: &dyn Any, _: &serde_json::Value| {
            let command = value
                .downcast_ref::<SubscribeCommand>()
                .expect("processor must receive a subscribe command");
            publish_any("external.test".into(), command);
            Ok(())
        };

        process_external_typed_message(&message, &mut processor)
            .expect("typed payload must process");
        assert_eq!(
            publications.borrow().len(),
            1,
            "processor publication must remain local"
        );

        publish_any("external.test".into(), &command);
        assert_eq!(
            publications.borrow().len(),
            2,
            "external egress must resume after processing"
        );
        reset_message_bus();
    }

    #[rstest]
    fn typed_processor_error_skips_internal_republishing() {
        let command = SubscribeCommand::Quotes(SubscribeQuotes::new(
            InstrumentId::from("AUD/USD.SIM"),
            Some(ClientId::from("EXTERNAL")),
            Some(Venue::from("SIM")),
            UUID4::from("00000000-0000-4000-8000-000000000001"),
            UnixNanos::from(1_000_000_000),
            None,
            None,
        ));
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_handler = received.clone();
        subscribe_any(
            "external.test".into(),
            ShareableMessageHandler::from_typed(move |command: &SubscribeCommand| {
                received_handler.borrow_mut().push(command.clone());
            }),
            None,
        );
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::SubscribeCommand);
        let message = BusMessage::with_str_topic(
            "external.test",
            BusPayloadType::SubscribeCommand,
            Bytes::from(serde_json::to_vec(&command).expect("typed payload must serialize")),
            SerializationEncoding::Json,
        );
        let mut processor =
            |_: &dyn Any, _: &serde_json::Value| anyhow::bail!("processor rejected payload");

        let error = process_external_typed_message(&message, &mut processor)
            .expect_err("processor failure must stop processing");

        assert_eq!(error.to_string(), "processor rejected payload");
        assert!(received.borrow().is_empty());
        reset_message_bus();
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn typed_payloads_round_trip_supported_encodings(#[case] encoding: SerializationEncoding) {
        let trader_id = TraderId::from("TRADER-001");
        let client_id = ClientId::from("EXTERNAL");
        let venue = Venue::from("SIM");
        let instrument_id = InstrumentId::from("AUD/USD.SIM");
        let account_id = AccountId::from("SIM-001");
        let venue_order_id = VenueOrderId::from("ORDER-001");
        let ts_init = UnixNanos::from(1_000_000_000);

        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::SubscribeCommand,
            "external.test",
            SubscribeCommand::Quotes(SubscribeQuotes::new(
                instrument_id,
                Some(client_id),
                Some(venue),
                UUID4::from("00000000-0000-4000-8000-000000000001"),
                ts_init,
                None,
                None,
            )),
        );
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::UnsubscribeCommand,
            "external.test",
            UnsubscribeCommand::Quotes(UnsubscribeQuotes::new(
                instrument_id,
                Some(client_id),
                Some(venue),
                UUID4::from("00000000-0000-4000-8000-000000000002"),
                ts_init,
                None,
                None,
            )),
        );
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::TradingCommand,
            "external.test",
            TradingCommand::CancelAllOrders(CancelAllOrders::new(
                trader_id,
                Some(client_id),
                StrategyId::from("STRATEGY-001"),
                instrument_id,
                Some(OrderSide::Buy),
                UUID4::from("00000000-0000-4000-8000-000000000003"),
                ts_init,
                None,
                Some(UUID4::from("00000000-0000-4000-8000-000000000004")),
            )),
        );
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::GenerateExecutionMassStatus,
            "external.test",
            GenerateExecutionMassStatus::new(
                trader_id,
                client_id,
                Some(venue),
                UUID4::from("00000000-0000-4000-8000-000000000005"),
                ts_init,
                None,
                Some(UUID4::from("00000000-0000-4000-8000-000000000006")),
            ),
        );

        let order_report = OrderStatusReport::new(
            account_id,
            instrument_id,
            Some(ClientOrderId::from("O-001")),
            venue_order_id,
            Some(OrderSide::Buy),
            OrderType::Limit,
            TimeInForce::Gtc,
            OrderStatus::Accepted,
            Quantity::from("10"),
            Quantity::from("2"),
            UnixNanos::from(1_000_000_001),
            UnixNanos::from(1_000_000_002),
            UnixNanos::from(1_000_000_003),
            Some(UUID4::from("00000000-0000-4000-8000-000000000007")),
        );
        let fill_report = FillReport::new(
            account_id,
            instrument_id,
            venue_order_id,
            TradeId::from("TRADE-001"),
            OrderSide::Buy,
            Quantity::from("2"),
            Price::from("1.25"),
            Money::from("0.10 USD"),
            LiquiditySide::Maker,
            Some(ClientOrderId::from("O-001")),
            Some(PositionId::from("P-001")),
            UnixNanos::from(1_000_000_004),
            UnixNanos::from(1_000_000_005),
            Some(UUID4::from("00000000-0000-4000-8000-000000000008")),
        );
        let position_report = PositionStatusReport::new(
            account_id,
            instrument_id,
            PositionSide::Long,
            Quantity::from("2"),
            UnixNanos::from(1_000_000_006),
            UnixNanos::from(1_000_000_007),
            Some(UUID4::from("00000000-0000-4000-8000-000000000009")),
            Some(PositionId::from("P-001")),
            Some(Decimal::new(125, 2)),
        );

        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::OrderStatusReport,
            "external.test",
            order_report.clone(),
        );
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::FillReport,
            "external.test",
            fill_report.clone(),
        );
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::PositionStatusReport,
            "external.test",
            position_report.clone(),
        );

        let mut mass_status = ExecutionMassStatus::new(
            client_id,
            account_id,
            venue,
            UnixNanos::from(1_000_000_008),
            Some(UUID4::from("00000000-0000-4000-8000-000000000010")),
        );
        mass_status.add_order_reports(vec![order_report]);
        mass_status.add_fill_reports(vec![fill_report]);
        mass_status.add_position_reports(vec![position_report]);
        assert_typed_payload_round_trips(
            encoding,
            BusPayloadType::ExecutionMassStatus,
            "external.test",
            mass_status,
        );
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_round_trips_custom_data(#[case] encoding: SerializationEncoding) {
        assert_custom_data_round_trips(encoding);
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_rejects_reserved_custom_metadata(
        #[case] encoding: SerializationEncoding,
    ) {
        let custom = stub_custom_data(317, 42, None, Some("catalog/317".to_string()));
        let mut envelope = serde_json::to_value(custom).unwrap();
        envelope["data_type"]["metadata"]["identifier"] = serde_json::json!("shadow");
        let payload = match encoding {
            SerializationEncoding::Json => serde_json::to_vec(&envelope).unwrap(),
            SerializationEncoding::MsgPack => rmp_serde::to_vec_named(&envelope).unwrap(),
            SerializationEncoding::Sbe => unreachable!("custom payloads require JSON or MsgPack"),
        };

        let payload_type = BusPayloadType::Custom(Ustr::from("StubCustomData"));
        let message = BusMessage::with_str_topic(
            "data.custom.StubCustomData",
            payload_type,
            Bytes::from(payload),
            encoding,
        );
        let received = Rc::new(RefCell::new(Vec::<CustomData>::new()));
        let received_handler = received.clone();
        subscribe_any(
            "data.custom.StubCustomData".into(),
            ShareableMessageHandler::from_typed(move |value: &CustomData| {
                received_handler.borrow_mut().push(value.clone());
            }),
            None,
        );

        republish_external_message(&message).unwrap();
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(payload_type);
        let error = republish_external_message(&message).unwrap_err();

        assert_eq!(
            error.root_cause().to_string(),
            "`identifier` is reserved for DataType catalog identity"
        );
        assert!(received.borrow().is_empty());
        reset_message_bus();
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn typed_processor_rejects_reserved_nested_custom_metadata(
        #[case] encoding: SerializationEncoding,
        #[values(false, true)] batch: bool,
    ) {
        let custom = stub_custom_data(317, 42, None, Some("catalog/317".to_string()));

        let data: Arc<dyn Any + Send + Sync> = if batch {
            Arc::new(vec![custom.clone()])
        } else {
            Arc::new(custom.clone())
        };

        let response = DataResponse::Data(CustomDataResponse {
            correlation_id: UUID4::new(),
            client_id: ClientId::from("CLIENT-001"),
            venue: Some(Venue::from("SIM")),
            data_type: custom.data_type,
            data,
            start: Some(UnixNanos::from(301)),
            end: Some(UnixNanos::from(313)),
            ts_init: UnixNanos::from(319),
            params: None,
        });

        let mut envelope = serde_json::to_value(response).unwrap();
        let item = if batch {
            &mut envelope["Data"]["data"]["Batch"][0]
        } else {
            &mut envelope["Data"]["data"]["Scalar"]
        };

        item["value"]["data_type"]["metadata"]["identifier"] = serde_json::Value::Null;

        let payload = match encoding {
            SerializationEncoding::Json => serde_json::to_vec(&envelope).unwrap(),
            SerializationEncoding::MsgPack => rmp_serde::to_vec_named(&envelope).unwrap(),
            SerializationEncoding::Sbe => unreachable!("custom payloads require JSON or MsgPack"),
        };

        let message = BusMessage::with_str_topic(
            "data.responses",
            BusPayloadType::DataResponse,
            Bytes::from(payload),
            encoding,
        );
        let mut processed = false;
        let error = process_external_typed_message(&message, &mut |_, _| {
            processed = true;
            Ok(())
        })
        .unwrap_err();

        assert!(
            error
                .root_cause()
                .to_string()
                .contains("`identifier` is reserved for DataType catalog identity")
        );
        assert!(!processed);
        reset_message_bus();
    }

    #[rstest]
    #[case(SerializationEncoding::Json)]
    #[case(SerializationEncoding::MsgPack)]
    fn republish_external_message_skips_unregistered_custom_payload(
        #[case] encoding: SerializationEncoding,
    ) {
        let envelope = serde_json::json!({
            "type": "UnregisteredCustomData",
            "data_type": {
                "type_name": "UnregisteredCustomData",
                "metadata": {"identifier": "shadow"},
            },
            "payload": {
                "value": 1,
            },
        });
        let payload = match encoding {
            SerializationEncoding::Json => {
                serde_json::to_vec(&envelope).expect("JSON envelope must serialize")
            }
            SerializationEncoding::MsgPack => {
                rmp_serde::to_vec_named(&envelope).expect("MsgPack envelope must serialize")
            }
            SerializationEncoding::Sbe => {
                unreachable!("schema encodings do not support custom payloads")
            }
        };
        let message = BusMessage::with_str_topic(
            "data.custom.UnregisteredCustomData",
            BusPayloadType::Custom(Ustr::from("UnregisteredCustomData")),
            Bytes::from(payload),
            encoding,
        );

        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::Custom(Ustr::from("UnregisteredCustomData")));
        republish_external_message(&message).unwrap();
        reset_message_bus();
    }

    #[rstest]
    fn republish_external_message_skips_untyped_custom_payload() {
        let message = BusMessage::with_str_topic(
            "events/control",
            BusPayloadType::Custom(Ustr::default()),
            Bytes::new(),
            SerializationEncoding::Json,
        );

        republish_external_message(&message).unwrap();
        reset_message_bus();
    }

    #[rstest]
    fn republish_external_message_skips_unregistered_streaming_type_before_decode() {
        let received = Rc::new(RefCell::new(Vec::<QuoteTick>::new()));
        let received_handler = received.clone();
        let handler = TypedHandler::from(move |quote: &QuoteTick| {
            received_handler.borrow_mut().push(*quote);
        });
        subscribe_quotes("data.quotes.*".into(), handler, None);

        let message = BusMessage::with_str_topic(
            "data.quotes.AUDUSD.SIM",
            BusPayloadType::QuoteTick,
            Bytes::from_static(b"not-json"),
            SerializationEncoding::Json,
        );

        republish_external_message(&message).unwrap();

        assert!(received.borrow().is_empty());
        reset_message_bus();
    }

    #[cfg(feature = "sbe")]
    fn assert_market_data_binary_round_trips(encoding: SerializationEncoding) {
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OrderBookDeltas,
            "data.book.deltas.AAPL.XNAS",
            stub_deltas(),
            publish_deltas,
            subscribe_book_deltas,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OrderBookDepth,
            "data.book.depth.AAPL.XNAS",
            stub_depth10(),
            publish_depth,
            subscribe_book_depth,
            assert_depth_market_eq,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::QuoteTick,
            "data.quotes.AUDUSD.SIM",
            QuoteTick::default(),
            publish_quote,
            subscribe_quotes,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::TradeTick,
            "data.trades.AUDUSD.SIM",
            TradeTick::default(),
            publish_trade,
            subscribe_trades,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::Bar,
            "data.bars.AUDUSD.SIM",
            Bar::default(),
            publish_bar,
            subscribe_bars,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::MarkPriceUpdate,
            "data.mark_prices.AUDUSD.SIM",
            mark_price_update(),
            publish_mark_price,
            subscribe_mark_prices,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::IndexPriceUpdate,
            "data.index_prices.AUDUSD.SIM",
            index_price_update(),
            publish_index_price,
            subscribe_index_prices,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::FundingRateUpdate,
            "data.funding_rates.AUDUSD.SIM",
            funding_rate_update(),
            publish_funding_rate,
            subscribe_funding_rates,
            assert_eq_ref,
        );
        assert_typed_external_round_trips(
            encoding,
            BusPayloadType::OptionGreeks,
            "data.option_greeks.BTC-30JUN23-40000-C.DERIBIT",
            option_greeks(),
            publish_option_greeks,
            subscribe_option_greeks,
            assert_eq_ref,
        );
    }

    #[cfg(feature = "sbe")]
    #[rstest]
    fn republish_external_message_round_trips_market_data_sbe() {
        assert_market_data_binary_round_trips(SerializationEncoding::Sbe);
    }

    #[rstest]
    #[case(BusPayloadType::AccountState)]
    fn republish_external_message_skips_unsupported_binary_payload(
        #[case] payload_type: BusPayloadType,
    ) {
        let received = Rc::new(RefCell::new(Vec::<serde_json::Value>::new()));
        let account_received = received.clone();
        subscribe_account_state(
            "events.unsupported.*".into(),
            TypedHandler::from(move |state: &AccountState| {
                account_received
                    .borrow_mut()
                    .push(serde_json::to_value(state).unwrap());
            }),
            None,
        );

        let message = BusMessage::with_str_topic(
            "events.unsupported.payload",
            payload_type,
            Bytes::from_static(b"malformed unsupported payload"),
            SerializationEncoding::Sbe,
        );
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(payload_type);
        republish_external_message(&message).unwrap();

        assert!(received.borrow().is_empty());
        reset_message_bus();
    }

    #[rstest]
    fn republish_external_message_errors_for_malformed_supported_payload() {
        let message = BusMessage::with_str_topic(
            "data.quotes.AUDUSD.SIM",
            BusPayloadType::QuoteTick,
            Bytes::from_static(b"not-json"),
            SerializationEncoding::Json,
        );

        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::QuoteTick);
        let error = republish_external_message(&message).unwrap_err();

        assert!(
            error
                .to_string()
                .contains("failed to decode JSON QuoteTick"),
            "{error:?}"
        );
        reset_message_bus();
    }

    #[rstest]
    fn republish_external_message_rejects_invalid_topic_and_processes_next_message() {
        let quote = QuoteTick::default();
        let payload = serde_json::to_vec(&quote).unwrap();
        let invalid_message = BusMessage::with_str_topic(
            "data.quotes.AUDUSD.SIM*",
            BusPayloadType::Custom(Ustr::from("UnregisteredCustomData")),
            Bytes::from(payload.clone()),
            SerializationEncoding::Json,
        );
        let valid_message = BusMessage::with_str_topic(
            "data.quotes.AUDUSD.SIM",
            BusPayloadType::QuoteTick,
            Bytes::from(payload),
            SerializationEncoding::Json,
        );
        let received = Rc::new(RefCell::new(Vec::<QuoteTick>::new()));
        let received_handler = received.clone();
        subscribe_quotes(
            "data.quotes.*".into(),
            TypedHandler::from(move |quote: &QuoteTick| {
                received_handler.borrow_mut().push(*quote);
            }),
            None,
        );
        get_message_bus()
            .borrow_mut()
            .add_streaming_type(BusPayloadType::QuoteTick);

        let error = republish_external_message(&invalid_message).unwrap_err();
        republish_external_message(&valid_message).unwrap();

        assert_eq!(
            format!("{error:#}"),
            "invalid external message topic: Topic `value` contained invalid characters, was data.quotes.AUDUSD.SIM*"
        );
        assert_eq!(*received.borrow(), vec![quote]);
        reset_message_bus();
    }

    #[cfg(feature = "sbe")]
    #[rstest]
    fn publish_quote_sbe_forwards_decodable_payload_to_external_egress() {
        let publications = install_capturing_external_egress(SerializationEncoding::Sbe);
        let quote = QuoteTick::default();

        publish_quote("data.quotes.TEST".into(), &quote);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].topic, "data.quotes.TEST");
        assert_eq!(
            QuoteTick::from_sbe(&publications[0].payload)
                .expect("SBE payload must decode as QuoteTick"),
            quote
        );
        drop(publications);
        reset_message_bus();
    }

    #[cfg(feature = "sbe")]
    #[rstest]
    fn publish_option_greeks_sbe_forwards_decodable_payload_to_external_egress() {
        let publications = install_capturing_external_egress(SerializationEncoding::Sbe);
        let greeks = option_greeks();

        publish_option_greeks("data.option_greeks.TEST".into(), &greeks);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].topic, "data.option_greeks.TEST");
        assert_eq!(
            OptionGreeks::from_sbe(&publications[0].payload)
                .expect("SBE payload must decode as OptionGreeks"),
            greeks
        );
        drop(publications);
        reset_message_bus();
    }

    #[cfg(not(feature = "sbe"))]
    #[rstest]
    fn publish_quote_sbe_without_feature_drops_payload() {
        let publications = install_capturing_external_egress(SerializationEncoding::Sbe);
        let quote = QuoteTick::default();

        publish_quote("data.quotes.TEST".into(), &quote);

        assert!(publications.borrow().is_empty());
        reset_message_bus();
    }

    #[rstest]
    fn publish_quote_external_egress_respects_filter_and_suppress_guard() {
        let publications = install_capturing_external_egress(SerializationEncoding::MsgPack);
        let quote = QuoteTick::default();

        get_message_bus()
            .borrow_mut()
            .set_types_filter(vec!["QuoteTick".to_string()]);
        publish_quote("data.quotes.FILTERED".into(), &quote);

        get_message_bus().borrow_mut().set_types_filter(Vec::new());
        {
            let _guard = SuppressExternalGuard::new();
            publish_quote("data.quotes.SUPPRESSED".into(), &quote);
        }
        publish_quote("data.quotes.PUBLISHED".into(), &quote);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].topic, "data.quotes.PUBLISHED");
        drop(publications);
        reset_message_bus();
    }

    #[rstest]
    #[case(None)]
    #[case(Some(Vec::new()))]
    fn publish_typed_report_external_egress_requires_external_streams(
        #[case] external_streams: Option<Vec<String>>,
    ) {
        let publications = install_capturing_external_egress_config(&MessageBusConfig {
            external_streams,
            ..Default::default()
        });

        publish_any(
            "reconciliation.raw.OrderStatusReport".into(),
            &order_status_report(),
        );

        assert!(publications.borrow().is_empty());
        reset_message_bus();
    }

    #[rstest]
    fn publish_typed_report_external_egress_respects_filter() {
        let publications = install_capturing_external_stream_egress(SerializationEncoding::Json);

        get_message_bus()
            .borrow_mut()
            .set_types_filter(vec!["OrderStatusReport".to_string()]);
        publish_any(
            "reconciliation.raw.OrderStatusReport".into(),
            &order_status_report(),
        );

        assert!(publications.borrow().is_empty());
        reset_message_bus();
    }

    fn order_status_report() -> OrderStatusReport {
        OrderStatusReport::new(
            AccountId::from("SIM-001"),
            InstrumentId::from("AUD/USD.SIM"),
            Some(ClientOrderId::from("O-001")),
            VenueOrderId::from("ORDER-001"),
            Some(OrderSide::Buy),
            OrderType::Limit,
            TimeInForce::Gtc,
            OrderStatus::Accepted,
            Quantity::from("10"),
            Quantity::from("2"),
            UnixNanos::from(1_000_000_001),
            UnixNanos::from(1_000_000_002),
            UnixNanos::from(1_000_000_003),
            Some(UUID4::from("00000000-0000-4000-8000-000000000001")),
        )
    }

    #[rstest]
    fn publish_quote_uses_market_data_encoding_override() {
        let publications = install_capturing_external_egress_config(&MessageBusConfig {
            encoding: SerializationEncoding::Json,
            encoding_market_data: Some(SerializationEncoding::MsgPack),
            ..Default::default()
        });
        let quote = QuoteTick::default();

        publish_quote("data.quotes.TEST".into(), &quote);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].encoding, SerializationEncoding::MsgPack);
        assert_eq!(
            rmp_serde::from_slice::<QuoteTick>(&publications[0].payload)
                .expect("MsgPack payload must decode as QuoteTick"),
            quote
        );
        drop(publications);
        reset_message_bus();
    }

    #[rstest]
    fn publish_custom_data_forwards_envelope_to_external_egress_and_respects_filter() {
        let publications = install_capturing_external_egress(SerializationEncoding::Json);
        let custom = stub_custom_data(100, 42, None, Some("stub-id".to_string()));

        publish_any("data.custom.StubCustomData".into(), &custom);

        get_message_bus()
            .borrow_mut()
            .set_types_filter(vec!["StubCustomData".to_string()]);
        publish_any("data.custom.FILTERED".into(), &custom);

        let publications = publications.borrow();
        assert_eq!(publications.len(), 1);
        assert_eq!(publications[0].topic, "data.custom.StubCustomData");

        let payload_value: serde_json::Value = serde_json::from_slice(&publications[0].payload)
            .expect("JSON payload must decode as a CustomData envelope");
        assert_eq!(
            payload_value.get("type").and_then(|value| value.as_str()),
            Some("StubCustomData")
        );
        assert_eq!(
            payload_value
                .pointer("/data_type/type_name")
                .and_then(|value| value.as_str()),
            Some("StubCustomData")
        );
        assert_eq!(
            payload_value
                .pointer("/data_type/identifier")
                .and_then(|value| value.as_str()),
            Some("stub-id")
        );
        assert_eq!(
            payload_value
                .pointer("/payload/value")
                .and_then(serde_json::Value::as_i64),
            Some(42)
        );
        drop(publications);
        reset_message_bus();
    }

    #[rstest]
    fn test_typed_quote_publish_subscribe_integration() {
        let msgbus = get_message_bus();
        let pub_count = msgbus.borrow().pub_count();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from(move |quote: &QuoteTick| {
            received_clone.borrow_mut().push(*quote);
        });

        subscribe_quotes("data.quotes.*".into(), handler, None);

        let quote = QuoteTick::default();
        publish_quote("data.quotes.TEST".into(), &quote);
        publish_quote("data.quotes.TEST".into(), &quote);

        assert_eq!(received.borrow().len(), 2);
        assert_eq!(msgbus.borrow().pub_count(), pub_count + 2);
    }

    #[rstest]
    fn test_typed_trade_publish_subscribe_integration() {
        let _msgbus = get_message_bus();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from(move |trade: &TradeTick| {
            received_clone.borrow_mut().push(*trade);
        });

        subscribe_trades("data.trades.*".into(), handler, None);

        let trade = TradeTick::default();
        publish_trade("data.trades.TEST".into(), &trade);

        assert_eq!(received.borrow().len(), 1);
    }

    #[rstest]
    fn test_typed_bar_publish_subscribe_integration() {
        let _msgbus = get_message_bus();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from(move |bar: &Bar| {
            received_clone.borrow_mut().push(*bar);
        });

        subscribe_bars("data.bars.*".into(), handler, None);

        let bar = Bar::default();
        publish_bar("data.bars.TEST".into(), &bar);

        assert_eq!(received.borrow().len(), 1);
    }

    #[rstest]
    fn test_typed_deltas_publish_subscribe_integration() {
        let _msgbus = get_message_bus();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from(move |deltas: &OrderBookDeltas| {
            received_clone.borrow_mut().push(deltas.clone());
        });

        subscribe_book_deltas("data.book.deltas.*".into(), handler, None);

        let instrument_id = InstrumentId::from("TEST.VENUE");
        let delta = OrderBookDelta::clear(instrument_id, 0, 1.into(), 2.into());
        let deltas = OrderBookDeltas::new(instrument_id, vec![delta]);
        publish_deltas("data.book.deltas.TEST".into(), &deltas);

        assert_eq!(received.borrow().len(), 1);
    }

    #[rstest]
    fn test_typed_unsubscribe_stops_delivery() {
        let _msgbus = get_message_bus();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from_with_id("unsub-test", move |quote: &QuoteTick| {
            received_clone.borrow_mut().push(*quote);
        });

        subscribe_quotes("data.quotes.UNSUB".into(), handler.clone(), None);

        let quote = QuoteTick::default();
        publish_quote("data.quotes.UNSUB".into(), &quote);
        assert_eq!(received.borrow().len(), 1);

        unsubscribe_quotes("data.quotes.UNSUB".into(), &handler);

        publish_quote("data.quotes.UNSUB".into(), &quote);
        assert_eq!(received.borrow().len(), 1);
    }

    #[rstest]
    fn test_typed_wildcard_pattern_matching() {
        let _msgbus = get_message_bus();
        let received = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();

        let handler = TypedHandler::from(move |quote: &QuoteTick| {
            received_clone.borrow_mut().push(*quote);
        });

        subscribe_quotes("data.quotes.WILD.*".into(), handler, None);

        let quote = QuoteTick::default();
        publish_quote("data.quotes.WILD.AAPL".into(), &quote);
        publish_quote("data.quotes.WILD.MSFT".into(), &quote);
        publish_quote("data.quotes.OTHER.AAPL".into(), &quote);

        assert_eq!(received.borrow().len(), 2);
    }

    #[rstest]
    fn test_typed_priority_ordering() {
        let _msgbus = get_message_bus();
        let order = Rc::new(RefCell::new(Vec::new()));

        let order1 = order.clone();
        let handler_low = TypedHandler::from_with_id("low-priority", move |_: &QuoteTick| {
            order1.borrow_mut().push("low");
        });

        let order2 = order.clone();
        let handler_high = TypedHandler::from_with_id("high-priority", move |_: &QuoteTick| {
            order2.borrow_mut().push("high");
        });

        subscribe_quotes("data.quotes.PRIO.*".into(), handler_low, Some(1));
        subscribe_quotes("data.quotes.PRIO.*".into(), handler_high, Some(10));

        let quote = QuoteTick::default();
        publish_quote("data.quotes.PRIO.TEST".into(), &quote);

        assert_eq!(*order.borrow(), vec!["high", "low"]);
    }

    #[rstest]
    fn test_typed_routing_isolation() {
        let _msgbus = get_message_bus();
        let quote_received = Rc::new(RefCell::new(false));
        let trade_received = Rc::new(RefCell::new(false));

        let qr = quote_received.clone();
        let quote_handler = TypedHandler::from(move |_: &QuoteTick| {
            *qr.borrow_mut() = true;
        });

        let tr = trade_received.clone();
        let trade_handler = TypedHandler::from(move |_: &TradeTick| {
            *tr.borrow_mut() = true;
        });

        subscribe_quotes("data.iso.*".into(), quote_handler, None);
        subscribe_trades("data.iso.*".into(), trade_handler, None);

        let quote = QuoteTick::default();
        publish_quote("data.iso.TEST".into(), &quote);

        assert!(*quote_received.borrow());
        assert!(!*trade_received.borrow());
    }

    #[rstest]
    fn test_send_data_allows_reentrant_topic_access() {
        use crate::msgbus::switchboard::get_quotes_topic;

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |data: Data| {
            let instrument_id = data.instrument_id();
            let _topic = get_quotes_topic(instrument_id);
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.data".into();
        register_data_endpoint(endpoint, handler);

        let quote = QuoteTick::default();
        send_data(endpoint, Data::Quote(quote));

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_send_trading_command_allows_reentrant_topic_access() {
        use nautilus_model::identifiers::{StrategyId, TraderId};

        use crate::{
            messages::execution::{TradingCommand, cancel::CancelAllOrders},
            msgbus::switchboard::get_trades_topic,
        };

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |cmd: TradingCommand| {
            let instrument_id = cmd.instrument_id();
            let _topic = get_trades_topic(instrument_id);
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.tradingCmd".into();
        register_trading_command_endpoint(endpoint, handler);

        let cmd = TradingCommand::CancelAllOrders(CancelAllOrders::new(
            TraderId::new("TESTER-001"),
            None,
            StrategyId::new("S-001"),
            InstrumentId::from("TEST.VENUE"),
            None,
            UUID4::new(),
            0.into(),
            None,
            None, // correlation_id
        ));
        send_trading_command(endpoint, cmd);

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_send_account_state_allows_reentrant_topic_access() {
        use nautilus_model::{enums::AccountType, identifiers::AccountId, types::Currency};

        use crate::msgbus::switchboard::get_quotes_topic;

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedHandler::from(move |_state: &AccountState| {
            let instrument_id = InstrumentId::from("TEST.VENUE");
            let _topic = get_quotes_topic(instrument_id);
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.accountState".into();
        register_account_state_endpoint(endpoint, handler);

        let state = AccountState::new(
            AccountId::new("SIM-001"),
            AccountType::Cash,
            vec![],
            vec![],
            true,
            UUID4::new(),
            0.into(),
            0.into(),
            Some(Currency::USD()),
        );
        send_account_state(endpoint, &state);

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_send_order_event_allows_reentrant_topic_access() {
        use crate::msgbus::switchboard::get_quotes_topic;

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |_event: OrderEventAny| {
            let instrument_id = InstrumentId::from("TEST.VENUE");
            let _topic = get_quotes_topic(instrument_id);
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.orderEvent".into();
        register_order_event_endpoint(endpoint, handler);

        let event = OrderEventAny::Denied(OrderDeniedSpec::builder().build());
        send_order_event(endpoint, event);

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_send_data_command_allows_reentrant_topic_access() {
        use crate::msgbus::switchboard::get_trades_topic;

        let msgbus = get_message_bus();
        let sent_count = msgbus.borrow().sent_count();
        let req_count = msgbus.borrow().req_count();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |_cmd: DataCommand| {
            let _topic = get_trades_topic(InstrumentId::from("TEST.VENUE"));
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.dataCmd".into();
        register_data_command_endpoint(endpoint, handler);

        let cmd = DataCommand::Subscribe(SubscribeCommand::Quotes(SubscribeQuotes::new(
            InstrumentId::from("TEST.VENUE"),
            Some(ClientId::new("SIM")),
            None,
            UUID4::new(),
            0.into(),
            None,
            None,
        )));
        send_data_command(endpoint, cmd);

        assert!(*topic_retrieved.borrow());
        assert_eq!(msgbus.borrow().sent_count(), sent_count + 1);
        assert_eq!(msgbus.borrow().req_count(), req_count);

        let request = DataCommand::Request(RequestCommand::Quotes(RequestQuotes::new(
            InstrumentId::from("TEST.VENUE"),
            None,
            None,
            None,
            Some(ClientId::new("SIM")),
            UUID4::new(),
            0.into(),
            None,
        )));
        send_data_command(endpoint, request);

        assert_eq!(msgbus.borrow().sent_count(), sent_count + 2);
        assert_eq!(msgbus.borrow().req_count(), req_count + 1);
    }

    #[rstest]
    fn test_send_data_request_without_endpoint_does_not_increment_counts() {
        let msgbus = get_message_bus();
        let sent_count = msgbus.borrow().sent_count();
        let req_count = msgbus.borrow().req_count();

        let request = DataCommand::Request(RequestCommand::Quotes(RequestQuotes::new(
            InstrumentId::from("MISSING.VENUE"),
            None,
            None,
            None,
            Some(ClientId::new("SIM")),
            UUID4::new(),
            0.into(),
            None,
        )));
        send_data_command("Missing.dataCmd".into(), request);

        assert_eq!(msgbus.borrow().sent_count(), sent_count);
        assert_eq!(msgbus.borrow().req_count(), req_count);
    }

    #[rstest]
    fn test_send_data_response_allows_reentrant_topic_access() {
        use nautilus_model::identifiers::ClientId;

        use crate::{
            messages::data::{DataResponse, QuotesResponse},
            msgbus::switchboard::get_quotes_topic,
        };

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |_resp: DataResponse| {
            let _topic = get_quotes_topic(InstrumentId::from("TEST.VENUE"));
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.dataResp".into();
        register_data_response_endpoint(endpoint, handler);

        let resp = DataResponse::Quotes(QuotesResponse {
            correlation_id: UUID4::new(),
            client_id: ClientId::new("SIM"),
            instrument_ids: vec![InstrumentId::from("TEST.VENUE")],
            data: (vec![]).into(),
            start: None,
            end: None,
            ts_init: 0.into(),
            params: None,
        });
        send_data_response(endpoint, resp);

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_send_response_increments_response_count() {
        use nautilus_model::identifiers::ClientId;

        use crate::messages::data::{DataResponse, QuotesResponse};

        let msgbus = get_message_bus();
        let res_count = msgbus.borrow().res_count();
        let resp = DataResponse::Quotes(QuotesResponse {
            correlation_id: UUID4::new(),
            client_id: ClientId::new("SIM"),
            instrument_ids: vec![InstrumentId::from("TEST.VENUE")],
            data: (vec![]).into(),
            start: None,
            end: None,
            ts_init: 0.into(),
            params: None,
        });

        send_response(&UUID4::new(), &resp);

        assert_eq!(msgbus.borrow().res_count(), res_count + 1);
    }

    #[rstest]
    fn test_send_response_consumes_handler_for_all_variants() {
        let client_id = ClientId::new("SIM");
        let instrument = audusd_sim();
        let instrument_id = instrument.id;
        let venue = Venue::new("SIM");
        let bar_type = BarType::from("AUD/USD.SIM-1-MINUTE-LAST-EXTERNAL");

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Data(CustomDataResponse::new(
            correlation_id,
            client_id,
            Some(venue),
            DataType::new("TestData", None, None),
            Vec::<u8>::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Instrument(Box::new(
            InstrumentResponse::new(
                correlation_id,
                client_id,
                instrument_id,
                InstrumentAny::CurrencyPair(instrument),
                None,
                None,
                UnixNanos::default(),
                None,
            ),
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Instruments(InstrumentsResponse::new(
            correlation_id,
            client_id,
            venue,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Book(BookResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            OrderBook::new(instrument_id, BookType::L2_MBP),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::BookDeltas(BookDeltasResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::BookDepth(BookDepthResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Quotes(QuotesResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Trades(TradesResponse::new(
            correlation_id,
            client_id,
            instrument_id,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::FundingRates(
            FundingRatesResponse::new(
                correlation_id,
                client_id,
                instrument_id,
                Vec::new(),
                None,
                None,
                UnixNanos::default(),
                None,
            ),
        ));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::OptionChainReferencePrice(
            OptionChainReferencePriceResponse::new(
                correlation_id,
                client_id,
                OptionSeriesId::new_derived(
                    venue,
                    Ustr::from("BTC"),
                    Ustr::from("BTC"),
                    UnixNanos::from(100),
                ),
                Some(Price::from("50123.45")),
                UnixNanos::default(),
                None,
            ),
        ));

        let correlation_id = UUID4::new();
        assert_response_handler_is_consumed(&DataResponse::Bars(BarsResponse::new(
            correlation_id,
            client_id,
            bar_type,
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        )));
    }

    #[rstest]
    fn test_send_response_allows_reentrant_same_correlation_registration() {
        reset_message_bus();

        let correlation_id = UUID4::new();
        let first_calls = Rc::new(Cell::new(0));
        let second_calls = Rc::new(Cell::new(0));
        let first_handler_calls = first_calls.clone();
        let second_handler_calls = second_calls.clone();
        register_response_handler(
            &correlation_id,
            TypedHandler::from(move |_response: &DataResponseEnvelope| {
                first_handler_calls.set(first_handler_calls.get() + 1);
                let second_handler_calls = second_handler_calls.clone();
                register_response_handler(
                    &correlation_id,
                    TypedHandler::from(move |_response: &DataResponseEnvelope| {
                        second_handler_calls.set(second_handler_calls.get() + 1);
                    }),
                );
            }),
        );

        let response = DataResponse::Quotes(QuotesResponse::new(
            correlation_id,
            ClientId::new("SIM"),
            InstrumentId::from("TEST.VENUE"),
            Vec::new(),
            None,
            None,
            UnixNanos::default(),
            None,
        ));

        send_response(&correlation_id, &response);
        assert_eq!(first_calls.get(), 1);
        assert_eq!(second_calls.get(), 0);
        assert!(
            get_message_bus()
                .borrow()
                .get_response_handler(&correlation_id)
                .is_some(),
        );

        send_response(&correlation_id, &response);
        assert_eq!(first_calls.get(), 1);
        assert_eq!(second_calls.get(), 1);
        assert!(
            get_message_bus()
                .borrow()
                .get_response_handler(&correlation_id)
                .is_none(),
        );
    }

    #[rstest]
    fn test_send_execution_report_allows_reentrant_topic_access() {
        use nautilus_model::{
            identifiers::{AccountId, ClientId, Venue},
            reports::ExecutionMassStatus,
        };

        use crate::{messages::execution::ExecutionReport, msgbus::switchboard::get_trades_topic};

        let _msgbus = get_message_bus();
        let topic_retrieved = Rc::new(RefCell::new(false));
        let topic_clone = topic_retrieved.clone();

        let handler = TypedIntoHandler::from(move |_report: ExecutionReport| {
            let _topic = get_trades_topic(InstrumentId::from("TEST.VENUE"));
            *topic_clone.borrow_mut() = true;
        });

        let endpoint: MStr<Endpoint> = "ReentrantTest.execReport".into();
        register_execution_report_endpoint(endpoint, handler);

        let report = ExecutionReport::MassStatus(Box::new(ExecutionMassStatus::new(
            ClientId::new("SIM"),
            AccountId::new("SIM-001"),
            Venue::new("TEST"),
            0.into(),
            None,
        )));
        send_execution_report(endpoint, report);

        assert!(*topic_retrieved.borrow());
    }

    #[rstest]
    fn test_order_event_handler_can_send_trading_command() {
        // Tests that a handler processing an order event can send a trading command
        // without causing a borrow conflict. This simulates the scenario where a
        // strategy's on_order_accepted() handler calls cancel_order().
        let _msgbus = get_message_bus();
        let command_sent = Rc::new(RefCell::new(false));
        let command_sent_clone = command_sent.clone();

        let cmd_received = Rc::new(RefCell::new(false));
        let cmd_received_clone = cmd_received.clone();
        let cmd_handler = TypedIntoHandler::from(move |_cmd: TradingCommand| {
            *cmd_received_clone.borrow_mut() = true;
        });
        let cmd_endpoint: MStr<Endpoint> = "ReentrantTest.execCmd".into();
        register_trading_command_endpoint(cmd_endpoint, cmd_handler);

        let event_handler = TypedIntoHandler::from(move |_event: OrderEventAny| {
            // Simulate strategy calling cancel_order from on_order_accepted
            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
                TraderId::new("TESTER-001"),
                None,
                StrategyId::new("S-001"),
                InstrumentId::from("TEST.VENUE"),
                Some(OrderSide::Buy),
                UUID4::new(),
                0.into(),
                None,
                None, // correlation_id
            ));
            send_trading_command(cmd_endpoint, command);
            *command_sent_clone.borrow_mut() = true;
        });

        let event_endpoint: MStr<Endpoint> = "ReentrantTest.orderEvt".into();
        register_order_event_endpoint(event_endpoint, event_handler);

        let event = OrderEventAny::Denied(OrderDeniedSpec::builder().build());
        send_order_event(event_endpoint, event);

        assert!(
            *command_sent.borrow(),
            "Order event handler should have run"
        );
        assert!(
            *cmd_received.borrow(),
            "Trading command should have been received"
        );
    }

    #[rstest]
    fn test_data_handler_can_send_data_command() {
        // Tests that a handler processing data can send a data command
        // without causing a borrow conflict.
        let _msgbus = get_message_bus();
        let command_sent = Rc::new(RefCell::new(false));
        let command_sent_clone = command_sent.clone();

        let cmd_received = Rc::new(RefCell::new(false));
        let cmd_received_clone = cmd_received.clone();
        let cmd_handler = TypedIntoHandler::from(move |_cmd: DataCommand| {
            *cmd_received_clone.borrow_mut() = true;
        });
        let cmd_endpoint: MStr<Endpoint> = "ReentrantTest.dataCmd2".into();
        register_data_command_endpoint(cmd_endpoint, cmd_handler);

        let data_handler = TypedIntoHandler::from(move |_data: Data| {
            let command = DataCommand::Subscribe(SubscribeCommand::Quotes(SubscribeQuotes::new(
                InstrumentId::from("TEST.VENUE"),
                Some(ClientId::new("SIM")),
                None,
                UUID4::new(),
                0.into(),
                None,
                None,
            )));
            send_data_command(cmd_endpoint, command);
            *command_sent_clone.borrow_mut() = true;
        });

        let data_endpoint: MStr<Endpoint> = "ReentrantTest.data2".into();
        register_data_endpoint(data_endpoint, data_handler);

        let quote = QuoteTick::default();
        send_data(data_endpoint, Data::Quote(quote));

        assert!(*command_sent.borrow(), "Data handler should have run");
        assert!(
            *cmd_received.borrow(),
            "Data command should have been received"
        );
    }

    #[rstest]
    fn test_trading_command_handler_can_send_order_event() {
        // Tests that a handler processing a trading command can send an order event
        // without causing a borrow conflict. This is the reverse direction of the
        // common re-entrancy scenario.
        let _msgbus = get_message_bus();
        let event_sent = Rc::new(RefCell::new(false));
        let event_sent_clone = event_sent.clone();

        let evt_received = Rc::new(RefCell::new(false));
        let evt_received_clone = evt_received.clone();
        let evt_handler = TypedIntoHandler::from(move |_event: OrderEventAny| {
            *evt_received_clone.borrow_mut() = true;
        });
        let evt_endpoint: MStr<Endpoint> = "ReentrantTest.orderEvt2".into();
        register_order_event_endpoint(evt_endpoint, evt_handler);

        let cmd_handler = TypedIntoHandler::from(move |_cmd: TradingCommand| {
            let event = OrderEventAny::Denied(OrderDeniedSpec::builder().build());
            send_order_event(evt_endpoint, event);
            *event_sent_clone.borrow_mut() = true;
        });

        let cmd_endpoint: MStr<Endpoint> = "ReentrantTest.execCmd2".into();
        register_trading_command_endpoint(cmd_endpoint, cmd_handler);

        let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
            TraderId::new("TESTER-001"),
            None,
            StrategyId::new("S-001"),
            InstrumentId::from("TEST.VENUE"),
            Some(OrderSide::Buy),
            UUID4::new(),
            0.into(),
            None,
            None, // correlation_id
        ));
        send_trading_command(cmd_endpoint, command);

        assert!(
            *event_sent.borrow(),
            "Trading command handler should have run"
        );
        assert!(
            *evt_received.borrow(),
            "Order event should have been received"
        );
    }

    #[rstest]
    fn test_nested_reentrant_calls() {
        // Tests deeply nested re-entrant calls: order event -> trading command -> order event.
        // This simulates a complex scenario where handlers chain multiple calls.
        let _msgbus = get_message_bus();
        let call_depth = Rc::new(RefCell::new(0u32));

        let final_received = Rc::new(RefCell::new(false));
        let final_received_clone = final_received.clone();
        let final_evt_handler = TypedIntoHandler::from(move |_event: OrderEventAny| {
            *final_received_clone.borrow_mut() = true;
        });
        let final_evt_endpoint: MStr<Endpoint> = "ReentrantTest.finalEvt".into();
        register_order_event_endpoint(final_evt_endpoint, final_evt_handler);

        let call_depth_clone2 = call_depth.clone();
        let mid_cmd_handler = TypedIntoHandler::from(move |_cmd: TradingCommand| {
            *call_depth_clone2.borrow_mut() += 1;
            let event = OrderEventAny::Denied(OrderDeniedSpec::builder().build());
            send_order_event(final_evt_endpoint, event);
        });
        let mid_cmd_endpoint: MStr<Endpoint> = "ReentrantTest.midCmd".into();
        register_trading_command_endpoint(mid_cmd_endpoint, mid_cmd_handler);

        let call_depth_clone1 = call_depth.clone();
        let init_evt_handler = TypedIntoHandler::from(move |_event: OrderEventAny| {
            *call_depth_clone1.borrow_mut() += 1;
            let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
                TraderId::new("TESTER-001"),
                None,
                StrategyId::new("S-001"),
                InstrumentId::from("TEST.VENUE"),
                Some(OrderSide::Buy),
                UUID4::new(),
                0.into(),
                None,
                None, // correlation_id
            ));
            send_trading_command(mid_cmd_endpoint, command);
        });
        let init_evt_endpoint: MStr<Endpoint> = "ReentrantTest.initEvt".into();
        register_order_event_endpoint(init_evt_endpoint, init_evt_handler);

        let event = OrderEventAny::Denied(OrderDeniedSpec::builder().build());
        send_order_event(init_evt_endpoint, event);

        assert_eq!(
            *call_depth.borrow(),
            2,
            "Both intermediate handlers should have run"
        );
        assert!(
            *final_received.borrow(),
            "Final event handler should have received the event"
        );
    }

    /// Recording tap used by the bus-tap registration tests. Stores every dispatched
    /// topic / endpoint plus a digest of the message so a test can assert the tap
    /// observed the exact dispatches it expected.
    #[derive(Default)]
    struct RecordingTap {
        publishes: RefCell<Vec<(String, std::any::TypeId)>>,
        sends: RefCell<Vec<(String, std::any::TypeId)>>,
        responses: RefCell<Vec<(UUID4, std::any::TypeId)>>,
    }

    impl RecordingTap {
        fn publish_topics(&self) -> Vec<String> {
            self.publishes
                .borrow()
                .iter()
                .map(|(t, _)| t.clone())
                .collect()
        }

        fn send_endpoints(&self) -> Vec<String> {
            self.sends.borrow().iter().map(|(e, _)| e.clone()).collect()
        }

        fn response_correlation_ids(&self) -> Vec<UUID4> {
            self.responses.borrow().iter().map(|(id, _)| *id).collect()
        }
    }

    impl BusTap for RecordingTap {
        fn on_publish(&self, topic: MStr<Topic>, message: &dyn std::any::Any) {
            self.publishes
                .borrow_mut()
                .push((topic.to_string(), message.type_id()));
        }

        fn on_send(&self, endpoint: MStr<Endpoint>, message: &dyn std::any::Any) {
            self.sends
                .borrow_mut()
                .push((endpoint.to_string(), message.type_id()));
        }

        fn on_response(&self, correlation_id: &UUID4, message: &dyn std::any::Any) {
            self.responses
                .borrow_mut()
                .push((*correlation_id, message.type_id()));
        }
    }

    #[rstest]
    fn try_publish_any_dispatches_handler_and_tap() {
        let msgbus = Rc::new(RefCell::new(MessageBus::default()));
        set_message_bus(msgbus.clone());
        clear_bus_tap();

        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let topic = "data.any.try.test";
        let (handler, checker) = get_call_check_handler(None);
        let pub_count = msgbus.borrow().pub_count();
        subscribe_any(topic.into(), handler, None);

        let payload: u32 = 42;
        let published = try_publish_any(topic.into(), &payload);

        clear_bus_tap();

        assert!(published);
        assert!(checker.was_called());
        assert_eq!(msgbus.borrow().pub_count(), pub_count + 1);
        assert_eq!(tap.publish_topics(), vec![topic]);
    }

    #[rstest]
    fn try_publish_any_without_registered_bus_returns_false() {
        let published = thread::spawn(|| {
            let payload: u32 = 42;
            try_publish_any("data.any.no-bus.test".into(), &payload)
        })
        .join()
        .expect("thread should join");

        assert!(!published);
    }

    #[rstest]
    fn try_publish_any_with_borrowed_bus_returns_false_without_tap() {
        let msgbus = Rc::new(RefCell::new(MessageBus::default()));
        set_message_bus(msgbus.clone());
        clear_bus_tap();

        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let bus_borrow = msgbus.borrow_mut();
        let payload: u32 = 42;
        let published = try_publish_any("data.any.borrowed.test".into(), &payload);
        drop(bus_borrow);

        clear_bus_tap();

        assert!(!published);
        assert_eq!(msgbus.borrow().pub_count(), 0);
        assert!(tap.publish_topics().is_empty());
    }

    #[rstest]
    fn set_bus_tap_then_publish_typed_invokes_tap() {
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let quote = QuoteTick::default();
        publish_quote("data.quotes.tap.test".into(), &quote);

        clear_bus_tap();

        assert_eq!(tap.publish_topics(), vec!["data.quotes.tap.test"]);
    }

    #[rstest]
    fn set_bus_tap_then_publish_any_invokes_tap() {
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let payload: u32 = 42;
        publish_any("data.any.tap.test".into(), &payload);

        clear_bus_tap();

        assert_eq!(tap.publish_topics(), vec!["data.any.tap.test"]);
    }

    #[rstest]
    fn set_bus_tap_then_send_any_value_invokes_tap() {
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let payload: u32 = 7;
        send_any_value("endpoint.send.any.value.test".into(), &payload);

        clear_bus_tap();

        assert_eq!(tap.send_endpoints(), vec!["endpoint.send.any.value.test"],);
    }

    #[rstest]
    fn send_any_variants_update_tap_and_count_before_handler() {
        let msgbus = Rc::new(RefCell::new(MessageBus::default()));
        set_message_bus(msgbus.clone());
        clear_bus_tap();

        let endpoint: MStr<Endpoint> = "endpoint.send.any.order.test".into();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let observations = Rc::new(RefCell::new(Vec::new()));
        let observations_clone = observations.clone();
        let handler = ShareableMessageHandler::from_any(move |message| {
            let value = *message.downcast_ref::<u32>().unwrap();
            observations_clone.borrow_mut().push((
                value,
                tap.send_endpoints(),
                get_message_bus().borrow().sent_count(),
            ));
        });
        register_any(endpoint, handler);

        send_any(endpoint, &11_u32);
        send_any_value(endpoint, &22_u32);

        clear_bus_tap();

        assert_eq!(
            *observations.borrow(),
            vec![
                (11, vec![endpoint.to_string()], 1),
                (22, vec![endpoint.to_string(), endpoint.to_string()], 2),
            ]
        );
        assert_eq!(msgbus.borrow().sent_count(), 2);
    }

    #[rstest]
    fn set_bus_tap_then_send_endpoint_owned_invokes_tap() {
        // send_trading_command and the other owned send functions reach the tap
        // through send_endpoint_owned_counted. Without this site instrumented, real
        // production order commands would bypass the audit log.
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let cancel_all = CancelAllOrders::new(
            TraderId::from("TRADER-001"),
            Some(ClientId::from("BINANCE")),
            StrategyId::from("S-001"),
            InstrumentId::from("ETHUSDT-PERP.BINANCE"),
            Some(OrderSide::Buy),
            UUID4::new(),
            nautilus_core::UnixNanos::from(1),
            None,
            None, // correlation_id
        );
        send_trading_command(
            "endpoint.send.trading.command.test".into(),
            TradingCommand::CancelAllOrders(cancel_all),
        );

        clear_bus_tap();

        assert_eq!(
            tap.send_endpoints(),
            vec!["endpoint.send.trading.command.test"],
        );
    }

    #[rstest]
    fn set_bus_tap_then_send_endpoint_ref_invokes_tap() {
        // send_quote and the other typed-reference send functions reach the tap through
        // send_endpoint_ref. Mirrors the owned path coverage.
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let quote = QuoteTick::default();
        send_quote("endpoint.send.quote.test".into(), &quote);

        clear_bus_tap();

        assert_eq!(tap.send_endpoints(), vec!["endpoint.send.quote.test"]);
    }

    #[rstest]
    fn has_quote_endpoint_returns_registration_state() {
        let _msgbus = get_message_bus();
        let endpoint: MStr<Endpoint> = "endpoint.has.quote.registered".into();

        assert!(!has_quote_endpoint(endpoint));

        let handler = TypedHandler::from_with_id(endpoint, |_quote: &QuoteTick| {});
        register_quote_endpoint(endpoint, handler);

        assert!(has_quote_endpoint(endpoint));
    }

    #[rstest]
    fn set_bus_tap_then_send_response_invokes_tap() {
        reset_message_bus();
        clear_bus_tap();
        let msgbus = get_message_bus();
        let initial_response_count = msgbus.borrow().res_count();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());

        let correlation_id = UUID4::new();
        let handler_calls = Rc::new(Cell::new(0));
        let handler_calls_clone = handler_calls.clone();
        register_response_handler(
            &correlation_id,
            TypedHandler::from(move |_resp: &DataResponseEnvelope| {
                handler_calls_clone.set(handler_calls_clone.get() + 1);
            }),
        );

        let response = DataResponse::Quotes(QuotesResponse {
            correlation_id,
            client_id: ClientId::new("SIM"),
            instrument_ids: vec![InstrumentId::from("TEST.VENUE")],
            data: (vec![]).into(),
            start: None,
            end: None,
            ts_init: 0.into(),
            params: None,
        });
        send_response(&correlation_id, &response);
        send_response(&correlation_id, &response);

        clear_bus_tap();

        assert_eq!(handler_calls.get(), 1);
        assert_eq!(
            tap.response_correlation_ids(),
            vec![correlation_id, correlation_id],
        );
        assert_eq!(msgbus.borrow().res_count(), initial_response_count + 2);
        assert!(
            msgbus
                .borrow()
                .get_response_handler(&correlation_id)
                .is_none(),
        );
    }

    #[rstest]
    fn clear_bus_tap_prevents_subsequent_dispatches_from_invoking_tap() {
        let _msgbus = get_message_bus();
        let tap = Rc::new(RecordingTap::default());
        set_bus_tap(tap.clone());
        clear_bus_tap();

        let quote = QuoteTick::default();
        publish_quote("data.quotes.after.clear".into(), &quote);
        send_quote("endpoint.send.after.clear".into(), &quote);

        let correlation_id = UUID4::new();
        register_response_handler(
            &correlation_id,
            TypedHandler::from(|_resp: &DataResponseEnvelope| {}),
        );
        let response = DataResponse::Quotes(QuotesResponse {
            correlation_id,
            client_id: ClientId::new("SIM"),
            instrument_ids: vec![InstrumentId::from("TEST.VENUE")],
            data: (vec![]).into(),
            start: None,
            end: None,
            ts_init: 0.into(),
            params: None,
        });
        send_response(&correlation_id, &response);

        assert!(tap.publish_topics().is_empty());
        assert!(tap.send_endpoints().is_empty());
        assert!(tap.response_correlation_ids().is_empty());
    }

    #[rstest]
    fn dispatch_with_no_tap_installed_is_a_noop() {
        // A fresh thread starts with BUS_TAP=None; dispatches must not panic and must
        // not allocate any tap state. Sanity check that the Option::None branch in
        // dispatch_tap_* is hit cleanly.
        let _msgbus = get_message_bus();

        let quote = QuoteTick::default();
        publish_quote("data.quotes.no.tap".into(), &quote);
        send_quote("endpoint.no.tap".into(), &quote);
    }

    struct ReinstallTap;

    impl BusTap for ReinstallTap {
        fn on_publish(&self, _topic: MStr<Topic>, _message: &dyn std::any::Any) {
            // Replace ourselves mid-dispatch; must not deadlock the RefCell
            set_bus_tap(Rc::new(NoopTap));
        }

        fn on_send(&self, _endpoint: MStr<Endpoint>, _message: &dyn std::any::Any) {}
    }

    struct NoopTap;

    impl BusTap for NoopTap {
        fn on_publish(&self, _topic: MStr<Topic>, _message: &dyn std::any::Any) {}
        fn on_send(&self, _endpoint: MStr<Endpoint>, _message: &dyn std::any::Any) {}
    }

    #[rstest]
    fn reentrant_set_bus_tap_during_dispatch_does_not_panic() {
        // dispatch_tap_publish clones the Rc out of BUS_TAP before invoking on_publish,
        // so a tap whose on_publish reinstalls a different tap must not panic on the
        // cell. The replaced tap stays alive through the cloned Rc until the dispatch
        // returns.
        let _msgbus = get_message_bus();
        set_bus_tap(Rc::new(ReinstallTap));

        let quote = QuoteTick::default();
        publish_quote("data.quotes.reentrant".into(), &quote);

        clear_bus_tap();
    }
}