hypercall_api/handlers/

replace_orders.rs

use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::time::Instant;

use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use utoipa::ToSchema;

use crate::error::ApiError;
use crate::middleware::SignerContext;
use crate::request_auth::verify_request;
use crate::sonic_json::SonicJson;
use crate::trading_halt::TradingHaltState;
use axum::extract::State;
use hypercall_runtime_api::increment_pending_requests;
use hypercall_types::utils::get_timestamp_millis;
use hypercall_types::{
    to_contract_units_decimal, validate_price_precision, ReplaceOrderRequest, WalletAddress,
    MAX_PRICE_SIGNIFICANT_FIGURES,
};
use hypercall_types::{
    OrderAction, OrderActionMessage, OrderInfo, OrderUpdateMessage, OrderUpdateStatus, TimeInForce,
};
use hypercall_types::{ParsedOptionSymbol, Side};
use serde::{Deserialize, Serialize};
use tokio::sync::{mpsc, RwLock};
use tokio::time::timeout;
use tracing::Instrument;
use uuid::Uuid;

#[cfg(feature = "otel-tracing")]
use tracing_opentelemetry::OpenTelemetrySpanExt;

use super::{ensure_order_creation_allowed, AppState, ENGINE_RESPONSE_TIMEOUT};

pub use super::bulk_orders::BulkOrderResult;
use super::bulk_orders::MAX_BULK_ORDER_SIZE;

#[derive(Debug, Deserialize, ToSchema)]
pub struct BulkReplaceOrderRequest {
    /// Array of replace requests (max 50)
    pub replacements: Vec<ReplaceOrderRequest>,
}

#[derive(Debug, Serialize, ToSchema)]
pub struct BulkReplaceOrderResponse {
    /// Results for each replace in the request
    pub results: Vec<BulkOrderResult>,
}

/// Atomically cancel an existing order and place a new one.
///
/// The old order is canceled first. If the cancel fails (order not found,
/// already filled), the new order is NOT placed and the endpoint returns
/// an error. If the cancel succeeds, the new order is placed in the same
/// engine tick with no gap.
#[utoipa::path(
    put,
    path = "/order",
    request_body = ReplaceOrderRequest,
    responses(
        (status = 200, description = "Replace result (new order status)", body = OrderUpdateMessage),
        (status = 400, description = "Invalid request"),
        (status = 401, description = "Unauthorized"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "Trading"
)]
pub async fn replace_order(
    State(state): State<AppState>,
    signer_ctx: SignerContext,
    SonicJson(request): SonicJson<ReplaceOrderRequest>,
) -> Result<SonicJson<OrderUpdateMessage>, ApiError> {
    let span = tracing::info_span!(
        "api.replace_order",
        wallet = %signer_ctx.wallet_address,
        cancel_order_id = %request.order_id,
        symbol = %request.symbol,
        side = ?request.side,
        size = %request.size,
        price = %request.price,
        client_id = ?request.client_id,
        action = "ReplaceOrder",
    );

    async move {
        // Parse price and size
        let price: Decimal = Decimal::from_str(&request.price).map_err(|_| {
            ApiError::bad_request(format!("Invalid price format: {}", request.price))
        })?;
        let size: Decimal = Decimal::from_str(&request.size)
            .map_err(|_| ApiError::bad_request(format!("Invalid size format: {}", request.size)))?;

        // Validate symbol
        let _parsed_symbol = ParsedOptionSymbol::from_symbol(&request.symbol)
            .map_err(|e| ApiError::bad_request(format!("Invalid symbol: {}", e)))?;

        ensure_order_creation_allowed(&state, &request.symbol).await?;

        if price <= dec!(0) {
            return Err(ApiError::bad_request("Price must be positive"));
        }
        if size <= dec!(0) {
            return Err(ApiError::bad_request("Size must be positive"));
        }

        let price_f64 = price
            .to_f64()
            .ok_or_else(|| ApiError::bad_request("Price value out of range"))?;
        if let Err(e) = validate_price_precision(price_f64, MAX_PRICE_SIGNIFICANT_FIGURES) {
            return Err(ApiError::bad_request(format!(
                "Invalid price precision: {}",
                e
            )));
        }

        tracing::info!(
            "Replacing order {} for wallet: {} (signed by: {}), new symbol: {}",
            request.order_id,
            signer_ctx.wallet_address,
            signer_ctx.signer_address,
            request.symbol
        );

        // Build OrderInfo with cancel target in order_id and new order details in the rest
        let order_info = OrderInfo {
            symbol: request.symbol.clone(),
            price,
            size: to_contract_units_decimal(&request.symbol, size),
            side: request.side,
            tif: request.tif,
            client_id: request.client_id,
            order_id: Some(request.order_id),
            is_perp: false,
            underlying: None,
            reduce_only: None,
            nonce: Some(request.nonce),
            signature: None,
            mmp_enabled: request.mmp_enabled,
            builder_code_address: request.builder_code_address,
        };

        let order_action_msg = OrderActionMessage {
            timestamp: get_timestamp_millis(),
            info: order_info,
            action: OrderAction::ReplaceOrder,
            wallet: request.wallet,
            api_wallet_address: Some(signer_ctx.signer_address),
            mmp_triggered: false,
            request_id: Some(Uuid::now_v7().to_string()),
        };

        let (response_tx, mut response_rx) = mpsc::channel(1);

        let engine_request = hypercall_runtime_api::UnifiedEngineRequest {
            message: order_action_msg,
            response_tx,
            enqueued_at: Instant::now(),
            #[cfg(feature = "otel-tracing")]
            trace_context: Some(tracing::Span::current().context()),
        };

        increment_pending_requests();
        state
            .order_sender
            .send(engine_request)
            .await
            .map_err(|_| ApiError::internal_error("Failed to send replace order to engine"))?;

        let response = match timeout(ENGINE_RESPONSE_TIMEOUT, response_rx.recv()).await {
            Ok(Some(resp)) => resp,
            Ok(None) => {
                return Err(ApiError::internal_error("No response from engine"));
            }
            Err(_) => {
                return Err(ApiError::gateway_timeout(
                    "Timeout waiting for engine response",
                ));
            }
        };

        Ok(SonicJson(response))
    }
    .instrument(span)
    .await
}

/// Replace multiple orders using cancel-all-then-create-all ordering.
///
/// Per-leg semantics mirror PUT /order (cancel old + place new), but across
/// a batch every leg's cancel is enqueued before any create, so the engine
/// cannot match a new leg against a sibling's still-resting opposite. This
/// eliminates the self-trade-prevention kill path that fired when the bulk
/// dispatched independent ReplaceOrder commands. A per-leg response is
/// returned in original index order; a leg whose cancel failed (already
/// filled / not found) is skipped for the create phase and returned with
/// the cancel rejection details. Not atomic across the batch: the gap
/// between a leg's cancel committing and its create landing is per-leg.
/// See engineering-docs/docs/flows/order-lifecycle.mdx for tradeoffs vs.
/// the single-command PUT /order path.
#[utoipa::path(
    put,
    path = "/bulk_order",
    request_body = BulkReplaceOrderRequest,
    responses(
        (status = 200, description = "Bulk replace results (per-leg, in input order)", body = BulkReplaceOrderResponse),
        (status = 400, description = "Invalid request (e.g., too many replacements)"),
        (status = 500, description = "Internal server error")
    ),
    security(("eip712_signature" = [])),
    tag = "Trading"
)]
/// Bulk replace orders.
///
/// Executes **cancel-all-then-create-all** ordering across the batch: every
/// leg's cancel is enqueued before any create, so the engine is guaranteed
/// to process all cancels (removing every old leg from the book) before any
/// new order matches. Without this, the engine's FIFO processing of
/// independent `OrderAction::ReplaceOrder` commands meant each command's
/// phase-2 place could match against a sibling's still-resting opposite
/// leg, triggering self-trade prevention and losing the new leg entirely.
///
/// The path is:
///   1. `validate_replace_request` per leg (sig / auth / parse / precision).
///   2. `orchestrate_bulk_replace` dispatches all `CancelOrder` commands,
///      awaits responses, then dispatches `CreateOrder` commands only for
///      legs whose cancel actually reached `Canceled`.
///   3. Per-leg results merge: prefer the create response; fall back to a
///      cancel-only result when the cancel landed but the create didn't.
///
/// Semantics vs. single-command `ReplaceOrder`: the `PUT /order` single-
/// replace path dispatches **one** `OrderAction::ReplaceOrder` engine command
/// that handles the cancel and place atomically within one engine tick and
/// one journal entry (`src/rsm/unified_engine/order_routing.rs:process_replace_order`).
/// The bulk path, by contrast, issues separate `CancelOrder` and `CreateOrder`
/// commands (2N journal entries per bulk of N), so there is a per-leg gap
/// between a leg's cancel committing and its create landing. Real-world MMs
/// re-quote every cycle, so a missed create is recovered on the next tick.
/// The `PUT /order` single-replace endpoint is unchanged.
pub async fn bulk_replace_order(
    State(state): State<AppState>,
    SonicJson(request): SonicJson<BulkReplaceOrderRequest>,
) -> Result<SonicJson<BulkReplaceOrderResponse>, ApiError> {
    if request.replacements.len() > MAX_BULK_ORDER_SIZE {
        return Err(ApiError::bad_request(format!(
            "Bulk replace request exceeds max size: {} > {}",
            request.replacements.len(),
            MAX_BULK_ORDER_SIZE
        )));
    }

    let n = request.replacements.len();
    tracing::info!(
        n,
        "Processing bulk replace (cancel-all-then-create-all ordering)"
    );

    // ---- Phase 0: per-leg validation (sig, auth, parsing) ----
    let mut validated: Vec<Result<ValidatedReplace, BulkOrderResult>> = Vec::with_capacity(n);
    for (index, replace_req) in request.replacements.into_iter().enumerate() {
        validated.push(validate_replace_request(&state, replace_req, index).await);
    }

    let results = orchestrate_bulk_replace(
        &state.order_sender,
        validated,
        BULK_REPLACE_PHASE_DEADLINE,
        BULK_REPLACE_PHASE_DEADLINE,
        Some(&state.trading_halt),
    )
    .await;

    Ok(SonicJson(BulkReplaceOrderResponse { results }))
}

/// Per-phase deadline for the bulk-replace handler. Unlike
/// `ENGINE_RESPONSE_TIMEOUT` (which is the per-message budget on the
/// singular endpoints), this is the budget for *all* legs in a phase
/// combined. Keeps worst-case bulk latency bounded even if a single
/// leg's response is hung. Phase 1 and Phase 2 each get their own
/// deadline of this size, so a 50-leg bulk is bounded by roughly
/// 2 × this duration end-to-end.
const BULK_REPLACE_PHASE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(5);

/// Drive the cancel-all-then-create-all dispatch against a provided engine
/// request sender.
///
/// - `cancel_phase_deadline` / `create_phase_deadline`: single budget for
///   *all* legs in each phase, measured from when that phase starts reading
///   responses. A 50-leg bulk with one hung cancel is bounded by at most
///   `cancel_phase_deadline + create_phase_deadline`, not
///   `N × per_leg_timeout`.
///
/// - `trading_halt`: optional reference to the global halt state. When
///   provided, a fresh snapshot is read **after phase 1 completes** (not
///   before it starts) so halts that fire while cancels are in flight
///   are caught before any create is dispatched. Tests that don't
///   exercise halts can pass `None`.
///
/// - Phase 2 dispatches a create only after observing a `Canceled`
///   response for that leg. The phase-1 cancel carries the signed nonce
///   and the phase-2 create is engine-internal, so a timeout/closed
///   response channel is treated as an unknown nonce-consumption outcome
///   and does not get a create.
async fn orchestrate_bulk_replace(
    order_sender: &mpsc::Sender<hypercall_runtime_api::UnifiedEngineRequest>,
    validated: Vec<Result<ValidatedReplace, BulkOrderResult>>,
    cancel_phase_deadline: std::time::Duration,
    create_phase_deadline: std::time::Duration,
    trading_halt: Option<&std::sync::Arc<RwLock<TradingHaltState>>>,
) -> Vec<BulkOrderResult> {
    // ---- Phase 1: dispatch all CancelOrder commands ----
    let mut cancel_rxs: Vec<(usize, mpsc::Receiver<OrderUpdateMessage>)> = Vec::new();
    let mut cancel_enqueue_failed: HashSet<usize> = HashSet::new();
    for (i, v) in validated.iter().enumerate() {
        let Ok(v) = v else { continue };
        match dispatch_to_engine(order_sender, build_cancel_message(v)).await {
            Ok(rx) => {
                cancel_rxs.push((i, rx));
            }
            Err(_) => {
                tracing::error!(index = i, "Failed to enqueue cancel to engine");
                cancel_enqueue_failed.insert(i);
            }
        }
    }

    // Phase-level deadline: start the clock *after* the last cancel was
    // enqueued, and share it across all per-leg recv()s.
    let mut cancel_responses: HashMap<usize, OrderUpdateMessage> = HashMap::new();
    let cancel_deadline = tokio::time::Instant::now() + cancel_phase_deadline;
    for (i, mut rx) in cancel_rxs {
        let now = tokio::time::Instant::now();
        let remaining = cancel_deadline.saturating_duration_since(now);
        match timeout(remaining, rx.recv()).await {
            Ok(Some(resp)) => {
                cancel_responses.insert(i, resp);
            }
            Ok(None) => tracing::error!(index = i, "No cancel response from engine"),
            Err(_) => tracing::warn!(
                index = i,
                "Cancel response not received before phase deadline; \
                 skipping create because nonce consumption is unknown"
            ),
        }
    }

    // Re-check trading-halt *after* phase 1 completes (cancels have all
    // been dispatched and their responses awaited). A halt can fire at
    // any point during that window, so a snapshot taken before phase 1
    // would be stale by now; reading it here guarantees that any leg
    // whose symbol became halted during phase 1 is blocked from phase 2
    // before we send any `CreateOrder`.
    let mut halted_after_cancel: HashSet<usize> = HashSet::new();
    if let Some(halt_lock) = trading_halt {
        let halt_state = halt_lock.read().await;
        for (i, v) in validated.iter().enumerate() {
            let Ok(v) = v else { continue };
            if halt_state
                .blocked_reason(&v.new_order_info.symbol)
                .is_some()
            {
                halted_after_cancel.insert(i);
            }
        }
    }

    // ---- Phase 2: dispatch CreateOrder ----
    //
    // Skip create for a leg when:
    //   * validation failed (no ValidatedReplace to build the message from), OR
    //   * cancel dispatch to the engine channel failed outright (we don't
    //     know the engine state), OR
    //   * cancel response came back as something other than `Canceled`
    //     (engine saw the cancel request and rejected it — the old order
    //     already filled, was already canceled, or the order id was bogus),
    //     OR
    //   * cancel response did not arrive before the phase deadline. The
    //     cancel is the nonce-consuming command. Since `build_create_message`
    //     clears the nonce for phase 2, creating after an unknown phase-1
    //     outcome could turn a replay into an engine-internal create.
    //     OR
    //   * the symbol became trading-halted between phases.
    let mut create_rxs: Vec<(usize, mpsc::Receiver<OrderUpdateMessage>)> = Vec::new();
    for (i, v) in validated.iter().enumerate() {
        let Ok(v) = v else { continue };
        if cancel_enqueue_failed.contains(&i) {
            continue;
        }
        if halted_after_cancel.contains(&i) {
            continue;
        }
        let cancel_allows_create = match cancel_responses.get(&i) {
            Some(resp) => resp.status == OrderUpdateStatus::Canceled,
            None => false,
        };
        if !cancel_allows_create {
            continue;
        }
        match dispatch_to_engine(order_sender, build_create_message(v)).await {
            Ok(rx) => create_rxs.push((i, rx)),
            Err(_) => {
                tracing::error!(index = i, "Failed to enqueue create to engine after cancel");
            }
        }
    }

    let mut create_responses: HashMap<usize, OrderUpdateMessage> = HashMap::new();
    let create_deadline = tokio::time::Instant::now() + create_phase_deadline;
    for (i, mut rx) in create_rxs {
        let now = tokio::time::Instant::now();
        let remaining = create_deadline.saturating_duration_since(now);
        match timeout(remaining, rx.recv()).await {
            Ok(Some(resp)) => {
                create_responses.insert(i, resp);
            }
            Ok(None) => tracing::error!(index = i, "No create response from engine"),
            Err(_) => tracing::warn!(
                index = i,
                "Create response not received before phase deadline"
            ),
        }
    }

    // ---- Phase 3: merge per-leg results in original order ----
    validated
        .into_iter()
        .enumerate()
        .map(|(i, v)| {
            merge_leg_result(
                i,
                v,
                cancel_responses.remove(&i),
                create_responses.remove(&i),
                cancel_enqueue_failed.contains(&i),
                halted_after_cancel.contains(&i),
            )
        })
        .collect()
}

fn merge_leg_result(
    index: usize,
    validated: Result<ValidatedReplace, BulkOrderResult>,
    cancel: Option<OrderUpdateMessage>,
    create: Option<OrderUpdateMessage>,
    cancel_enqueue_failed: bool,
    halted_after_cancel: bool,
) -> BulkOrderResult {
    if let Err(validation_failure) = validated {
        return validation_failure;
    }
    if cancel_enqueue_failed {
        return BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some("Failed to enqueue cancel to engine (queue closed or full)".to_string()),
        };
    }
    if let Some(create_resp) = create {
        let success = matches!(
            create_resp.status,
            OrderUpdateStatus::Acked
                | OrderUpdateStatus::Open
                | OrderUpdateStatus::Filled
                | OrderUpdateStatus::PartiallyFilled
        );
        return BulkOrderResult {
            index,
            success,
            data: Some(create_resp),
            error: None,
        };
    }
    if halted_after_cancel {
        return BulkOrderResult {
            index,
            success: false,
            data: cancel,
            error: Some(
                "Trading halted for this symbol after cancel phase; create was skipped".to_string(),
            ),
        };
    }
    if let Some(cancel_resp) = cancel {
        return if cancel_resp.status == OrderUpdateStatus::Canceled {
            BulkOrderResult {
                index,
                success: false,
                data: Some(cancel_resp),
                error: Some(
                    "Cancel succeeded but create response did not arrive before the phase deadline"
                        .to_string(),
                ),
            }
        } else {
            BulkOrderResult {
                index,
                success: false,
                data: Some(cancel_resp),
                error: Some("Replace aborted: cancel of existing order failed".to_string()),
            }
        };
    }
    // Cancel enqueued, no response by phase deadline. Do not dispatch the
    // engine-internal create because the nonce-consuming cancel outcome is
    // unknown.
    BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some(
            "Cancel enqueued but no cancel response was received before the phase deadline; create skipped because nonce consumption is unknown"
                .to_string(),
        ),
    }
}

/// A replace request that has passed signature / authorization / parsing
/// validation and is ready to be turned into a `(CancelOrder, CreateOrder)`
/// pair by the bulk-replace orchestrator. Splitting validation from
/// dispatch lets `orchestrate_bulk_replace` be unit-tested with a mocked
/// engine sender.
#[derive(Debug, Clone)]
struct ValidatedReplace {
    wallet: WalletAddress,
    signer_address: WalletAddress,
    old_order_id: u64,
    /// `new_order_info` already carries `mmp_enabled`, `builder_code_address`,
    /// the size converted to contract units, and `order_id = None` so the
    /// engine allocates a fresh id at place time.
    new_order_info: OrderInfo,
}

/// Run signature verification, agent authorization, symbol parsing,
/// trading-allowed checks, and price/size parsing + precision checks.
/// Returns a `ValidatedReplace` on success, or a pre-populated
/// `BulkOrderResult` carrying the rejection reason on any failure.
async fn validate_replace_request(
    state: &AppState,
    req: ReplaceOrderRequest,
    index: usize,
) -> Result<ValidatedReplace, BulkOrderResult> {
    let authorized = verify_request(
        state.agent_auth.as_ref(),
        &req,
        state.runtime_config.signing_chain_id,
    )
    .map_err(|e| BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some(e.to_string()),
    })?;

    if authorized.signer.wallet_address != req.wallet {
        return Err(BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some("Verified wallet does not match request wallet".to_string()),
        });
    }

    ParsedOptionSymbol::from_symbol(&req.symbol).map_err(|e| BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some(format!("Invalid symbol: {}", e)),
    })?;

    ensure_order_creation_allowed(state, &req.symbol)
        .await
        .map_err(|err| BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some(err.message),
        })?;

    let price = Decimal::from_str(&req.price).map_err(|_| BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some(format!("Invalid price format: {}", req.price)),
    })?;
    let size = Decimal::from_str(&req.size).map_err(|_| BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some(format!("Invalid size format: {}", req.size)),
    })?;
    if price <= dec!(0) {
        return Err(BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some("Price must be greater than 0".to_string()),
        });
    }
    if size <= dec!(0) {
        return Err(BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some("Size must be greater than 0".to_string()),
        });
    }
    let price_f64 = price.to_f64().ok_or_else(|| BulkOrderResult {
        index,
        success: false,
        data: None,
        error: Some("Price value out of range".to_string()),
    })?;
    validate_price_precision(price_f64, MAX_PRICE_SIGNIFICANT_FIGURES).map_err(|e| {
        BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some(format!("Price validation failed: {}", e)),
        }
    })?;

    let new_order_info = OrderInfo {
        symbol: req.symbol.clone(),
        price,
        size: to_contract_units_decimal(&req.symbol, size),
        side: req.side,
        tif: req.tif,
        client_id: req.client_id,
        order_id: None,
        is_perp: false,
        underlying: None,
        reduce_only: None,
        nonce: Some(req.nonce),
        signature: None,
        mmp_enabled: req.mmp_enabled,
        builder_code_address: req.builder_code_address,
    };

    Ok(ValidatedReplace {
        wallet: req.wallet,
        signer_address: authorized.signer.signer_address,
        old_order_id: req.order_id,
        new_order_info,
    })
}

/// Build an `OrderAction::CancelOrder` message targeting this replace's
/// pre-existing order. `symbol` is intentionally empty — the engine resolves
/// it from `order_id` in its per-wallet order index (see how
/// `cancel_order()` does it in this file).
fn build_cancel_message(v: &ValidatedReplace) -> OrderActionMessage {
    let cancel_info = OrderInfo {
        symbol: String::new(),
        price: dec!(0),
        size: dec!(0),
        side: Side::Buy,
        tif: TimeInForce::GTC,
        client_id: None,
        order_id: Some(v.old_order_id),
        is_perp: false,
        underlying: None,
        reduce_only: None,
        // Bulk replace splits one signed ReplaceOrder into cancel and
        // create commands. Consume the signed nonce on the cancel command
        // so a replay is rejected before the old order can be removed.
        nonce: v.new_order_info.nonce,
        signature: None,
        mmp_enabled: false,
        builder_code_address: None,
    };
    OrderActionMessage {
        timestamp: get_timestamp_millis(),
        info: cancel_info,
        action: OrderAction::CancelOrder,
        wallet: v.wallet,
        api_wallet_address: Some(v.signer_address),
        mmp_triggered: false,
        request_id: Some(Uuid::now_v7().to_string()),
    }
}

fn build_create_message(v: &ValidatedReplace) -> OrderActionMessage {
    let mut info = v.new_order_info.clone();
    info.nonce = None;
    OrderActionMessage {
        timestamp: get_timestamp_millis(),
        info,
        action: OrderAction::CreateOrder,
        wallet: v.wallet,
        api_wallet_address: Some(v.signer_address),
        mmp_triggered: false,
        request_id: Some(Uuid::now_v7().to_string()),
    }
}

/// Send an `OrderActionMessage` through the engine queue. Returns the
/// response receiver. Returns `Err(())` if the engine queue is closed.
async fn dispatch_to_engine(
    order_sender: &mpsc::Sender<hypercall_runtime_api::UnifiedEngineRequest>,
    msg: OrderActionMessage,
) -> Result<mpsc::Receiver<OrderUpdateMessage>, ()> {
    let (response_tx, response_rx) = mpsc::channel(1);
    let engine_request = hypercall_runtime_api::UnifiedEngineRequest {
        message: msg,
        response_tx,
        enqueued_at: Instant::now(),
        #[cfg(feature = "otel-tracing")]
        trace_context: Some(tracing::Span::current().context()),
    };
    increment_pending_requests();
    order_sender.send(engine_request).await.map_err(|_| ())?;
    Ok(response_rx)
}

#[cfg(test)]
mod bulk_replace_tests {
    //! Unit tests for `orchestrate_bulk_replace` and its helpers.
    //!
    //! These tests wire up a mock engine task that reads the shared
    //! `UnifiedEngineRequest` channel the handler would push into and
    //! responds per-request with a scripted `OrderUpdateStatus`. The
    //! orchestrator's job is:
    //!   1. Send all `CancelOrder` messages before any `CreateOrder`.
    //!   2. Skip the create phase for any leg whose cancel didn't reach
    //!      `Canceled`.
    //!   3. Pass through validation errors unchanged.
    //!   4. Merge per-leg results in original index order.
    //!
    //! Each test below exercises one of these invariants.
    use super::*;
    use rust_decimal_macros::dec;
    use std::time::Duration;
    use tokio::sync::mpsc;

    fn test_wallet(id: u8) -> WalletAddress {
        let mut bytes = [0u8; 20];
        bytes[19] = id;
        WalletAddress::from(bytes)
    }

    fn mk_validated(old_order_id: u64) -> ValidatedReplace {
        mk_validated_with(old_order_id, "BTC-20260101-75000-C".to_string(), false)
    }

    fn mk_validated_with_symbol(old_order_id: u64, symbol: String) -> ValidatedReplace {
        mk_validated_with(old_order_id, symbol, false)
    }

    fn mk_validated_mmp(old_order_id: u64) -> ValidatedReplace {
        mk_validated_with(old_order_id, "BTC-20260101-75000-C".to_string(), true)
    }

    fn mk_validated_with(old_order_id: u64, symbol: String, mmp_enabled: bool) -> ValidatedReplace {
        let order_info = OrderInfo {
            symbol,
            price: dec!(100),
            size: dec!(1),
            side: Side::Buy,
            tif: TimeInForce::GTC,
            client_id: None,
            order_id: None,
            is_perp: false,
            underlying: None,
            reduce_only: None,
            nonce: Some(1_700_000_000_000 + old_order_id),
            signature: None,
            mmp_enabled,
            builder_code_address: None,
        };
        ValidatedReplace {
            wallet: test_wallet(1),
            signer_address: test_wallet(2),
            old_order_id,
            new_order_info: order_info,
        }
    }

    fn mk_validation_error(index: usize) -> BulkOrderResult {
        BulkOrderResult {
            index,
            success: false,
            data: None,
            error: Some("Signature verification failed: test".to_string()),
        }
    }

    fn mk_response(
        order_id: Option<u64>,
        status: OrderUpdateStatus,
        wallet: WalletAddress,
    ) -> OrderUpdateMessage {
        OrderUpdateMessage {
            timestamp: 0,
            info: OrderInfo {
                symbol: "BTC-20260101-75000-C".to_string(),
                price: dec!(100),
                size: dec!(1),
                side: Side::Buy,
                tif: TimeInForce::GTC,
                client_id: None,
                order_id,
                is_perp: false,
                underlying: None,
                reduce_only: None,
                nonce: None,
                signature: None,
                mmp_enabled: false,
                builder_code_address: None,
            },
            status,
            reason: None,
            filled_size: dec!(0),
            order_id,
            wallet_address: wallet,
            mmp_triggered: false,
            request_id: None,
        }
    }

    /// Drives the mock engine: consume requests from the shared receiver,
    /// record the `OrderAction` of each, reply using `responder`. Returns
    /// the ordered list of actions the orchestrator enqueued.
    async fn run_mock_engine<F>(
        mut engine_rx: mpsc::Receiver<hypercall_runtime_api::UnifiedEngineRequest>,
        mut responder: F,
    ) -> Vec<OrderAction>
    where
        F: FnMut(&hypercall_runtime_api::UnifiedEngineRequest) -> Option<OrderUpdateMessage>
            + Send
            + 'static,
    {
        let mut observed: Vec<OrderAction> = Vec::new();
        while let Some(req) = engine_rx.recv().await {
            observed.push(req.message.action.clone());
            if let Some(resp) = responder(&req) {
                let _ = req.response_tx.send(resp).await;
            }
        }
        observed
    }

    /// Invariant 1: every CancelOrder arrives at the engine before any
    /// CreateOrder. This is the whole reason the handler exists.
    #[tokio::test]
    async fn cancels_dispatched_before_creates_for_batch_of_three() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let status = match req.message.action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    Some(mk_response(req.message.info.order_id, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(101)),
            Ok(mk_validated(102)),
            Ok(mk_validated(103)),
        ];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            None,
        )
        .await;
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        assert_eq!(results.len(), 3);
        assert!(results.iter().all(|r| r.success), "all 3 replaces succeed");
        assert_eq!(observed.len(), 6, "3 cancels + 3 creates");

        let cancel_idxs: Vec<usize> = observed
            .iter()
            .enumerate()
            .filter(|(_, a)| **a == OrderAction::CancelOrder)
            .map(|(i, _)| i)
            .collect();
        let create_idxs: Vec<usize> = observed
            .iter()
            .enumerate()
            .filter(|(_, a)| **a == OrderAction::CreateOrder)
            .map(|(i, _)| i)
            .collect();
        assert_eq!(cancel_idxs, vec![0, 1, 2], "cancels are first three");
        assert_eq!(create_idxs, vec![3, 4, 5], "creates come after all cancels");
    }

    /// Invariant 2: when a cancel comes back as anything other than
    /// Canceled (e.g. the order was already filled), the create phase
    /// for that leg is skipped. The leg returns a cancel-only result
    /// with an informative error string.
    #[tokio::test]
    async fn cancel_failure_skips_create_for_that_leg() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let oid = req.message.info.order_id;
                    let status = match (&req.message.action, oid) {
                        // Leg with old_order_id = 201 is "already filled".
                        (OrderAction::CancelOrder, Some(201)) => OrderUpdateStatus::Rejected,
                        (OrderAction::CancelOrder, _) => OrderUpdateStatus::Canceled,
                        (OrderAction::CreateOrder, _) => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    Some(mk_response(oid, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(200)), // will succeed
            Ok(mk_validated(201)), // cancel fails
            Ok(mk_validated(202)), // will succeed
        ];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            None,
        )
        .await;
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        // 3 cancels + 2 creates (skipped 201's create).
        assert_eq!(observed.len(), 5);
        assert_eq!(
            observed
                .iter()
                .filter(|a| **a == OrderAction::CancelOrder)
                .count(),
            3
        );
        assert_eq!(
            observed
                .iter()
                .filter(|a| **a == OrderAction::CreateOrder)
                .count(),
            2
        );

        assert!(results[0].success);
        assert!(!results[1].success);
        assert!(results[1]
            .error
            .as_deref()
            .unwrap_or("")
            .contains("cancel of existing order failed"));
        assert!(results[2].success);
    }

    /// Invariant 3: validation errors pass through unchanged to the
    /// correct output index, and a validation-failed leg is never sent
    /// to the engine.
    #[tokio::test]
    async fn validation_failures_skip_engine_dispatch() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let status = match req.message.action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    Some(mk_response(req.message.info.order_id, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(301)),
            Err(mk_validation_error(1)),
            Ok(mk_validated(303)),
        ];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            None,
        )
        .await;
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        // Validation failure means no engine traffic for that leg → 2 cancels + 2 creates.
        assert_eq!(observed.len(), 4);
        assert!(results[0].success);
        assert!(!results[1].success);
        assert_eq!(results[1].index, 1);
        assert!(results[1]
            .error
            .as_deref()
            .unwrap_or("")
            .contains("Signature verification failed"));
        assert!(results[2].success);
    }

    /// Invariant 4: when a cancel *response* times out (no reply by the
    /// phase deadline), phase 2 skips the create for that leg. The cancel
    /// carries the signed nonce and the create is intentionally
    /// engine-internal, so a missing cancel response leaves nonce
    /// consumption unknown and must not permit a replay to place a new
    /// order.
    #[tokio::test]
    async fn cancel_response_timeout_skips_create() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        // Mock engine deliberately silences the cancel response for oid=401
        // so the orchestrator cannot know whether the signed nonce was consumed.
        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let oid = req.message.info.order_id;
                    if oid == Some(401) && req.message.action == OrderAction::CancelOrder {
                        return None; // silence this cancel response
                    }
                    let status = match req.message.action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    Some(mk_response(oid, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(400)),
            Ok(mk_validated(401)), // cancel response silenced
            Ok(mk_validated(402)),
        ];
        let short_deadline = Duration::from_millis(200);
        let start = std::time::Instant::now();
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            short_deadline,
            Duration::from_secs(2),
            None,
        )
        .await;
        let elapsed = start.elapsed();
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        // Phase deadline is shared across legs, so one hung cancel can't
        // stretch total time beyond ~deadline + create phase.
        assert!(elapsed < Duration::from_secs(3), "elapsed {:?}", elapsed);

        // All 3 cancels and only 2 creates were dispatched. The timed-out
        // cancel leg is skipped in phase 2.
        assert_eq!(observed.len(), 5);
        assert_eq!(
            observed
                .iter()
                .filter(|a| **a == OrderAction::CreateOrder)
                .count(),
            2,
            "create for the hung-cancel leg must be skipped"
        );

        assert!(results[0].success);
        assert!(
            !results[1].success,
            "leg with silenced cancel response must not create"
        );
        assert!(
            results[1]
                .error
                .as_deref()
                .unwrap_or("")
                .contains("nonce consumption is unknown"),
            "unexpected timeout error: {:?}",
            results[1].error
        );
        assert!(results[2].success);
    }

    /// Invariant 5: if the cancel succeeds but the create never responds,
    /// the leg ends with a "cancel succeeded but create response did not
    /// arrive" error.  The old order is gone engine-side; caller should
    /// know to re-quote on the next cycle.
    #[tokio::test]
    async fn create_response_timeout_returns_cancel_only_result() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let oid = req.message.info.order_id;
                    match req.message.action {
                        OrderAction::CancelOrder => {
                            Some(mk_response(oid, OrderUpdateStatus::Canceled, wallet))
                        }
                        OrderAction::CreateOrder => None, // hang all creates
                        _ => None,
                    }
                })
                .await
            }
        });

        let validated = vec![Ok(mk_validated(501))];
        let short_deadline = Duration::from_millis(200);
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            short_deadline,
            None,
        )
        .await;
        drop(engine_tx);
        let _ = engine_handle.await;

        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert_eq!(
            results[0].data.as_ref().map(|r| r.status),
            Some(OrderUpdateStatus::Canceled)
        );
        assert!(
            results[0]
                .error
                .as_deref()
                .unwrap_or("")
                .contains("create response did not arrive"),
            "unexpected error: {:?}",
            results[0].error
        );
    }

    /// Invariant 7: a leg whose symbol becomes trading-halted *between
    /// phase 1 and phase 2* does NOT get a create dispatched, and returns
    /// a cancel-only result with an explicit halt reason. `orchestrate_bulk_replace`
    /// itself reads the halt state after phase 1 completes (see
    /// "Recompute halt gate after cancel phase completes" in the codex
    /// review); this test verifies that path by flipping the market halt
    /// on leg 1's symbol *while phase 1 is in flight* and asserting the
    /// create phase correctly skips it.
    #[tokio::test]
    async fn halt_fired_during_phase_one_blocks_only_matching_leg_in_phase_two() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        // Build a halt state we'll mutate from inside the mock engine once
        // phase-1 cancels are observed — simulating a halt that fires
        // while cancels are in flight.
        let halt_state = std::sync::Arc::new(RwLock::new(TradingHaltState::new()));
        let halt_state_engine = halt_state.clone();
        let halted_symbol = "BTC-20260101-80000-C".to_string();
        let halted_symbol_engine = halted_symbol.clone();

        let engine_handle = tokio::spawn({
            async move {
                let mut cancels_seen = 0usize;
                run_mock_engine(engine_rx, move |req| {
                    let status = match req.message.action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    // As soon as we've seen every cancel respond (phase 1 is
                    // effectively done), flip the halt. The orchestrator's
                    // post-phase-1 snapshot must see this.
                    if matches!(req.message.action, OrderAction::CancelOrder) {
                        cancels_seen += 1;
                        if cancels_seen == 3 {
                            let halt_state = halt_state_engine.clone();
                            let sym = halted_symbol_engine.clone();
                            tokio::spawn(async move {
                                halt_state.write().await.set_market_halt(
                                    &sym,
                                    true,
                                    "test halt".to_string(),
                                    "test".to_string(),
                                );
                            });
                        }
                    }
                    Some(mk_response(req.message.info.order_id, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(701)),
            Ok(mk_validated_with_symbol(702, halted_symbol.clone())),
            Ok(mk_validated(703)),
        ];

        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            Some(&halt_state),
        )
        .await;
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        // 3 cancels + 2 creates (leg 1 skipped in phase 2 because its
        // symbol got halted during phase 1).
        assert_eq!(
            observed
                .iter()
                .filter(|a| **a == OrderAction::CancelOrder)
                .count(),
            3
        );
        assert_eq!(
            observed
                .iter()
                .filter(|a| **a == OrderAction::CreateOrder)
                .count(),
            2
        );

        assert!(results[0].success);
        assert!(!results[1].success);
        assert!(
            results[1]
                .error
                .as_deref()
                .unwrap_or("")
                .contains("Trading halted"),
            "unexpected error: {:?}",
            results[1].error
        );
        assert!(results[2].success);
    }

    /// Invariant 8: if the engine request sender is closed (channel queue
    /// dropped — a liveness bug or shutdown race), the leg returns a
    /// distinct "Failed to enqueue cancel" error rather than a generic
    /// no-response-from-engine, so operators can tell a dead-queue state
    /// apart from an engine-timeout state.
    #[tokio::test]
    async fn cancel_enqueue_failure_reports_explicit_error() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        // Close the receiver side immediately so every `send(...)` returns Err.
        drop(engine_rx);

        let validated = vec![Ok(mk_validated(801))];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_millis(200),
            Duration::from_millis(200),
            None,
        )
        .await;

        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert!(
            results[0]
                .error
                .as_deref()
                .unwrap_or("")
                .contains("Failed to enqueue cancel"),
            "unexpected error: {:?}",
            results[0].error
        );
    }

    /// Invariant 9: total bulk latency is bounded by the sum of the two
    /// phase deadlines, not N times a per-leg timeout. A 5-leg bulk with
    /// every cancel response silenced should finish in roughly the shared
    /// cancel deadline, not five times that.
    #[tokio::test]
    async fn phase_deadline_bounds_total_bulk_latency() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        // Engine silences every cancel response. No creates should be sent
        // because the nonce-consuming cancel outcome is unknown.
        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| match req.message.action {
                    OrderAction::CancelOrder => None,
                    OrderAction::CreateOrder => Some(mk_response(
                        req.message.info.order_id,
                        OrderUpdateStatus::Open,
                        wallet,
                    )),
                    _ => None,
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(901)),
            Ok(mk_validated(902)),
            Ok(mk_validated(903)),
            Ok(mk_validated(904)),
            Ok(mk_validated(905)),
        ];
        let cancel_deadline = Duration::from_millis(300);
        let create_deadline = Duration::from_millis(300);
        let start = std::time::Instant::now();
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            cancel_deadline,
            create_deadline,
            None,
        )
        .await;
        let elapsed = start.elapsed();
        drop(engine_tx);
        let _ = engine_handle.await;

        // Phase deadlines are shared, so total time is O(cancel_deadline),
        // not O(N * per_leg_timeout). Give some slack
        // for scheduling overhead but strictly bound below N × deadline.
        assert!(
            elapsed < cancel_deadline + create_deadline + Duration::from_millis(500),
            "elapsed {:?} exceeded phase-deadline budget",
            elapsed
        );

        assert_eq!(results.len(), 5);
        assert!(results.iter().all(|r| !r.success));
        assert!(results.iter().all(|r| r
            .error
            .as_deref()
            .unwrap_or("")
            .contains("nonce consumption is unknown")));
    }

    /// Invariant 6: result indices are preserved — result[i] corresponds
    /// to input[i], regardless of success/failure mix.
    #[tokio::test]
    async fn result_order_matches_input_order() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                run_mock_engine(engine_rx, move |req| {
                    let status = match req.message.action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Open,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    Some(mk_response(req.message.info.order_id, status, wallet))
                })
                .await
            }
        });

        let validated = vec![
            Ok(mk_validated(600)),
            Err(mk_validation_error(1)),
            Ok(mk_validated(602)),
            Err(mk_validation_error(3)),
            Ok(mk_validated(604)),
        ];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            None,
        )
        .await;
        drop(engine_tx);
        let _ = engine_handle.await;

        assert_eq!(results.len(), 5);
        for (i, r) in results.iter().enumerate() {
            assert_eq!(r.index, i, "result {} has wrong index", i);
        }
        assert!(results[0].success);
        assert!(!results[1].success);
        assert!(results[2].success);
        assert!(!results[3].success);
        assert!(results[4].success);
    }

    /// `build_cancel_message` encodes enough information for the engine
    /// to cancel by order_id without needing the symbol up front.
    #[test]
    fn build_cancel_message_sets_action_and_order_id() {
        let v = mk_validated(999);
        let msg = build_cancel_message(&v);
        assert_eq!(msg.action, OrderAction::CancelOrder);
        assert_eq!(msg.info.order_id, Some(999));
        assert_eq!(msg.info.nonce, v.new_order_info.nonce);
        // Symbol is intentionally empty — engine resolves it from the per-
        // wallet order_id index (matches the singular cancel_order() path).
        assert_eq!(msg.info.symbol, "");
        assert_eq!(msg.wallet, v.wallet);
    }

    /// `build_create_message` forwards the new order's price/size/side and
    /// clears the order_id so the engine allocates a fresh id.
    #[test]
    fn build_create_message_uses_new_order_info() {
        let v = mk_validated(1234);
        let msg = build_create_message(&v);
        assert_eq!(msg.action, OrderAction::CreateOrder);
        assert_eq!(msg.info.order_id, None);
        assert_eq!(msg.info.price, v.new_order_info.price);
        assert_eq!(msg.info.size, v.new_order_info.size);
        assert_eq!(msg.info.nonce, None);
        assert_eq!(msg.wallet, v.wallet);
    }

    /// `merge_leg_result` branch coverage — each terminal outcome path
    /// should produce the right success flag, result data, and error
    /// string. Covers the new flags `cancel_enqueue_failed` and
    /// `halted_after_cancel` in addition to the original four response
    /// combinations.
    #[test]
    fn merge_leg_result_branches() {
        let wallet = test_wallet(1);

        // (a) validation error → returned verbatim, flags ignored.
        let ve = mk_validation_error(7);
        let r = merge_leg_result(7, Err(ve.clone()), None, None, false, false);
        assert_eq!(r.index, 7);
        assert!(!r.success);
        assert_eq!(r.error, ve.error);

        // (b) create present + Acked → success, cancel data discarded in favour of create.
        let cancel = mk_response(Some(1), OrderUpdateStatus::Canceled, wallet);
        let create = mk_response(Some(10), OrderUpdateStatus::Open, wallet);
        let r = merge_leg_result(
            0,
            Ok(mk_validated(1)),
            Some(cancel),
            Some(create),
            false,
            false,
        );
        assert!(r.success);
        assert_eq!(r.data.unwrap().status, OrderUpdateStatus::Open);

        // (c) cancel Canceled, no create response → "create response did not arrive".
        let cancel = mk_response(Some(1), OrderUpdateStatus::Canceled, wallet);
        let r = merge_leg_result(0, Ok(mk_validated(1)), Some(cancel), None, false, false);
        assert!(!r.success);
        assert!(r
            .error
            .as_deref()
            .unwrap()
            .contains("create response did not arrive"));

        // (d) cancel Rejected → cancel-failed path, no create attempted.
        let cancel = mk_response(Some(1), OrderUpdateStatus::Rejected, wallet);
        let r = merge_leg_result(0, Ok(mk_validated(1)), Some(cancel), None, false, false);
        assert!(!r.success);
        assert!(r
            .error
            .as_deref()
            .unwrap()
            .contains("cancel of existing order failed"));

        // (e) halted_after_cancel=true → explicit halt error, never success.
        let cancel = mk_response(Some(1), OrderUpdateStatus::Canceled, wallet);
        let r = merge_leg_result(0, Ok(mk_validated(1)), Some(cancel), None, false, true);
        assert!(!r.success);
        assert!(r.error.as_deref().unwrap().contains("Trading halted"));

        // (f) cancel enqueue failed → explicit "Failed to enqueue" error.
        let r = merge_leg_result(0, Ok(mk_validated(1)), None, None, true, false);
        assert!(!r.success);
        assert!(r
            .error
            .as_deref()
            .unwrap()
            .contains("Failed to enqueue cancel"));

        // (g) no signals at all (cancel enqueued but no response AND no create
        //     response) — terminal, informative.
        let r = merge_leg_result(0, Ok(mk_validated(1)), None, None, false, false);
        assert!(!r.success);
        assert!(r
            .error
            .as_deref()
            .unwrap()
            .contains("nonce consumption is unknown"));
    }

    /// MMP propagation invariant. The real MMP logic lives in the engine
    /// (`src/rsm/unified_engine/matching.rs` → `mmp_cache.process_fill`);
    /// the bulk-replace handler only has to guarantee two things:
    ///
    ///   1. `mmp_enabled=true` on the `ReplaceOrderRequest` arrives on the
    ///      `CreateOrder` message that the engine ultimately sees. A flag
    ///      that silently gets flipped to `false` by `build_create_message`
    ///      would defeat the market maker's MMP config — the engine can't
    ///      retroactively decide a fill was MMP-eligible.
    ///
    ///   2. Phase-1 `CancelOrder` messages never carry `mmp_enabled=true`.
    ///      Cancels don't fill and shouldn't count toward any MMP
    ///      accounting; more importantly, setting it would be a lie about
    ///      the order being replaced.
    ///
    ///   3. When the engine sends back an `OrderUpdateMessage` carrying
    ///      `mmp_triggered=true` on a create response (MMP fired during
    ///      the fill processing of that leg), the merged bulk result
    ///      must surface that status field faithfully so clients can
    ///      react (re-configure MMP, back off quoting, etc.).
    ///
    /// This test drives all three via the mock engine: one leg has
    /// `mmp_enabled=true`, one has it false, and the engine tags the
    /// mmp-enabled leg's create response with `mmp_triggered=true`.
    #[tokio::test]
    async fn mmp_flag_propagates_to_create_and_not_to_cancel() {
        let (engine_tx, engine_rx) = mpsc::channel(32);
        let wallet = test_wallet(1);

        let engine_handle = tokio::spawn({
            async move {
                // Collect the full messages this time (not just the actions)
                // so we can inspect `info.mmp_enabled` per leg per phase.
                let mut rx: mpsc::Receiver<hypercall_runtime_api::UnifiedEngineRequest> = engine_rx;
                let mut observed: Vec<(OrderAction, Option<u64>, bool)> = Vec::new();
                while let Some(req) = rx.recv().await {
                    let action = req.message.action.clone();
                    let oid = req.message.info.order_id;
                    let mmp = req.message.info.mmp_enabled;
                    observed.push((action.clone(), oid, mmp));
                    let status = match action {
                        OrderAction::CancelOrder => OrderUpdateStatus::Canceled,
                        OrderAction::CreateOrder => OrderUpdateStatus::Filled,
                        _ => OrderUpdateStatus::Rejected,
                    };
                    // On the mmp-enabled leg's create (old_order_id=502 →
                    // new OrderInfo.order_id=None on create, so match via
                    // mmp flag), tag mmp_triggered=true on the response.
                    let mut resp = mk_response(oid, status, wallet);
                    if matches!(action, OrderAction::CreateOrder) && mmp {
                        resp.mmp_triggered = true;
                    }
                    let _ = req.response_tx.send(resp).await;
                }
                observed
            }
        });

        let validated = vec![
            Ok(mk_validated(501)),     // mmp_enabled = false
            Ok(mk_validated_mmp(502)), // mmp_enabled = true
        ];
        let results = orchestrate_bulk_replace(
            &engine_tx,
            validated,
            Duration::from_secs(2),
            Duration::from_secs(2),
            None,
        )
        .await;
        drop(engine_tx);
        let observed = engine_handle.await.expect("engine task");

        assert_eq!(observed.len(), 4, "2 cancels + 2 creates");

        // Invariant 1: create's mmp_enabled mirrors the request.
        let creates: Vec<_> = observed
            .iter()
            .filter(|(a, _, _)| *a == OrderAction::CreateOrder)
            .collect();
        assert_eq!(creates.len(), 2);
        // The only leg with mmp_enabled=true on its new order is 502.
        assert!(
            creates.iter().any(|(_, _, mmp)| !*mmp),
            "leg 501 create must have mmp_enabled=false"
        );
        assert!(
            creates.iter().any(|(_, _, mmp)| *mmp),
            "leg 502 create must have mmp_enabled=true"
        );

        // Invariant 2: every cancel carries mmp_enabled=false, regardless
        // of the leg's new-order mmp_enabled setting. Cancels don't fill;
        // MMP accounting must not be triggered by a cancel.
        let cancels: Vec<_> = observed
            .iter()
            .filter(|(a, _, _)| *a == OrderAction::CancelOrder)
            .collect();
        assert_eq!(cancels.len(), 2);
        assert!(
            cancels.iter().all(|(_, _, mmp)| !*mmp),
            "cancels must never carry mmp_enabled=true"
        );

        // Invariant 3: the merged bulk result surfaces `mmp_triggered=true`
        // on the leg whose fill breached MMP, so clients can react.
        assert_eq!(results.len(), 2);
        assert!(results[0].success);
        let leg0 = results[0].data.as_ref().expect("leg 0 has response data");
        assert!(
            !leg0.mmp_triggered,
            "leg 0 (mmp_enabled=false) must not report mmp_triggered"
        );
        assert!(results[1].success);
        let leg1 = results[1].data.as_ref().expect("leg 1 has response data");
        assert!(
            leg1.mmp_triggered,
            "leg 1 (mmp_enabled=true) must surface engine's mmp_triggered=true"
        );
    }
}