nautilus_hyperliquid/websocket/

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

//! WebSocket execution dispatch for the Hyperliquid execution client.
//!
//! Implements the two-tier execution dispatch contract from
//! `docs/developer_guide/adapters.md#tracked-and-external-execution-updates`:
//!
//! 1. The execution client registers an [`OrderContext`] in [`WsDispatchState`]
//!    when it submits an order, and refreshes the cached venue order id when a
//!    modify is sent so the WebSocket consumer can detect cancel-replace.
//! 2. Incoming [`OrderStatusReport`] and [`FillReport`] messages are routed
//!    through [`dispatch_order_event`] and [`dispatch_order_fill`].
//!    For tracked orders these build typed [`OrderEventAny`] events and emit
//!    them via [`ExecutionEventEmitter::send_order_event`]. For untracked /
//!    external orders the dispatch falls back to forwarding the raw report.
//!
//! The dispatch state lives in an `Arc<WsDispatchState>` shared between the
//! main client task (which registers contexts at submission time) and the
//! spawned WebSocket consumer task.
//!
//! # GH-3827 cancel-replace handling
//!
//! Hyperliquid implements `modify` as a cancel-and-replace: the venue emits an
//! `ACCEPTED(new_voi)` together with a `CANCELED(old_voi)` under the same
//! `client_order_id`. The dispatch detects the replacement leg by comparing
//! `report.venue_order_id` to the last cached value, promotes it to an
//! `OrderUpdated` event, and suppresses the stale cancel so strategies never
//! observe a spurious termination.
//!
//! Each in-flight modify is tracked as an intent in a per-order chain (keyed
//! on `client_order_id`), pushed by `modify_order` before the HTTP call. An
//! intent lets dispatch skip an early `CANCELED(old_voi)` that arrives before
//! the replacement `ACCEPTED(new_voi)`, regardless of whether the WS message
//! races ahead of the HTTP response. Rapid repeated modifies under one stable
//! CLOID queue as a chain so a later modify cannot overwrite an earlier
//! intent's old-leg suppression, and a failed modify clears only its own
//! generation (leaving newer intents intact). The front intent is claimed on
//! promotion, advancing the next intent's old leg to the promoted replacement;
//! a rejected front reparents the next intent to the same still-live leg.
//!
//! A `CANCELED` for the bound leg while an intent targets that leg may instead
//! come from a user or venue cancel, which makes the modify fail. Dispatch holds
//! that cancel on the chain: a promotion drops it as the modify's own cancel
//! leg, and clearing the last intent on its leg releases it to the WebSocket
//! consumer loop, which applies it as `OrderCanceled`.
//!
//! A fill carrying the replacement `venue_order_id` during an in-flight modify
//! promotes the binding directly (the same `OrderUpdated` path as the
//! replacement `ACCEPTED`), so a dropped `ACCEPTED` does not strand the fill.
//! A fill is buffered into [`WsDispatchState::buffered_fills`] only when the
//! context has no price to promote with; `handle_accepted` drains the buffer
//! on the replacement `ACCEPTED`. A delayed earlier-leg fill during a chained
//! modify is a known limitation. See GH-3972.
//!
//! When neither the replacement `ACCEPTED` nor a fill arrives, a query that
//! resolves the replacement by `cloid` promotes the binding the same way via
//! [`promote_replacement_from_query`], so a dropped `ACCEPTED` with no fill
//! cannot leave the order bound to the canceled leg.

use std::{
    collections::VecDeque,
    sync::atomic::{AtomicBool, Ordering},
};

use ahash::AHashMap;
use dashmap::{DashMap, DashSet};
use nautilus_common::cache::fifo::FifoCache;
use nautilus_core::{UUID4, UnixNanos};
use nautilus_live::{ExecutionEventEmitter, execution::context::OrderContext};
use nautilus_model::{
    enums::{OrderStatus, OrderType},
    events::{
        OrderAccepted, OrderCanceled, OrderEventAny, OrderExpired, OrderFilled, OrderRejected,
        OrderTriggered, OrderUpdated,
    },
    identifiers::{AccountId, ClientOrderId, TradeId, VenueOrderId},
    orders::LIMIT_ORDER_TYPES,
    reports::{FillReport, OrderStatusReport},
    types::{Price, Quantity},
};
use parking_lot::Mutex;
use ustr::Ustr;

use crate::{
    common::consts::HYPERLIQUID_POST_ONLY_WOULD_MATCH,
    http::models::HyperliquidExchangePlaceOrderRequest,
};

pub const DEDUP_CAPACITY: usize = 10_000;

/// Maximum in-flight modify intents tracked per order. Rapid repricing rarely
/// queues more than one or two unacknowledged modifies at once; the cap bounds
/// memory if replacement acks stall. On overflow the oldest intent is evicted.
pub const MAX_PENDING_MODIFY_INTENTS: usize = 32;

/// A single in-flight Hyperliquid modify awaiting its replacement leg.
///
/// Rapid repeated modifies under one stable CLOID queue as a chain of intents
/// so a later modify cannot overwrite an earlier pending old-leg marker, and a
/// failed modify clears only its own generation rather than a newer one's
/// state. Each intent carries the venue leg it cancel-replaces
/// (`old_venue_order_id`), the user-intended absolute total quantity, and the
/// exact request sent (used to size a corrective reduce).
#[derive(Debug, Clone)]
pub struct ModifyIntent {
    /// Monotonic per-order generation, used to clear a specific intent on failure.
    pub generation: u64,
    /// Venue order id this modify cancel-replaces, once known.
    pub old_venue_order_id: Option<VenueOrderId>,
    /// User-intended absolute total quantity for the replacement.
    pub target_qty: Quantity,
    /// Exact venue request sent, used to size a corrective reduce.
    pub sent_request: Option<HyperliquidExchangePlaceOrderRequest>,
}

/// A corrective reduce queued for the WebSocket consumer loop to post.
///
/// The reduce registers its own modify intent, so a failed post clears only
/// that intent and leaves other modifies of the order queued.
#[derive(Debug, Clone)]
pub struct CorrectiveReduce {
    /// Venue order id of the oversized replacement the reduce modifies.
    pub oid: u64,
    /// Generation of the modify intent the reduce registered.
    pub generation: u64,
    /// Reduced request to send.
    pub request: HyperliquidExchangePlaceOrderRequest,
}

/// Bounded FIFO chain of in-flight modify intents for one order.
///
/// The venue processes chained modifies in submission order, so the front
/// (oldest) intent is the next to promote; on promotion the next intent's old
/// leg advances to the replacement just accepted. `held_cancel` keeps a
/// cancel of a targeted leg until the chain shows whether a modify caused it.
#[derive(Debug, Default)]
struct ModifyChain {
    intents: VecDeque<ModifyIntent>,
    next_generation: u64,
    held_cancel: Option<OrderStatusReport>,
}

impl ModifyChain {
    fn targets(&self, venue_order_id: VenueOrderId) -> bool {
        self.intents
            .iter()
            .any(|intent| intent.old_venue_order_id == Some(venue_order_id))
    }

    // Once no queued modify can cancel-replace the leg, its cancel came from elsewhere
    fn release_held_cancel(&mut self) -> Option<OrderStatusReport> {
        let venue_order_id = self.held_cancel.as_ref()?.venue_order_id;

        if self.targets(venue_order_id) {
            return None;
        }

        self.held_cancel.take()
    }
}

/// Per-client dispatch state shared between order submission and the
/// WebSocket consumer task.
///
/// Tracks which orders were submitted through this client (so we can route
/// venue events to typed [`OrderEventAny`] emissions for tracked orders and
/// fall back to reports for external orders), provides cross-stream dedup
/// for `OrderAccepted`, `OrderTriggered`, and `OrderFilled` emissions, and
/// carries the GH-3827 cancel-replace state (`cached_venue_order_ids` and
/// `pending_modify_keys`). Cancels released from a modify chain queue in
/// `released_cancels` until the WebSocket consumer loop applies them. A rejection
/// with undispatched fills waits for them, then releases the unfilled remainder
/// through the same cancel queue. Fill totals are retained per venue order ID so
/// fills on an earlier leg cannot satisfy a later leg's reported filled quantity.
#[derive(Debug)]
pub struct WsDispatchState {
    /// Tracked orders keyed by full Nautilus [`ClientOrderId`].
    ///
    /// The dispatch functions read the context to build typed order events for
    /// tracked orders without needing access to the engine cache (which is
    /// `!Send` and unreachable from the spawned WebSocket task). `quantity` and
    /// `price` are refreshed from subsequent status reports so a cancel-replace
    /// `ACCEPTED` that omits `price` can still produce an `OrderUpdated`
    /// carrying an accurate value.
    pub order_contexts: DashMap<ClientOrderId, OrderContext>,
    /// Client order IDs for which an `OrderAccepted` event has been emitted.
    pub emitted_accepted: DashSet<ClientOrderId>,
    emitted_triggered: DashSet<ClientOrderId>,
    /// Tracked submissions whose POST response has not resolved yet.
    pending_submissions: DashSet<ClientOrderId>,
    /// Submission-time rejections held until the POST path can preserve its
    /// more detailed venue error string.
    pending_submission_rejections: DashMap<ClientOrderId, OrderStatusReport>,
    /// Client order IDs that have reached the filled terminal state.
    ///
    /// Retained past `cleanup_terminal` so that late replay of the same
    /// status or fill does not re-emit events.
    pub filled_orders: DashSet<ClientOrderId>,
    /// Trade IDs for which an `OrderFilled` event has been emitted.
    ///
    /// Bounded FIFO dedup to bound memory while keeping recent trade ids
    /// deduped across reconnects.
    emitted_trades: Mutex<FifoCache<TradeId, DEDUP_CAPACITY>>,
    /// Raw Hyperliquid CLOIDs whose terminal state has already been dispatched.
    terminal_cloids: Mutex<FifoCache<Ustr, DEDUP_CAPACITY>>,
    closed_venue_order_ids: Mutex<FifoCache<VenueOrderId, DEDUP_CAPACITY>>,
    /// Last venue order id observed for a tracked client order id.
    ///
    /// Populated on the first `OrderAccepted` and refreshed on every
    /// cancel-replace promotion. A later `ACCEPTED` with a different venue
    /// order id under the same client order id is treated as the
    /// replacement leg of a Hyperliquid modify and emitted as `OrderUpdated`.
    pub cached_venue_order_ids: DashMap<ClientOrderId, VenueOrderId>,
    /// Per-order chain of in-flight modify intents, keyed by `client_order_id`.
    ///
    /// Rapid repeated modifies under one stable CLOID queue as a chain so a
    /// later modify cannot overwrite an earlier pending old-leg marker, and a
    /// failed modify clears only its own generation rather than a newer one's
    /// state. Populated by `modify_order` before the HTTP call so the WS cancel
    /// handler sees an intent even when `CANCELED(old_voi)` arrives before the
    /// HTTP response. A `CANCELED(old_voi)` matching any queued intent's old
    /// leg is suppressed so the later `ACCEPTED(new_voi)` can flow through the
    /// `OrderUpdated` path; the front intent is claimed on promotion and the
    /// next intent's old leg advances to the promoted replacement.
    pending_modify_chains: DashMap<ClientOrderId, ModifyChain>,
    /// `FillReport`s buffered only when a cancel-replace fill cannot be promoted
    /// (the context carries no price); drained by the cancel-replace branch of
    /// `handle_accepted`. The common path promotes on the fill instead. See
    /// GH-3972.
    pub buffered_fills: DashMap<ClientOrderId, Vec<FillReport>>,
    /// Cumulative filled quantity per tracked order. Compared against
    /// `OrderContext::quantity` to decide when to clean up tracked state.
    pub order_filled_qty: DashMap<ClientOrderId, Quantity>,
    order_filled_qty_by_venue: Mutex<AHashMap<ClientOrderId, AHashMap<VenueOrderId, Quantity>>>,
    pending_rejections: Mutex<AHashMap<ClientOrderId, OrderStatusReport>>,
    /// Corrective reduce queued by the cancel-replace promotion, keyed by
    /// client order id. Drained by the WS loop.
    pub pending_corrective: DashMap<ClientOrderId, CorrectiveReduce>,
    released_cancels: Mutex<Vec<OrderStatusReport>>,
    released_cancel_notify: tokio::sync::Notify,
    clearing: AtomicBool,
}

impl Default for WsDispatchState {
    fn default() -> Self {
        Self {
            order_contexts: DashMap::new(),
            emitted_accepted: DashSet::default(),
            emitted_triggered: DashSet::default(),
            pending_submissions: DashSet::default(),
            pending_submission_rejections: DashMap::new(),
            filled_orders: DashSet::default(),
            emitted_trades: Mutex::new(FifoCache::new()),
            terminal_cloids: Mutex::new(FifoCache::new()),
            closed_venue_order_ids: Mutex::new(FifoCache::new()),
            cached_venue_order_ids: DashMap::new(),
            pending_modify_chains: DashMap::new(),
            buffered_fills: DashMap::new(),
            order_filled_qty: DashMap::new(),
            order_filled_qty_by_venue: Mutex::new(AHashMap::new()),
            pending_rejections: Mutex::new(AHashMap::new()),
            pending_corrective: DashMap::new(),
            released_cancels: Mutex::new(Vec::new()),
            released_cancel_notify: tokio::sync::Notify::new(),
            clearing: AtomicBool::new(false),
        }
    }
}

impl WsDispatchState {
    /// Creates a new empty dispatch state.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Registers an order context. Called by the execution client at order
    /// submission time, before any WebSocket events for the order can arrive.
    pub fn register_context(&self, context: OrderContext) {
        self.order_contexts
            .insert(context.identity.client_order_id, context);
    }

    /// Returns a copy of the context for the given client order id, if any.
    #[must_use]
    pub fn lookup_context(&self, client_order_id: &ClientOrderId) -> Option<OrderContext> {
        self.order_contexts.get(client_order_id).map(|r| *r)
    }

    /// Marks a tracked order as awaiting its submission POST response.
    pub fn mark_submission_pending(&self, client_order_id: ClientOrderId) {
        self.pending_submissions.insert(client_order_id);
    }

    /// Returns whether the order still awaits its submission POST response.
    #[must_use]
    pub fn submission_pending(&self, client_order_id: &ClientOrderId) -> bool {
        self.pending_submissions.contains(client_order_id)
    }

    /// Holds a submission-time rejection until the POST response resolves.
    /// A replay for the same venue order ID cannot reduce its reported filled quantity.
    pub fn buffer_submission_rejection(
        &self,
        client_order_id: ClientOrderId,
        report: OrderStatusReport,
    ) {
        self.pending_submission_rejections
            .entry(client_order_id)
            .and_modify(|pending| retain_rejection_report(pending, &report))
            .or_insert(report);
    }

    /// Resolves submission tracking and returns any early rejection report.
    #[must_use]
    pub fn resolve_submission(&self, client_order_id: &ClientOrderId) -> Option<OrderStatusReport> {
        self.pending_submissions.remove(client_order_id);
        self.pending_submission_rejections
            .remove(client_order_id)
            .map(|(_, report)| report)
    }

    /// Refreshes the tracked price for a modify ack when the new report
    /// carries an updated price.
    pub fn update_context_price(&self, client_order_id: &ClientOrderId, price: Option<Price>) {
        if let Some(price) = price
            && let Some(mut entry) = self.order_contexts.get_mut(client_order_id)
        {
            entry.price = Some(price);
        }
    }

    /// Refreshes the tracked quantity for a modify ack.
    pub fn update_context_quantity(&self, client_order_id: &ClientOrderId, quantity: Quantity) {
        if let Some(mut entry) = self.order_contexts.get_mut(client_order_id) {
            entry.quantity = quantity;
        }
    }

    /// Marks an `OrderAccepted` event as emitted for this order.
    pub fn insert_accepted(&self, cid: ClientOrderId) {
        self.evict_if_full(&self.emitted_accepted);
        self.emitted_accepted.insert(cid);
    }

    // Returns false when the order's `OrderTriggered` was already emitted
    fn insert_triggered(&self, cid: ClientOrderId) -> bool {
        self.evict_if_full(&self.emitted_triggered);
        self.emitted_triggered.insert(cid)
    }

    /// Marks an order as having reached a terminal state.
    ///
    /// Returns `true` when this call claimed the terminal state, and `false`
    /// when another path had already claimed it.
    pub fn insert_filled(&self, cid: ClientOrderId) -> bool {
        self.evict_if_full(&self.filled_orders);
        self.filled_orders.insert(cid)
    }

    /// Atomically inserts a trade id into the dedup set.
    ///
    /// Returns `true` when the trade was already present (i.e. it is a
    /// duplicate), `false` otherwise.
    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
        let mut set = self.emitted_trades.lock();
        !set.insert(trade_id)
    }

    #[must_use]
    pub(crate) fn trade_emitted(&self, trade_id: &TradeId) -> bool {
        self.emitted_trades.lock().contains(trade_id)
    }

    /// Records a terminal raw Hyperliquid CLOID.
    ///
    /// Recording it lets the normal CLOID mapping be removed while a late
    /// unresolved order update is suppressed instead of forwarded as an external report.
    pub fn insert_terminal_cloid(&self, cloid: Ustr) {
        let mut set = self.terminal_cloids.lock();
        let _ = set.insert(cloid);
    }

    /// Returns whether a raw Hyperliquid CLOID already reached a terminal state.
    #[must_use]
    pub fn terminal_cloid_seen(&self, cloid: &Ustr) -> bool {
        let set = self.terminal_cloids.lock();
        set.contains(cloid)
    }

    /// Caches the venue order id observed for a tracked client order id.
    pub fn record_venue_order_id(
        &self,
        client_order_id: ClientOrderId,
        venue_order_id: VenueOrderId,
    ) {
        self.cached_venue_order_ids
            .insert(client_order_id, venue_order_id);
    }

    /// Returns the previously cached venue order id, if any.
    #[must_use]
    pub fn cached_venue_order_id(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
        self.cached_venue_order_ids.get(client_order_id).map(|r| *r)
    }

    /// Queues an in-flight modify intent for cancel-before-accept suppression
    /// and records the target absolute total qty for the cancel-replace
    /// promotion. Returns the intent's generation.
    ///
    /// The generation lets the submission path clear only this modify on
    /// failure via [`Self::clear_modify_generation`], leaving newer queued
    /// modifies intact. Chained modifies append rather than overwrite, so a
    /// later modify cannot drop an earlier pending old-leg marker.
    pub fn mark_pending_modify(
        &self,
        client_order_id: ClientOrderId,
        old_venue_order_id: VenueOrderId,
        target_qty: Quantity,
    ) -> u64 {
        let mut chain = self
            .pending_modify_chains
            .entry(client_order_id)
            .or_default();
        let generation = chain.next_generation;
        chain.next_generation += 1;
        chain.intents.push_back(ModifyIntent {
            generation,
            old_venue_order_id: Some(old_venue_order_id),
            target_qty,
            sent_request: None,
        });

        if chain.intents.len() > MAX_PENDING_MODIFY_INTENTS {
            chain.intents.pop_front();
            log::warn!(
                "Modify chain for {client_order_id} exceeded {MAX_PENDING_MODIFY_INTENTS}; \
                 evicting oldest intent",
            );
        }
        generation
    }

    /// Clears the entire pending modify chain for a client order id.
    pub fn clear_pending_modify(&self, client_order_id: &ClientOrderId) {
        self.pending_modify_chains.remove(client_order_id);
    }

    /// Removes a single modify intent by generation, leaving newer queued
    /// modifies intact. Drops the chain entry when it empties.
    ///
    /// When the removed intent is the front, the next queued modify inherits
    /// its old leg: a rejected modify does not cancel-replace, so the resting
    /// leg it targeted is still live and the next modify cancel-replaces the
    /// same one. A non-front removal needs no reparenting; the front's
    /// promotion (or its own removal) advances the chain. A held cancel whose
    /// leg no remaining intent targets is released to the WebSocket consumer loop.
    pub fn clear_modify_generation(&self, client_order_id: &ClientOrderId, generation: u64) {
        let Some(mut chain) = self.pending_modify_chains.get_mut(client_order_id) else {
            return;
        };
        let removed_front_old = chain
            .intents
            .front()
            .filter(|front| front.generation == generation)
            .and_then(|front| front.old_venue_order_id);
        chain
            .intents
            .retain(|intent| intent.generation != generation);

        if let Some(old) = removed_front_old
            && let Some(new_front) = chain.intents.front_mut()
        {
            new_front.old_venue_order_id = Some(old);
        }

        let released = chain.release_held_cancel();
        drop(chain);
        // Remove only if still empty: a concurrent mark for the same order may
        // queue a new intent between the drop above and this remove
        self.pending_modify_chains
            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());

        if let Some(report) = released {
            self.release_cancel(client_order_id, report);
        }
    }

    /// Stashes the exact venue request sent onto the most recently queued
    /// modify intent for the order.
    pub fn stash_modify_request(
        &self,
        client_order_id: ClientOrderId,
        request: HyperliquidExchangePlaceOrderRequest,
    ) {
        if let Some(mut chain) = self.pending_modify_chains.get_mut(&client_order_id)
            && let Some(back) = chain.intents.back_mut()
        {
            back.sent_request = Some(request);
        } else {
            log::debug!(
                "Stash modify request for {client_order_id} with no pending intent; ignoring"
            );
        }
    }

    /// Returns a clone of the front intent's stashed modify request, if any.
    #[must_use]
    pub fn modify_request(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Option<HyperliquidExchangePlaceOrderRequest> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().and_then(|i| i.sent_request.clone()))
    }

    /// Claims the front (oldest) modify intent for promotion.
    ///
    /// Advances the next queued intent's old leg to `new_venue_order_id`: its
    /// cancel-replace targets the replacement just promoted, not the leg it was
    /// queued against. Drops a held cancel, which the promotion shows was the
    /// cancel leg of a modify. Returns the claimed intent, or `None` when no
    /// intent is queued (an external modify with no local marker). Drops the
    /// chain entry when it empties.
    pub fn claim_front_modify(
        &self,
        client_order_id: &ClientOrderId,
        new_venue_order_id: VenueOrderId,
    ) -> Option<ModifyIntent> {
        let mut chain = self.pending_modify_chains.get_mut(client_order_id)?;
        let claimed = chain.intents.pop_front();
        chain.held_cancel = None;
        if let Some(next) = chain.intents.front_mut() {
            next.old_venue_order_id = Some(new_venue_order_id);
        }
        drop(chain);
        // Remove only if still empty: a concurrent mark for the same order may
        // queue a new intent between the drop above and this remove
        self.pending_modify_chains
            .remove_if(client_order_id, |_, chain| chain.intents.is_empty());
        claimed
    }

    pub(crate) fn record_closed_venue_order_id(&self, venue_order_id: VenueOrderId) {
        self.closed_venue_order_ids.lock().add(venue_order_id);
    }

    /// Queues a corrective reduce for the WebSocket consumer loop to post.
    pub fn queue_corrective(&self, client_order_id: ClientOrderId, corrective: CorrectiveReduce) {
        self.pending_corrective.insert(client_order_id, corrective);
    }

    /// Removes and returns a queued corrective reduce, if any.
    #[must_use]
    pub fn take_corrective(&self, client_order_id: &ClientOrderId) -> Option<CorrectiveReduce> {
        self.pending_corrective
            .remove(client_order_id)
            .map(|(_, v)| v)
    }

    /// Returns whether any modify intent is queued for the client order id.
    #[must_use]
    pub fn has_pending_modify(&self, client_order_id: &ClientOrderId) -> bool {
        self.pending_modify_chains
            .get(client_order_id)
            .is_some_and(|chain| !chain.intents.is_empty())
    }

    /// Returns the front intent's old venue order id, if any.
    #[must_use]
    pub fn pending_modify(&self, client_order_id: &ClientOrderId) -> Option<VenueOrderId> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().and_then(|i| i.old_venue_order_id))
    }

    /// Returns whether any queued intent cancel-replaces `venue_order_id`.
    ///
    /// Used to suppress the `CANCELED(old_voi)` leg of any in-flight modify in
    /// the chain, not only the oldest.
    #[must_use]
    pub fn pending_modify_contains_old(
        &self,
        client_order_id: &ClientOrderId,
        venue_order_id: VenueOrderId,
    ) -> bool {
        self.pending_modify_chains
            .get(client_order_id)
            .is_some_and(|chain| chain.targets(venue_order_id))
    }

    // A cancel of a leg a queued modify targets is either that modify's cancel leg or a cancel
    // from elsewhere that fails the modify, so the chain keeps it until it can tell them apart.
    // Checks and holds under one lock, so a modify cleared concurrently cannot drop the cancel
    pub(crate) fn hold_cancel(
        &self,
        client_order_id: &ClientOrderId,
        report: &OrderStatusReport,
    ) -> bool {
        let Some(mut chain) = self.pending_modify_chains.get_mut(client_order_id) else {
            return false;
        };

        if !chain.targets(report.venue_order_id) {
            return false;
        }

        chain.held_cancel = Some(report.clone());
        true
    }

    fn release_cancel(&self, client_order_id: &ClientOrderId, report: OrderStatusReport) {
        log::debug!(
            "Releasing held terminal report for {} on {client_order_id}: status={:?}",
            report.venue_order_id,
            report.order_status,
        );
        self.released_cancels.lock().push(report);
        self.released_cancel_notify.notify_one();
    }

    // Cancel safe: released cancels stay queued until a call returns them
    pub(crate) async fn recv_released_cancels(&self) -> Vec<OrderStatusReport> {
        loop {
            let released = std::mem::take(&mut *self.released_cancels.lock());

            if !released.is_empty() {
                return released;
            }

            self.released_cancel_notify.notified().await;
        }
    }

    /// Returns the front intent's recorded target absolute total qty, if any.
    #[must_use]
    pub fn pending_modify_target_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
        self.pending_modify_chains
            .get(client_order_id)
            .and_then(|chain| chain.intents.front().map(|i| i.target_qty))
    }

    /// Buffers a `FillReport` arrived during an in-flight cancel-replace.
    pub fn buffer_fill(&self, client_order_id: ClientOrderId, fill: FillReport) {
        self.buffered_fills
            .entry(client_order_id)
            .or_default()
            .push(fill);
    }

    /// Removes and returns buffered fills for the cid, in arrival order.
    #[must_use]
    pub fn drain_buffered_fills(&self, client_order_id: &ClientOrderId) -> Vec<FillReport> {
        self.buffered_fills
            .remove(client_order_id)
            .map(|(_, v)| v)
            .unwrap_or_default()
    }

    /// Number of buffered fills for the cid.
    #[must_use]
    pub fn buffered_fill_count(&self, client_order_id: &ClientOrderId) -> usize {
        self.buffered_fills
            .get(client_order_id)
            .map_or(0, |r| r.len())
    }

    /// Records cumulative filled quantity for a tracked order.
    pub fn record_filled_qty(&self, client_order_id: ClientOrderId, qty: Quantity) {
        self.order_filled_qty.insert(client_order_id, qty);
    }

    /// Returns the previously recorded cumulative filled quantity, if any.
    #[must_use]
    pub fn previous_filled_qty(&self, client_order_id: &ClientOrderId) -> Option<Quantity> {
        self.order_filled_qty.get(client_order_id).map(|r| *r)
    }

    /// Removes all dispatch state for an order that has reached a terminal state.
    ///
    /// `filled_orders` is intentionally *not* cleared here: the marker is
    /// used to suppress stale replays and must outlive the context cleanup.
    pub fn cleanup_terminal(&self, client_order_id: &ClientOrderId) {
        self.order_contexts.remove(client_order_id);
        self.emitted_accepted.remove(client_order_id);
        self.emitted_triggered.remove(client_order_id);
        self.pending_submissions.remove(client_order_id);
        self.pending_submission_rejections.remove(client_order_id);
        self.cached_venue_order_ids.remove(client_order_id);
        self.pending_modify_chains.remove(client_order_id);
        self.pending_corrective.remove(client_order_id);
        self.buffered_fills.remove(client_order_id);
        self.order_filled_qty.remove(client_order_id);
        self.order_filled_qty_by_venue
            .lock()
            .remove(client_order_id);
        self.pending_rejections.lock().remove(client_order_id);
    }

    fn evict_if_full(&self, set: &DashSet<ClientOrderId>) {
        if set.len() >= DEDUP_CAPACITY
            && self
                .clearing
                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
                .is_ok()
        {
            set.clear();
            self.clearing.store(false, Ordering::Release);
        }
    }
}

/// Outcome of a single dispatch call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DispatchOutcome {
    /// The report was for a tracked order. Typed events have been emitted
    /// (or intentionally skipped, e.g. dedup hit). The caller must not
    /// forward the report as a fallback.
    Tracked,
    /// The report is for an external / untracked order. The caller should
    /// forward the report via [`ExecutionEventEmitter::send_order_status_report`]
    /// or [`ExecutionEventEmitter::send_fill_report`] so the engine can
    /// reconcile.
    External,
    /// The report was recognized as stale (e.g. cancel leg of a
    /// cancel-replace modify, or replay after terminal state). The caller
    /// must drop it without forwarding.
    Skip,
}

/// Dispatches an [`OrderStatusReport`] using the two-tier routing contract.
///
/// Returns [`DispatchOutcome::Tracked`] when the report maps to a tracked
/// order (typed events have been emitted or dedup hit), [`External`] when
/// the caller should forward the report as an untracked fallback, or
/// [`Skip`] when the report is a stale / race leg that must be dropped.
///
/// [`External`]: DispatchOutcome::External
/// [`Skip`]: DispatchOutcome::Skip
pub fn dispatch_order_event(
    report: &OrderStatusReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let Some(client_order_id) = report.client_order_id else {
        return DispatchOutcome::External;
    };

    if state.filled_orders.contains(&client_order_id) {
        log::debug!(
            "Skipping stale report for filled order: cid={client_order_id}, status={:?}",
            report.order_status,
        );
        return DispatchOutcome::Skip;
    }

    let client_order_id_str = client_order_id.as_str();
    if client_order_id_str.starts_with("0x")
        && state.terminal_cloid_seen(&Ustr::from(client_order_id_str))
    {
        log::debug!(
            "Skipping stale terminal report for raw cloid: cid={client_order_id}, status={:?}",
            report.order_status,
        );
        return DispatchOutcome::Skip;
    }

    let Some(context) = state.lookup_context(&client_order_id) else {
        return DispatchOutcome::External;
    };

    match report.order_status {
        OrderStatus::Accepted => {
            handle_accepted(report, client_order_id, &context, state, emitter, ts_init)
        }
        OrderStatus::Triggered => {
            handle_triggered(report, client_order_id, &context, state, emitter, ts_init)
        }
        OrderStatus::Canceled => {
            handle_canceled(report, client_order_id, &context, state, emitter, ts_init)
        }
        OrderStatus::Expired => {
            handle_expired(report, client_order_id, &context, state, emitter, ts_init)
        }
        OrderStatus::Rejected => {
            handle_rejected(report, client_order_id, &context, state, emitter, ts_init)
        }
        OrderStatus::Filled => handle_filled_marker(client_order_id, state),
        OrderStatus::PartiallyFilled => {
            // Fills come via `FillReport`; nothing to emit from the status path.
            DispatchOutcome::Tracked
        }
        OrderStatus::PendingUpdate
        | OrderStatus::PendingCancel
        | OrderStatus::Submitted
        | OrderStatus::Initialized
        | OrderStatus::Denied
        | OrderStatus::Released
        | OrderStatus::Emulated
        | OrderStatus::Voided => DispatchOutcome::Tracked,
    }
}

/// Dispatches a [`FillReport`] using the two-tier routing contract.
///
/// Returns [`DispatchOutcome::Tracked`] when the fill has been emitted as
/// an `OrderFilled` event (or skipped via trade dedup), [`External`] when
/// the caller should forward the fill via
/// [`ExecutionEventEmitter::send_fill_report`], or [`Skip`] when the fill
/// is a replay for an already-terminal order and must be dropped.
///
/// [`External`]: DispatchOutcome::External
/// [`Skip`]: DispatchOutcome::Skip
pub fn dispatch_order_fill(
    report: &FillReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let Some(client_order_id) = report.client_order_id else {
        return DispatchOutcome::External;
    };

    if state.filled_orders.contains(&client_order_id) {
        return if state.check_and_insert_trade(report.trade_id) {
            DispatchOutcome::Skip
        } else {
            DispatchOutcome::External
        };
    }

    let Some(mut context) = state.lookup_context(&client_order_id) else {
        return DispatchOutcome::External;
    };

    // Set when a fill promotes, so the corrective-reduce runs after the fill applies
    let mut promoted_corrective: Option<(Quantity, HyperliquidExchangePlaceOrderRequest)> = None;

    // Promote the binding from the fill so a dropped replacement ACCEPTED cannot
    // strand it (see module docs).
    // An older leg's fill must not move the binding back to that leg
    if state.has_pending_modify(&client_order_id)
        && let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && report.venue_order_id != cached_voi
        && venue_oid(report.venue_order_id) >= venue_oid(cached_voi)
    {
        let target = state.pending_modify_target_qty(&client_order_id);
        let sent_request = state.modify_request(&client_order_id);
        // Prefer the modify target price over the stale cached context price
        let price = sent_request
            .as_ref()
            .zip(context.price)
            .and_then(|(r, cached)| Price::from_decimal_dp(r.price, cached.precision).ok())
            .or(context.price);
        let Some(price) = price else {
            log::warn!(
                "Cannot promote cancel-replace for {client_order_id} from fill: no target \
                 or cached price; buffering until the replacement ACCEPTED arrives",
            );
            state.buffer_fill(client_order_id, report.clone());
            return DispatchOutcome::Tracked;
        };
        let updated_quantity = target.unwrap_or(context.quantity);
        promote_cancel_replace(
            client_order_id,
            &context,
            state,
            emitter,
            report.venue_order_id,
            report.account_id,
            price,
            updated_quantity,
            None,
            report.ts_event,
            ts_init,
        );
        // Re-read the context advanced by the promotion (quantity and price)
        context = state.lookup_context(&client_order_id).unwrap_or(context);

        if let (Some(target), Some(sent_request)) = (target, sent_request) {
            promoted_corrective = Some((target, sent_request));
        }
    }

    if state.check_and_insert_trade(report.trade_id) {
        log::debug!(
            "Skipping duplicate fill for {client_order_id}: trade_id={}",
            report.trade_id
        );
        return DispatchOutcome::Tracked;
    }

    let previous = state
        .previous_filled_qty(&client_order_id)
        .unwrap_or_else(|| Quantity::zero(report.last_qty.precision));
    let cumulative = previous + report.last_qty;

    let is_terminal_fill = cumulative >= context.quantity;
    if is_terminal_fill && !claim_terminal_order(client_order_id, state, OrderStatus::Filled) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        &context,
        state,
        emitter,
        report.ts_event,
        ts_init,
    );

    let filled = OrderFilled::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        report.venue_order_id,
        report.account_id,
        report.trade_id,
        context.identity.order_side,
        context.identity.order_type,
        report.last_qty,
        report.last_px,
        report.commission.currency,
        report.liquidity_side,
        UUID4::new(),
        report.ts_event,
        ts_init,
        false,
        report.venue_position_id,
        Some(report.commission),
        None,
    );
    emitter.send_order_event(OrderEventAny::Filled(filled));

    state.record_filled_qty(client_order_id, cumulative);
    {
        let mut orders = state.order_filled_qty_by_venue.lock();
        let fills = orders.entry(client_order_id).or_default();
        let quantity = fills
            .entry(report.venue_order_id)
            .or_insert_with(|| Quantity::zero(report.last_qty.precision));
        *quantity = *quantity + report.last_qty;
    }

    // Cumulative now includes this fill, so the reduce sizes against the true remaining
    if let Some((target, sent_request)) = promoted_corrective {
        maybe_queue_corrective_reduce(
            state,
            client_order_id,
            report.venue_order_id,
            target,
            sent_request,
        );
    }

    if is_terminal_fill {
        state.cleanup_terminal(&client_order_id);
        return DispatchOutcome::Tracked;
    }

    release_rejection_after_fill(client_order_id, state);
    DispatchOutcome::Tracked
}

fn release_rejection_after_fill(client_order_id: ClientOrderId, state: &WsDispatchState) {
    let mut pending = state.pending_rejections.lock();

    let Some(rejection) = pending.get(&client_order_id) else {
        return;
    };

    if filled_qty_for_venue(state, client_order_id, rejection.venue_order_id) < rejection.filled_qty
    {
        return;
    }

    let mut canceled = rejection.clone();
    pending.remove(&client_order_id);
    drop(pending);
    canceled.order_status = OrderStatus::Canceled;
    state.release_cancel(&client_order_id, canceled);
}

fn handle_accepted(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let venue_order_id = report.venue_order_id;
    let ts_event = report.ts_last;
    let account_id = report.account_id;

    // Cancel-replace detection: if an earlier ACCEPTED cached a different
    // venue_order_id under the same client_order_id, this ACCEPTED is the
    // replacement leg of a Hyperliquid modify and must be promoted to
    // OrderUpdated. See GH-3827.
    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && cached_voi != venue_order_id
    {
        // A replayed accept for an older leg must not restore a superseded binding
        if venue_oid(venue_order_id) < venue_oid(cached_voi) {
            log::debug!(
                "Skipping ACCEPTED for superseded leg {venue_order_id} on {client_order_id}"
            );
            return DispatchOutcome::Skip;
        }

        let price = report.price.or(context.price);
        let Some(price) = price else {
            log::warn!(
                "Cannot emit OrderUpdated for cancel-replace {client_order_id}: \
                 no price on report and no cached price on context",
            );
            return DispatchOutcome::Skip;
        };

        // Prefer user target over venue's remaining-only `report.quantity`;
        // fall back when no marker (external modify).
        let target_total_qty = state.pending_modify_target_qty(&client_order_id);
        let updated_quantity = target_total_qty.unwrap_or(report.quantity);
        let sent_request = state.modify_request(&client_order_id);

        promote_cancel_replace(
            client_order_id,
            context,
            state,
            emitter,
            venue_order_id,
            account_id,
            price,
            updated_quantity,
            report.trigger_price,
            ts_event,
            ts_init,
        );

        if let (Some(target), Some(sent_request)) = (target_total_qty, sent_request) {
            maybe_queue_corrective_reduce(
                state,
                client_order_id,
                venue_order_id,
                target,
                sent_request,
            );
        }

        return DispatchOutcome::Tracked;
    }

    if state.emitted_accepted.contains(&client_order_id) {
        // Repeat ACCEPTED for an already-accepted order. Nothing to emit;
        // refresh the cached price so a subsequent cancel-replace without a
        // report price can still recover an accurate value.
        state.update_context_price(&client_order_id, report.price);
        return DispatchOutcome::Tracked;
    }

    state.insert_accepted(client_order_id);
    state.record_venue_order_id(client_order_id, venue_order_id);
    state.update_context_price(&client_order_id, report.price);

    let accepted = OrderAccepted::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        venue_order_id,
        account_id,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
    );
    emitter.send_order_event(OrderEventAny::Accepted(accepted));
    DispatchOutcome::Tracked
}

// Shared by the ACCEPTED branch and the fill path (dropped-ACCEPTED recovery) so the
// cancel-replace binding is recovered from whichever arrives first. See GH-3827, GH-3972.
#[allow(
    clippy::too_many_arguments,
    reason = "promotion needs the full OrderUpdated field set, sourced from two report shapes"
)]
fn promote_cancel_replace(
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    venue_order_id: VenueOrderId,
    account_id: AccountId,
    price: Price,
    quantity: Quantity,
    trigger_price: Option<Price>,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) {
    state.record_venue_order_id(client_order_id, venue_order_id);
    state.update_context_quantity(&client_order_id, quantity);
    state.update_context_price(&client_order_id, Some(price));
    // Claim the front intent; the next queued modify advances to this replacement
    state.claim_front_modify(&client_order_id, venue_order_id);

    let updated = OrderUpdated::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        quantity,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
        Some(venue_order_id),
        Some(account_id),
        LIMIT_ORDER_TYPES
            .contains(&context.identity.order_type)
            .then_some(price),
        trigger_price,
        None,
        false,
    );
    emitter.send_order_event(OrderEventAny::Updated(updated));

    // Drain fills buffered before the binding advanced. Bypasses
    // `handle_execution_report`; FIFO-bounded caches make any residue benign.
    let buffered = state.drain_buffered_fills(&client_order_id);
    for fill in buffered {
        dispatch_order_fill(&fill, state, emitter, ts_init);
    }
}

/// Promotes a cancel-replace replacement surfaced by a query during an in-flight modify.
///
/// When the query returns the replacement leg (`Accepted`, `venue_order_id` diverging from the
/// cached one, modify tracked), emits the `OrderUpdated` that rebinds the order, so a dropped
/// replacement `Accepted` with no fill cannot strand the binding on the canceled leg. Returns
/// `true` when promoted; the caller still forwards the report so the engine confirms the order.
pub fn promote_replacement_from_query(
    report: &OrderStatusReport,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> bool {
    if report.order_status != OrderStatus::Accepted {
        return false;
    }

    let Some(client_order_id) = report.client_order_id else {
        return false;
    };

    if !state.has_pending_modify(&client_order_id) {
        return false;
    }

    let Some(cached_voi) = state.cached_venue_order_id(&client_order_id) else {
        return false;
    };

    if report.venue_order_id == cached_voi {
        return false;
    }

    let Some(context) = state.lookup_context(&client_order_id) else {
        return false;
    };

    let Some(price) = report.price.or(context.price) else {
        log::warn!(
            "Cannot promote cancel-replace from query for {client_order_id}: \
             no price on report and no cached price on context",
        );
        return false;
    };

    // Prefer the user target over the venue's remaining-only `report.quantity`
    let updated_quantity = state
        .pending_modify_target_qty(&client_order_id)
        .unwrap_or(report.quantity);

    promote_cancel_replace(
        client_order_id,
        &context,
        state,
        emitter,
        report.venue_order_id,
        report.account_id,
        price,
        updated_quantity,
        report.trigger_price,
        report.ts_last,
        ts_init,
    );

    log::debug!("Promoted cancel-replace replacement for {client_order_id} from query");

    true
}

// Queue a corrective reduce when a fill that raced the modify left the replacement
// oversized. Reached from both promotion paths; the engine overfill guard backstops.
fn maybe_queue_corrective_reduce(
    state: &WsDispatchState,
    client_order_id: ClientOrderId,
    venue_order_id: VenueOrderId,
    target: Quantity,
    sent_request: HyperliquidExchangePlaceOrderRequest,
) {
    // Recovery opens closed legs before replaying their fills, those legs cannot be reduced
    if state
        .closed_venue_order_ids
        .lock()
        .contains(&venue_order_id)
    {
        return;
    }

    let Ok(new_oid) = venue_order_id.as_str().parse::<u64>() else {
        return;
    };

    let filled = state
        .previous_filled_qty(&client_order_id)
        .unwrap_or_else(|| Quantity::zero(target.precision));
    if filled >= target {
        return;
    }

    let remaining = (target - filled).as_decimal().normalize();

    let sent_size = sent_request.size;
    if sent_size > remaining {
        let mut corrective = sent_request;
        corrective.size = remaining;

        let generation = state.mark_pending_modify(client_order_id, venue_order_id, target);
        state.stash_modify_request(client_order_id, corrective.clone());
        state.queue_corrective(
            client_order_id,
            CorrectiveReduce {
                oid: new_oid,
                generation,
                request: corrective,
            },
        );

        log::warn!(
            "Cancel-replace left {client_order_id} oversized on {venue_order_id} \
             (sent {sent_size}, remaining {remaining}); queuing corrective reduce",
        );
    }
}

fn handle_triggered(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if !matches!(
        context.identity.order_type,
        OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched
    ) {
        log::debug!(
            "Ignoring TRIGGERED status for non-triggerable order type {:?}: {client_order_id}",
            context.identity.order_type,
        );
        return DispatchOutcome::Tracked;
    }

    // A recovery read replays a trigger the stream or an earlier read already applied
    if !state.insert_triggered(client_order_id) {
        log::debug!("Skipping repeated TRIGGERED for {client_order_id}");
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        context,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let triggered = OrderTriggered::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(report.venue_order_id),
        Some(report.account_id),
    );
    emitter.send_order_event(OrderEventAny::Triggered(triggered));
    DispatchOutcome::Tracked
}

fn handle_canceled(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    let venue_order_id = report.venue_order_id;

    if let Some(cached_voi) = state.cached_venue_order_id(&client_order_id)
        && cached_voi != venue_order_id
    {
        if !state.has_pending_modify(&client_order_id)
            || venue_oid(venue_order_id) <= venue_oid(cached_voi)
        {
            log::debug!(
                "Skipping stale CANCELED for {venue_order_id} (cached {cached_voi}) on {client_order_id}",
            );
            return DispatchOutcome::Skip;
        }

        let Some(price) = report.price.or(context.price) else {
            log::warn!(
                "Cannot promote canceled replacement for {client_order_id}: no report or cached price",
            );
            return DispatchOutcome::Skip;
        };

        let quantity = state
            .pending_modify_target_qty(&client_order_id)
            .unwrap_or(report.quantity);

        // The replacement can close before its acceptance arrives, promote without queuing
        // a corrective reduce against the already closed leg.
        promote_cancel_replace(
            client_order_id,
            context,
            state,
            emitter,
            venue_order_id,
            report.account_id,
            price,
            quantity,
            report.trigger_price,
            report.ts_last,
            ts_init,
        );
    }

    // Cancel-before-accept race: an in-flight modify may deliver
    // CANCELED(old_voi) before the replacement ACCEPTED(new_voi). Any queued
    // intent whose old leg matches (marked before the HTTP call, cleared on
    // failure) holds that cancel so the later ACCEPTED routes through
    // OrderUpdated, and a failed modify releases it. See GH-3827.
    if state.hold_cancel(&client_order_id, report) {
        log::debug!(
            "Holding cancel-before-accept leg for {client_order_id}: venue_order_id={venue_order_id}",
        );
        return DispatchOutcome::Skip;
    }

    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        venue_order_id,
        report.account_id,
        context,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let canceled = OrderCanceled::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(venue_order_id),
        Some(report.account_id),
        None,
    );
    emitter.send_order_event(OrderEventAny::Canceled(canceled));

    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn handle_expired(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    ensure_accepted_emitted(
        client_order_id,
        report.venue_order_id,
        report.account_id,
        context,
        state,
        emitter,
        report.ts_last,
        ts_init,
    );

    let expired = OrderExpired::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        Some(report.venue_order_id),
        Some(report.account_id),
    );
    emitter.send_order_event(OrderEventAny::Expired(expired));
    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn handle_rejected(
    report: &OrderStatusReport,
    client_order_id: ClientOrderId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_init: UnixNanos,
) -> DispatchOutcome {
    if state.submission_pending(&client_order_id) {
        state.buffer_submission_rejection(client_order_id, report.clone());
        return DispatchOutcome::Skip;
    }

    // A rejected replacement never became the bound order. Its original leg's
    // cancellation closes the order once the failed modify releases any held cancel.
    if let Some(bound) = state.cached_venue_order_id(&client_order_id)
        && bound != report.venue_order_id
    {
        return DispatchOutcome::Skip;
    }

    {
        // Check and park atomically so the final fill cannot miss the pending rejection
        let mut pending = state.pending_rejections.lock();
        if !report.filled_qty.is_zero()
            && filled_qty_for_venue(state, client_order_id, report.venue_order_id)
                < report.filled_qty
        {
            pending
                .entry(client_order_id)
                .and_modify(|pending| retain_rejection_report(pending, report))
                .or_insert_with(|| report.clone());
            return DispatchOutcome::Skip;
        }

        if pending.contains_key(&client_order_id) {
            return DispatchOutcome::Skip;
        }
    }

    if !report.filled_qty.is_zero()
        || state
            .previous_filled_qty(&client_order_id)
            .is_some_and(|qty| !qty.is_zero())
    {
        return handle_canceled(report, client_order_id, context, state, emitter, ts_init);
    }

    if !claim_terminal_order(client_order_id, state, report.order_status) {
        return DispatchOutcome::Skip;
    }

    let reason = report
        .cancel_reason
        .clone()
        .unwrap_or_else(|| "Order rejected by exchange".to_string());
    let rejected = OrderRejected::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        report.account_id,
        Ustr::from(&reason),
        UUID4::new(),
        report.ts_last,
        ts_init,
        false,
        report.post_only && reason.contains(HYPERLIQUID_POST_ONLY_WOULD_MATCH),
    );
    emitter.send_order_event(OrderEventAny::Rejected(rejected));
    state.cleanup_terminal(&client_order_id);
    DispatchOutcome::Tracked
}

fn retain_rejection_report(pending: &mut OrderStatusReport, report: &OrderStatusReport) {
    if report.venue_order_id != pending.venue_order_id || report.filled_qty >= pending.filled_qty {
        *pending = report.clone();
    }
}

fn filled_qty_for_venue(
    state: &WsDispatchState,
    client_order_id: ClientOrderId,
    venue_order_id: VenueOrderId,
) -> Quantity {
    state
        .order_filled_qty_by_venue
        .lock()
        .get(&client_order_id)
        .and_then(|fills| fills.get(&venue_order_id).copied())
        .unwrap_or_else(|| Quantity::zero(0))
}

fn claim_terminal_order(
    client_order_id: ClientOrderId,
    state: &WsDispatchState,
    status: OrderStatus,
) -> bool {
    let claimed = state.insert_filled(client_order_id);
    if !claimed {
        log::debug!("Skipping duplicate terminal event for {client_order_id}: status={status:?}",);
    }

    claimed
}

fn handle_filled_marker(
    _client_order_id: ClientOrderId,
    _state: &WsDispatchState,
) -> DispatchOutcome {
    // A status-only `FILLED` marker does not carry fill data; the actual
    // `OrderFilled` is emitted from `dispatch_order_fill` when the matching
    // trade arrives. Do *not* set `filled_orders` here, otherwise the
    // follow-up fill would be classified as a stale replay and dropped
    // before the terminal `OrderFilled` event can be emitted. The fill
    // path installs the marker itself once the cumulative fill quantity
    // matches the tracked order quantity.
    DispatchOutcome::Tracked
}

/// Synthesizes and emits an `OrderAccepted` event when one has not yet been
/// emitted for the given order.
///
/// Used before emitting non-Accepted events so strategies always observe the
/// canonical `Submitted -> Accepted -> ...` lifecycle even when the venue
/// compresses the placement and follow-up event into a single message (fast
/// fills).
#[allow(clippy::too_many_arguments)]
fn ensure_accepted_emitted(
    client_order_id: ClientOrderId,
    venue_order_id: VenueOrderId,
    account_id: AccountId,
    context: &OrderContext,
    state: &WsDispatchState,
    emitter: &ExecutionEventEmitter,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) {
    if state.emitted_accepted.contains(&client_order_id) {
        return;
    }
    state.insert_accepted(client_order_id);
    state.record_venue_order_id(client_order_id, venue_order_id);

    let accepted = OrderAccepted::new(
        emitter.trader_id(),
        context.identity.strategy_id,
        context.identity.instrument_id,
        client_order_id,
        venue_order_id,
        account_id,
        UUID4::new(),
        ts_event,
        ts_init,
        false,
    );
    emitter.send_order_event(OrderEventAny::Accepted(accepted));
}

// Hyperliquid assigns venue order IDs in increasing order, so a higher one is a newer leg
pub(crate) fn venue_oid(venue_order_id: VenueOrderId) -> u64 {
    venue_order_id.as_str().parse().unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, time::Duration};

    use futures_util::FutureExt;
    use nautilus_common::{messages::ExecutionEvent, testing::wait_until};
    use nautilus_core::time::get_atomic_clock_realtime;
    use nautilus_live::execution::context::OrderIdentity;
    use nautilus_model::{
        enums::{AccountType, LiquiditySide, OrderSide, TimeInForce},
        identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId},
        types::Money,
    };
    use rstest::rstest;
    use rust_decimal::Decimal;

    use super::*;
    use crate::http::models::{
        HyperliquidExchangeLimitParams, HyperliquidExchangeOrderKind, HyperliquidExchangeTif,
    };

    fn make_context(client_order_id: ClientOrderId) -> OrderContext {
        OrderContext {
            identity: OrderIdentity {
                client_order_id,
                strategy_id: StrategyId::from("S-001"),
                instrument_id: InstrumentId::from("BTC-USD-PERP.HYPERLIQUID"),
                order_side: OrderSide::Buy,
                order_type: OrderType::Limit,
            },
            quantity: Quantity::from("0.0001"),
            price: None,
            trigger_price: None,
            trigger_type: None,
            time_in_force: TimeInForce::Gtc,
            is_post_only: false,
            is_reduce_only: false,
            is_quote_quantity: false,
        }
    }

    #[rstest]
    fn test_rejection_registers_atomically_with_fill_completion() {
        let state = Arc::new(WsDispatchState::new());
        let (emitter, mut rx) = test_emitter();
        let cid = ClientOrderId::new("O-REJECTION-RACE");
        let context = make_context(cid);
        state.register_context(context);

        let first = FillReport::new(
            AccountId::new("HYPERLIQUID-001"),
            context.identity.instrument_id,
            VenueOrderId::new("100"),
            TradeId::new("T-RACE-1"),
            OrderSide::Buy,
            Quantity::from("0.00003"),
            Price::from("56730.0"),
            Money::from("0.17 USD"),
            LiquiditySide::Taker,
            Some(cid),
            None,
            UnixNanos::default(),
            UnixNanos::default(),
            None,
        );
        dispatch_order_fill(&first, &state, &emitter, UnixNanos::default());

        let rejection = OrderStatusReport::new(
            first.account_id,
            first.instrument_id,
            Some(cid),
            first.venue_order_id,
            OrderSide::Buy.into(),
            OrderType::Limit,
            TimeInForce::Gtc,
            OrderStatus::Rejected,
            context.quantity,
            Quantity::from("0.00005"),
            UnixNanos::default(),
            UnixNanos::default(),
            UnixNanos::default(),
            None,
        );
        let fills = state.order_filled_qty_by_venue.lock();
        let reject_state = state.clone();
        let reject_emitter = emitter.clone();

        let rejected = std::thread::spawn(move || {
            dispatch_order_event(
                &rejection,
                &reject_state,
                &reject_emitter,
                UnixNanos::default(),
            )
        });

        let timeout = Duration::from_secs(5);
        wait_until(|| state.pending_rejections.is_locked(), timeout);
        let mut last = first;
        last.trade_id = TradeId::new("T-RACE-2");
        last.last_qty = Quantity::from("0.00002");
        last.commission = Money::from("0.29 USD");
        let fill_state = state.clone();
        let fill_emitter = emitter.clone();
        let filled = std::thread::spawn(move || {
            dispatch_order_fill(&last, &fill_state, &fill_emitter, UnixNanos::default())
        });

        wait_until(
            || state.previous_filled_qty(&cid) == Some(Quantity::from("0.00005")),
            timeout,
        );
        drop(fills);
        rejected.join().unwrap();
        assert_eq!(filled.join().unwrap(), DispatchOutcome::Tracked);

        if let Some(released) = state.recv_released_cancels().now_or_never() {
            assert_eq!(released.len(), 1);
            dispatch_order_event(&released[0], &state, &emitter, UnixNanos::default());
        }

        let mut events = Vec::new();
        while let Ok(event) = rx.try_recv() {
            events.push(event);
        }

        let [
            ExecutionEvent::Order(OrderEventAny::Accepted(_)),
            ExecutionEvent::Order(OrderEventAny::Filled(first)),
            ExecutionEvent::Order(OrderEventAny::Filled(last)),
            ExecutionEvent::Order(OrderEventAny::Canceled(_)),
        ] = events.as_slice()
        else {
            panic!("expected two fills followed by cancellation: {events:?}");
        };

        assert_eq!(first.trade_id, TradeId::new("T-RACE-1"));
        assert_eq!(first.last_qty, Quantity::from("0.00003"));
        assert_eq!(first.commission, Some(Money::from("0.17 USD")));
        assert_eq!(last.trade_id, TradeId::new("T-RACE-2"));
        assert_eq!(last.last_qty, Quantity::from("0.00002"));
        assert_eq!(last.commission, Some(Money::from("0.29 USD")));
        assert_eq!(state.lookup_context(&cid), None);
        assert!(state.filled_orders.contains(&cid));
    }

    #[rstest]
    fn test_register_and_lookup_context() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-001");
        state.register_context(make_context(cid));

        assert_eq!(state.lookup_context(&cid), Some(make_context(cid)));
    }

    #[rstest]
    fn test_lookup_context_missing_returns_none() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("not-tracked");
        assert!(state.lookup_context(&cid).is_none());
    }

    #[rstest]
    fn test_update_context_refreshes_price_and_quantity() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-003");
        state.register_context(make_context(cid));

        state.update_context_price(&cid, Some(Price::from("56731.5")));
        state.update_context_quantity(&cid, Quantity::from("0.0002"));

        assert_eq!(
            state.lookup_context(&cid),
            Some(OrderContext {
                price: Some(Price::from("56731.5")),
                quantity: Quantity::from("0.0002"),
                ..make_context(cid)
            }),
        );
    }

    #[rstest]
    fn test_update_context_price_without_price_keeps_cached_price() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-004");
        let cached = OrderContext {
            price: Some(Price::from("56730.0")),
            ..make_context(cid)
        };
        state.register_context(cached);

        state.update_context_price(&cid, None);

        assert_eq!(state.lookup_context(&cid), Some(cached));
    }

    #[rstest]
    fn test_insert_accepted_dedup() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-002");
        assert!(!state.emitted_accepted.contains(&cid));
        state.insert_accepted(cid);
        assert!(state.emitted_accepted.contains(&cid));
        state.insert_accepted(cid);
        assert!(state.emitted_accepted.contains(&cid));
    }

    #[rstest]
    fn test_check_and_insert_trade_detects_duplicates() {
        let state = WsDispatchState::new();
        let trade = TradeId::new("trade-1");
        assert!(!state.check_and_insert_trade(trade));
        assert!(state.check_and_insert_trade(trade));
    }

    #[rstest]
    fn test_pending_modify_roundtrip() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-010");
        let voi = VenueOrderId::new("v-1");
        let target_qty = Quantity::from("0.0001");

        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
        state.mark_pending_modify(cid, voi, target_qty);
        assert_eq!(state.pending_modify(&cid), Some(voi));
        assert_eq!(state.pending_modify_target_qty(&cid), Some(target_qty));
        state.clear_pending_modify(&cid);
        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
    }

    #[rstest]
    fn test_cleanup_terminal_preserves_filled_marker() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-020");
        state.register_context(make_context(cid));
        state.insert_accepted(cid);
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.0001"));
        state.insert_filled(cid);
        state.cleanup_terminal(&cid);

        assert!(state.lookup_context(&cid).is_none());
        assert!(!state.emitted_accepted.contains(&cid));
        assert!(state.pending_modify(&cid).is_none());
        assert!(state.pending_modify_target_qty(&cid).is_none());
        // `filled_orders` outlives `cleanup_terminal` so replays stay suppressed.
        assert!(state.filled_orders.contains(&cid));
    }

    #[rstest]
    fn test_cleanup_terminal_clears_corrective_state() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-021");
        let request = sample_request(Decimal::from(1));
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("1"));
        state.stash_modify_request(cid, request.clone());
        state.queue_corrective(
            cid,
            CorrectiveReduce {
                oid: 1,
                generation: 0,
                request,
            },
        );

        assert!(state.modify_request(&cid).is_some());

        state.cleanup_terminal(&cid);

        assert!(state.modify_request(&cid).is_none());
        assert!(state.take_corrective(&cid).is_none());
        assert!(state.pending_modify(&cid).is_none());
    }

    fn sample_request(size: Decimal) -> HyperliquidExchangePlaceOrderRequest {
        HyperliquidExchangePlaceOrderRequest {
            asset: 0,
            is_buy: true,
            price: "100".parse::<Decimal>().unwrap(),
            size,
            reduce_only: false,
            kind: HyperliquidExchangeOrderKind::Limit {
                limit: HyperliquidExchangeLimitParams {
                    tif: HyperliquidExchangeTif::Gtc,
                },
            },
            cloid: None,
        }
    }

    #[rstest]
    fn test_modify_chain_keeps_both_intents_on_rapid_modifies() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-100");
        let g0 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        let g1 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));

        assert_ne!(g0, g1);
        assert!(state.has_pending_modify(&cid));
        // Front is the oldest intent
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert_eq!(
            state.pending_modify_target_qty(&cid),
            Some(Quantity::from("0.00020")),
        );
        // Both queued old legs suppress their cancel-before-accept
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
    }

    #[rstest]
    fn test_clear_modify_generation_preserves_newer_intent() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-101");
        // Two rapid modifies queued before either acked, both against the live
        // leg v-0.
        let g0 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // Failure of the first modify clears only its generation; the second
        // stays, still targeting the live leg v-0.
        state.clear_modify_generation(&cid, g0);

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert_eq!(
            state.pending_modify_target_qty(&cid),
            Some(Quantity::from("0.00030")),
        );
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
    }

    #[rstest]
    fn test_claim_front_modify_advances_next_old_id() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-102");
        // Both queued against the same stale old leg (M2 fired before M1 acked)
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // Promoting the first replacement claims the front and advances the
        // next intent's old leg to the replacement id.
        let claimed = state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
        assert_eq!(
            claimed.map(|i| i.target_qty),
            Some(Quantity::from("0.00020"))
        );

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
        // The stale leg no longer matches once advanced
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));

        // Claiming the last intent empties the chain
        let claimed2 = state.claim_front_modify(&cid, VenueOrderId::new("v-2"));
        assert_eq!(
            claimed2.map(|i| i.target_qty),
            Some(Quantity::from("0.00030"))
        );
        assert!(!state.has_pending_modify(&cid));
        assert!(state.pending_modify(&cid).is_none());
    }

    #[rstest]
    fn test_modify_chain_caps_and_evicts_oldest() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-106");
        // Queue one past the cap with no promotion or clear to drain them
        for i in 0..=MAX_PENDING_MODIFY_INTENTS {
            let voi = format!("v-{i}");
            state.mark_pending_modify(cid, VenueOrderId::new(&voi), Quantity::from("0.00020"));
        }

        // The oldest intent was evicted; the newest remains and the front
        // advanced to the second-oldest.
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
        let newest = format!("v-{MAX_PENDING_MODIFY_INTENTS}");
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new(&newest)));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
    }

    #[rstest]
    fn test_clear_front_modify_reparents_next_old() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-104");
        // Three rapid modifies where the first already promoted to v-1
        // (advancing the front to old=v-1); the third still holds stale v-0.
        let g_front =
            state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00020"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // The front is rejected; the next intent must inherit the live leg
        // (v-1), not keep stale v-0, or CANCELED(v-1) would surface as a real
        // cancel.
        state.clear_modify_generation(&cid, g_front);

        assert!(state.has_pending_modify(&cid));
        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-1")));
        assert!(state.pending_modify_contains_old(&cid, VenueOrderId::new("v-1")));
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-0")));
    }

    #[rstest]
    fn test_clear_non_front_modify_leaves_front_old() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-105");
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        let g_back =
            state.mark_pending_modify(cid, VenueOrderId::new("v-9"), Quantity::from("0.00030"));

        // Removing a non-front intent must not disturb the front's old leg
        state.clear_modify_generation(&cid, g_back);

        assert_eq!(state.pending_modify(&cid), Some(VenueOrderId::new("v-0")));
        assert!(!state.pending_modify_contains_old(&cid, VenueOrderId::new("v-9")));
    }

    #[rstest]
    fn test_stash_modify_request_targets_latest_intent() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-103");
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        state.stash_modify_request(cid, sample_request(Decimal::from(1)));
        state.mark_pending_modify(cid, VenueOrderId::new("v-1"), Quantity::from("0.00030"));
        state.stash_modify_request(cid, sample_request(Decimal::from(2)));

        // Front intent keeps its own request
        assert_eq!(
            state.modify_request(&cid).map(|r| r.size),
            Some(Decimal::from(1)),
        );
        // After claiming the front, the next intent's request surfaces
        state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
        assert_eq!(
            state.modify_request(&cid).map(|r| r.size),
            Some(Decimal::from(2)),
        );
    }

    fn canceled_report(client_order_id: ClientOrderId, venue_order_id: &str) -> OrderStatusReport {
        OrderStatusReport::new(
            AccountId::from("HYPERLIQUID-001"),
            InstrumentId::from("BTC-USD-PERP.HYPERLIQUID"),
            Some(client_order_id),
            VenueOrderId::new(venue_order_id),
            Some(OrderSide::Buy),
            OrderType::Limit,
            TimeInForce::Gtc,
            OrderStatus::Canceled,
            Quantity::from("0.0001"),
            Quantity::from("0"),
            UnixNanos::default(),
            UnixNanos::default(),
            UnixNanos::default(),
            Some(UUID4::new()),
        )
    }

    fn released_cancels(state: &WsDispatchState) -> Vec<OrderStatusReport> {
        state
            .recv_released_cancels()
            .now_or_never()
            .unwrap_or_default()
    }

    fn test_emitter() -> (
        ExecutionEventEmitter,
        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
    ) {
        let mut emitter = ExecutionEventEmitter::new(
            get_atomic_clock_realtime(),
            TraderId::from("TESTER-001"),
            AccountId::from("HYPERLIQUID-001"),
            AccountType::Margin,
            None,
        );
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        emitter.set_sender(tx);
        (emitter, rx)
    }

    #[rstest]
    #[case::bound_leg(Some("v-0"), true)]
    #[case::unbound_leg(None, true)]
    #[case::superseded_leg(Some("v-1"), false)]
    fn test_canceled_for_modified_leg_is_held_unless_superseded(
        #[case] bound_venue_order_id: Option<&str>,
        #[case] expected_held: bool,
    ) {
        let (emitter, mut rx) = test_emitter();
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-107");
        state.register_context(make_context(cid));

        if let Some(voi) = bound_venue_order_id {
            state.record_venue_order_id(cid, VenueOrderId::new(voi));
        }

        let generation =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.0001"));
        let canceled = canceled_report(cid, "v-0");

        let outcome = dispatch_order_event(&canceled, &state, &emitter, UnixNanos::default());
        state.clear_modify_generation(&cid, generation);

        assert_eq!(outcome, DispatchOutcome::Skip);
        assert!(rx.try_recv().is_err());
        assert_eq!(
            released_cancels(&state),
            expected_held
                .then_some(canceled)
                .into_iter()
                .collect::<Vec<_>>(),
        );
    }

    #[rstest]
    fn test_held_cancel_releases_when_last_modify_of_its_leg_fails() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-108");
        let leg = VenueOrderId::new("v-0");
        state.record_venue_order_id(cid, leg);
        let g0 = state.mark_pending_modify(cid, leg, Quantity::from("0.00020"));
        let g1 = state.mark_pending_modify(cid, leg, Quantity::from("0.00030"));
        let canceled = canceled_report(cid, "v-0");

        let held = state.hold_cancel(&cid, &canceled);
        state.clear_modify_generation(&cid, g0);
        let released_while_targeted = released_cancels(&state);
        state.clear_modify_generation(&cid, g1);

        assert!(held);
        assert_eq!(released_while_targeted, vec![]);
        assert_eq!(released_cancels(&state), vec![canceled]);
        assert!(!state.has_pending_modify(&cid));
    }

    #[rstest]
    #[case::response_cancel_reject(["response", "cancel", "reject"])]
    #[case::response_reject_cancel(["response", "reject", "cancel"])]
    #[case::cancel_response_reject(["cancel", "response", "reject"])]
    #[case::cancel_reject_response(["cancel", "reject", "response"])]
    #[case::reject_response_cancel(["reject", "response", "cancel"])]
    #[case::reject_cancel_response(["reject", "cancel", "response"])]
    fn test_failed_modify_closes_original_once(
        #[case] sequence: [&str; 3],
        #[values(false, true)] filled: bool,
    ) {
        let (emitter, mut rx) = test_emitter();
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-MODIFY-REJECT");
        let original = VenueOrderId::new("100");
        state.register_context(make_context(cid));
        state.record_venue_order_id(cid, original);
        state.insert_accepted(cid);
        if filled {
            state.record_filled_qty(cid, Quantity::from("0.00004"));
        }

        let generation = state.mark_pending_modify(cid, original, Quantity::from("0.0001"));
        let mut canceled = canceled_report(cid, "100");
        canceled.ts_last = UnixNanos::from(100);
        let mut rejected = canceled_report(cid, "101");
        rejected.order_status = OrderStatus::Rejected;
        rejected.ts_last = UnixNanos::from(200);

        for action in sequence {
            match action {
                "response" => state.clear_modify_generation(&cid, generation),
                "cancel" => {
                    dispatch_order_event(&canceled, &state, &emitter, UnixNanos::default());
                }
                "reject" => {
                    dispatch_order_event(&rejected, &state, &emitter, UnixNanos::default());
                }
                _ => unreachable!(),
            }

            for report in released_cancels(&state) {
                dispatch_order_event(&report, &state, &emitter, UnixNanos::default());
            }
        }

        let canceled_replay =
            dispatch_order_event(&canceled, &state, &emitter, UnixNanos::default());
        let rejected_replay =
            dispatch_order_event(&rejected, &state, &emitter, UnixNanos::default());
        let event = rx.try_recv().unwrap();

        let ExecutionEvent::Order(OrderEventAny::Canceled(event)) = event else {
            panic!("expected OrderCanceled, received {event:?}");
        };

        assert_eq!(event.trader_id, emitter.trader_id());
        assert_eq!(event.strategy_id, StrategyId::from("S-001"));
        assert_eq!(
            event.instrument_id,
            InstrumentId::from("BTC-USD-PERP.HYPERLIQUID")
        );
        assert_eq!(event.client_order_id, cid);
        assert_eq!(event.venue_order_id, Some(original));
        assert_eq!(event.account_id, Some(canceled.account_id));
        assert_eq!(event.ts_event, canceled.ts_last);
        assert_eq!(event.ts_init, UnixNanos::default());
        assert!(!event.reconciliation);
        assert_eq!(event.reason, None);
        assert_eq!(event.causation_id, None);
        assert!(rx.try_recv().is_err());
        assert_eq!(canceled_replay, DispatchOutcome::Skip);
        assert_eq!(rejected_replay, DispatchOutcome::Skip);
        assert!(state.lookup_context(&cid).is_none());
        assert!(!state.has_pending_modify(&cid));
    }

    #[rstest]
    fn test_promotion_drops_held_cancel() {
        let state = WsDispatchState::new();
        let cid = ClientOrderId::new("O-109");
        state.record_venue_order_id(cid, VenueOrderId::new("v-0"));
        state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00020"));
        let g1 =
            state.mark_pending_modify(cid, VenueOrderId::new("v-0"), Quantity::from("0.00030"));

        // The replacement shows the cancel was the first modify's cancel leg, so the second
        // modify failing later must not apply it
        let held = state.hold_cancel(&cid, &canceled_report(cid, "v-0"));
        state.claim_front_modify(&cid, VenueOrderId::new("v-1"));
        state.clear_modify_generation(&cid, g1);

        assert!(held);
        assert_eq!(released_cancels(&state), vec![]);
        assert!(!state.has_pending_modify(&cid));
    }
}