hypercall_api/rfq/

rfq_manager.rs

use crate::rfq::quote_provider_cache::QuoteProviderCache;
use dashmap::DashMap;
use hypercall_db::RfqWriter;
use hypercall_runtime_api::{
    RfqExecuteCommand, RfqExecuteLeg, RfqExecuteRequest, RfqExecuteResult,
};
use hypercall_types::utils::get_timestamp_millis as current_time_ms;
use hypercall_types::{RfqStatus, Side, WalletAddress, RFQ_SELF_TRADE_REJECTION_REASON};
use hypercall_ws_protocol::{QpRfqLeg, QpServerMessage as QpOutboundMessage};
use metrics::{counter, histogram};
use rust_decimal::Decimal;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing::{info, warn};
use uuid::Uuid;

pub trait QpSessionRegistry: Send + Sync {
    fn send_rfq_to_eligible(&self, eligible_wallets: &[WalletAddress], message: QpOutboundMessage);

    fn get_connected_wallets(&self) -> Vec<WalletAddress>;
}

/// Testnet option premium tick used for the first RFQ-RPI cutover.
/// Public instrument metadata currently exposes `tick_size = 0.0001`.
const OPTION_PREMIUM_TICK_SCALE: u32 = 4;

fn is_nonce_rejection_reason(reason: &str) -> bool {
    reason.starts_with("nonce ")
}

/// Configuration for the RFQ system.
#[derive(Debug, Clone)]
pub struct RfqConfig {
    pub response_deadline_ms: u64,
    pub rpi_auction_ms: u64,
    pub min_improvement_ticks: u32,
    pub reveal_limit_to_qps: bool,
    pub max_valid_for_ms: u64,
    pub rfq_lifetime_ms: u64,
    pub max_legs: usize,
}

impl Default for RfqConfig {
    fn default() -> Self {
        Self {
            response_deadline_ms: 10_000,
            rpi_auction_ms: 2_000,
            min_improvement_ticks: 1,
            reveal_limit_to_qps: true,
            max_valid_for_ms: 30_000,
            rfq_lifetime_ms: 60_000,
            max_legs: 10,
        }
    }
}

impl RfqConfig {
    pub fn min_improvement_tick(&self) -> Decimal {
        Decimal::new(
            i64::from(self.min_improvement_ticks),
            OPTION_PREMIUM_TICK_SCALE,
        )
    }
}

/// A single leg in an RFQ.
#[derive(Debug, Clone)]
pub struct RfqLeg {
    pub instrument: String,
    pub side: Side,
    pub size: Decimal,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AutoAcceptLimitMode {
    DebitCap,
    CreditFloor,
}

fn auto_accept_limit_mode(legs: &[RfqLeg]) -> Result<AutoAcceptLimitMode, String> {
    let first_side = legs
        .first()
        .ok_or_else(|| "RFQ must have at least one leg".to_string())?
        .side;

    if legs.iter().any(|leg| leg.side != first_side) {
        return Err(
            "Auto-execute RFQ requires all legs on the same side; mixed-side packages require explicit quote acceptance"
                .to_string(),
        );
    }

    Ok(match first_side {
        Side::Buy => AutoAcceptLimitMode::DebitCap,
        Side::Sell => AutoAcceptLimitMode::CreditFloor,
    })
}

fn quote_satisfies_auto_accept_limit(
    legs: &[RfqLeg],
    quote_net_premium: Decimal,
    limit: Decimal,
    _min_tick: Decimal,
) -> Result<bool, String> {
    if limit <= Decimal::ZERO {
        return Err("auto_accept_limit must be positive".to_string());
    }

    match auto_accept_limit_mode(legs)? {
        // Taker buys encode as negative premium. The limit is the maximum
        // debit they authorized. Standard auto-execute RFQ accepts quotes at
        // the exact signed limit; RPI price improvement is checked separately.
        AutoAcceptLimitMode::DebitCap => Ok(quote_net_premium >= -limit),
        // Taker sells encode as positive premium. The limit is the minimum
        // credit they authorized. Standard auto-execute RFQ accepts quotes at
        // the exact signed limit; RPI price improvement is checked separately.
        AutoAcceptLimitMode::CreditFloor => Ok(quote_net_premium >= limit),
    }
}

pub fn quote_qualifies_for_rpi(
    side: Side,
    limit_price: Decimal,
    best_executable_book_price: Option<Decimal>,
    quote_price: Decimal,
    min_tick: Decimal,
) -> bool {
    if limit_price <= Decimal::ZERO || quote_price <= Decimal::ZERO || min_tick <= Decimal::ZERO {
        return false;
    }

    let satisfies_limit = match side {
        Side::Buy => quote_price <= limit_price,
        Side::Sell => quote_price >= limit_price,
    };
    if !satisfies_limit {
        return false;
    }

    match (side, best_executable_book_price) {
        (Side::Buy, Some(best_ask)) => quote_price <= best_ask - min_tick,
        (Side::Sell, Some(best_bid)) => quote_price >= best_bid + min_tick,
        (_, None) => true,
    }
}

fn quote_satisfies_rpi_auction(
    side: Side,
    limit_price: Decimal,
    reference_price: Option<Decimal>,
    quote_price: Decimal,
    min_tick: Decimal,
) -> bool {
    quote_qualifies_for_rpi(side, limit_price, reference_price, quote_price, min_tick)
}

fn rpi_quote_rejection(
    side: Side,
    limit_price: Decimal,
    reference_price: Option<Decimal>,
    quote_price: Decimal,
    min_tick: Decimal,
) -> (&'static str, String) {
    if !quote_qualifies_for_rpi(side, limit_price, None, quote_price, min_tick) {
        return (
            "outside_limit",
            format!(
                "quote price {} does not satisfy limit {}",
                quote_price, limit_price
            ),
        );
    }

    if let Some(reference_price) = reference_price {
        return (
            "no_price_improvement",
            format!(
                "quote price {} does not improve reference {} by tick {}",
                quote_price, reference_price, min_tick
            ),
        );
    }

    (
        "outside_limit",
        format!(
            "quote price {} does not satisfy limit {}",
            quote_price, limit_price
        ),
    )
}

fn validate_rpi_quote_shape(requested_legs: &[RfqLeg], quote: &QpQuote) -> Result<(), String> {
    if quote.legs.len() != requested_legs.len() {
        return Err("RPI quote must cover the full requested leg set".to_string());
    }

    for (requested, quoted) in requested_legs.iter().zip(&quote.legs) {
        if quoted.instrument != requested.instrument {
            return Err(format!(
                "RPI quote instrument {} does not match requested {}",
                quoted.instrument, requested.instrument
            ));
        }
        if quoted.side != requested.side {
            return Err(format!(
                "RPI quote side {:?} does not match requested {:?}",
                quoted.side, requested.side
            ));
        }
        if quoted.size != requested.size {
            return Err(format!(
                "RPI quote size {} does not fully cover requested size {}",
                quoted.size, requested.size
            ));
        }
    }

    Ok(())
}

fn sorted_rpi_candidate_quote_ids(record: &RfqRecord) -> Result<Vec<Uuid>, String> {
    if record.rpi_auction.is_none() {
        return Err("RFQ is not an RPI auction".to_string());
    }
    let side = record
        .legs
        .first()
        .ok_or_else(|| "RPI RFQ missing leg".to_string())?
        .side;

    let mut quotes = record.quotes.clone();
    quotes.sort_by(|a, b| {
        let a_price = a.legs.first().map(|leg| leg.price).unwrap_or(Decimal::ZERO);
        let b_price = b.legs.first().map(|leg| leg.price).unwrap_or(Decimal::ZERO);
        let price_order = match side {
            Side::Buy => a_price.cmp(&b_price),
            Side::Sell => b_price.cmp(&a_price),
        };
        price_order.then_with(|| a.received_at.cmp(&b.received_at))
    });

    Ok(quotes.into_iter().map(|q| q.quote_id).collect())
}

fn should_close_rpi_auction(record: &RfqRecord, now_ms: u64) -> bool {
    let all_qps_responded = !record.eligible_qps.is_empty()
        && record
            .eligible_qps
            .iter()
            .all(|wallet| record.responded_qps.contains(wallet));
    let no_qps = record.eligible_qps.is_empty();
    let timed_out = now_ms > record.deadline_at;

    no_qps || all_qps_responded || timed_out || record.status.is_terminal()
}

/// A firm quote from a QP.
#[derive(Debug, Clone)]
pub struct QpQuote {
    pub quote_id: Uuid,
    pub qp_wallet: WalletAddress,
    pub legs: Vec<QuoteLeg>,
    pub net_premium: Decimal,
    pub valid_for_ms: u64,
    pub qp_signature: String,
    pub qp_nonce: u64,
    pub received_at: u64,
}

#[derive(Debug, Clone)]
pub struct QuoteLeg {
    pub instrument: String,
    pub side: Side,
    pub price: Decimal,
    pub size: Decimal,
}

#[derive(Debug, Clone)]
pub struct RpiAuctionContext {
    pub limit_price: Decimal,
    pub reference_price: Option<Decimal>,
    pub min_tick: Decimal,
}

#[derive(Debug, Clone)]
pub struct SubmitRpiAuction {
    pub rfq_id: Uuid,
    pub taker_wallet: WalletAddress,
    pub taker_signer: WalletAddress,
    pub builder_code_address: Option<WalletAddress>,
    pub legs: Vec<RfqLeg>,
    pub legs_hash: [u8; 32],
    pub taker_signature: String,
    pub taker_nonce: u64,
    pub limit_price: Decimal,
    pub reference_price: Option<Decimal>,
    pub min_tick: Decimal,
}

/// In-memory RFQ record tracking the full lifecycle.
#[derive(Debug, Clone)]
pub struct RfqRecord {
    pub rfq_id: Uuid,
    pub taker_wallet: WalletAddress,
    pub taker_signer: WalletAddress,
    pub builder_code_address: Option<WalletAddress>,
    pub underlying: String,
    pub legs: Vec<RfqLeg>,
    pub legs_hash: [u8; 32],
    pub taker_signature: String,
    pub taker_nonce: u64,
    pub status: RfqStatus,
    pub quotes: Vec<QpQuote>,
    pub created_at: u64,
    pub deadline_at: u64,
    pub expires_at: u64,
    /// When set, the first quote satisfying the taker's directional limit
    /// triggers immediate auto-execution without a separate accept RTT.
    /// Buy RFQs use this as a max debit. Sell RFQs use it as a min credit.
    pub auto_accept_limit: Option<Decimal>,
    /// The quote that won the RFQ (set on accept or auto-accept).
    /// Used to reliably identify the winning quote for late-arriving
    /// QPs, instead of relying on `quotes.last()`.
    pub accepted_quote_id: Option<Uuid>,
    pub execution_quote_id: Option<Uuid>,
    pub execution_request_id: Option<String>,
    pub execution_fill_id: Option<String>,
    pub execution_timestamp_ms: Option<u64>,
    pub eligible_qps: Vec<WalletAddress>,
    pub responded_qps: HashSet<WalletAddress>,
    pub rpi_auction: Option<RpiAuctionContext>,
}

impl RfqRecord {
    fn execution_facts_for_quote(
        &mut self,
        quote_id: Uuid,
        timestamp_ms: u64,
    ) -> (String, String, u64) {
        if self.execution_quote_id == Some(quote_id) {
            if let (Some(request_id), Some(fill_id), Some(stored_timestamp_ms)) = (
                &self.execution_request_id,
                &self.execution_fill_id,
                self.execution_timestamp_ms,
            ) {
                return (request_id.clone(), fill_id.clone(), stored_timestamp_ms);
            }
        }

        let request_id = deterministic_rfq_execution_uuid("request", self.rfq_id, quote_id);
        let fill_id = deterministic_rfq_execution_uuid("fill", self.rfq_id, quote_id);
        self.execution_quote_id = Some(quote_id);
        self.execution_request_id = Some(request_id.clone());
        self.execution_fill_id = Some(fill_id.clone());
        self.execution_timestamp_ms = Some(timestamp_ms);
        (request_id, fill_id, timestamp_ms)
    }

    fn clear_execution_selection(&mut self) {
        self.accepted_quote_id = None;
        self.execution_quote_id = None;
        self.execution_request_id = None;
        self.execution_fill_id = None;
        self.execution_timestamp_ms = None;
    }

    fn revert_auto_accept_to_quotes_received(&mut self) {
        self.status = RfqStatus::QuotesReceived;
        self.clear_execution_selection();
        self.auto_accept_limit = None;
    }
}

fn deterministic_rfq_execution_uuid(purpose: &str, rfq_id: Uuid, quote_id: Uuid) -> String {
    use sha3::{Digest, Sha3_256};

    let mut hasher = Sha3_256::new();
    hasher.update(b"rfq-execution:");
    hasher.update(purpose.as_bytes());
    hasher.update(b":");
    hasher.update(rfq_id.as_bytes());
    hasher.update(b":");
    hasher.update(quote_id.as_bytes());
    let digest = hasher.finalize();

    let mut bytes = [0u8; 16];
    bytes.copy_from_slice(&digest[..16]);
    bytes[6] = (bytes[6] & 0x0F) | 0x80;
    bytes[8] = (bytes[8] & 0x3F) | 0x80;
    Uuid::from_bytes(bytes).to_string()
}

/// Result of `handle_qp_response` distinguishing a simple quote store
/// from an auto-execution trigger.
#[derive(Debug)]
pub enum HandleQpResponseResult {
    /// Quote stored normally; taker must explicitly accept.
    QuoteStored(RfqRecord),
    /// The quote was within the taker's pre-authorized limit and the RFQ
    /// has been atomically transitioned to Accepted. The caller should
    /// dispatch the execution command to the engine.
    AutoExecuted {
        record: RfqRecord,
        accepted_quote_id: Uuid,
    },
    /// The taker had auto_accept_limit set but the quote exceeded it.
    /// The caller should notify the taker so they can fall back immediately
    /// instead of waiting for the timeout.
    AutoExecuteSkipped {
        record: RfqRecord,
        quote_premium: Decimal,
        limit: Decimal,
    },
    /// RPI quote was accepted as an auction candidate. Final execution waits
    /// until the auction closes so the best price wins.
    RpiQuoteStored(RfqRecord),
    /// QP responded, but the quote did not satisfy the mandatory price
    /// improvement rule, or was otherwise not a full-size RPI candidate.
    RpiQuoteRejected { record: RfqRecord, reason: String },
}

/// Manages RFQ lifecycle: creation, QP fan-out, quote collection, acceptance.
pub struct RfqManager {
    active_rfqs: DashMap<Uuid, RfqRecord>,
    qp_cache: Arc<QuoteProviderCache>,
    qp_sessions: Arc<dyn QpSessionRegistry>,
    rfq_sender: mpsc::Sender<RfqExecuteRequest>,
    db: Option<Arc<dyn RfqWriter>>,
    config: RfqConfig,
}

impl RfqManager {
    pub fn new(
        qp_cache: Arc<QuoteProviderCache>,
        qp_sessions: Arc<dyn QpSessionRegistry>,
        rfq_sender: mpsc::Sender<RfqExecuteRequest>,
        config: RfqConfig,
    ) -> Self {
        Self {
            active_rfqs: DashMap::new(),
            qp_cache,
            qp_sessions,
            rfq_sender,
            db: None,
            config,
        }
    }

    pub fn min_improvement_tick(&self) -> Decimal {
        self.config.min_improvement_tick()
    }

    /// Persist an RFQ status change to the `rfq_records` table. Non-fatal on
    /// failure — logs a warning and keeps going so in-memory state
    /// transitions stay responsive even if the DB hiccups. Callers MUST NOT
    /// hold a DashMap lock (e.g. `get_mut` guard or `alter_all` closure) while
    /// invoking this; drop the entry first.
    fn persist_status(&self, rfq_id: &Uuid, status: RfqStatus) {
        if let Some(ref handler) = self.db {
            if let Err(e) = handler.update_rfq_status_sync(rfq_id, status.as_str()) {
                warn!(
                    "Failed to persist RFQ {} status {}: {}",
                    rfq_id,
                    status.as_str(),
                    e
                );
            }
        }
    }

    /// Set the persistence writer for RFQ records and quotes.
    pub fn set_db(&mut self, handler: Arc<dyn RfqWriter>) {
        self.db = Some(handler);
    }

    /// Submit a new RFQ from a taker.
    pub async fn submit_rfq(
        &self,
        rfq_id: Uuid,
        taker_wallet: WalletAddress,
        taker_signer: WalletAddress,
        legs: Vec<RfqLeg>,
        legs_hash: [u8; 32],
        taker_signature: String,
        taker_nonce: u64,
        auto_accept_limit: Option<Decimal>,
    ) -> Result<RfqRecord, String> {
        // Validate leg count
        if legs.is_empty() || legs.len() > self.config.max_legs {
            return Err(format!(
                "Invalid number of legs: {} (must be 1-{})",
                legs.len(),
                self.config.max_legs
            ));
        }

        // Extract underlying (all legs must share the same one)
        let underlying = legs[0]
            .instrument
            .split('-')
            .next()
            .ok_or("Invalid instrument format")?
            .to_string();

        for leg in &legs[1..] {
            let leg_underlying = leg
                .instrument
                .split('-')
                .next()
                .ok_or("Invalid instrument format")?;
            if leg_underlying != underlying {
                return Err("All legs must share the same underlying".to_string());
            }
        }

        if let Some(limit) = auto_accept_limit {
            if limit <= Decimal::ZERO {
                return Err("auto_accept_limit must be positive".to_string());
            }
            auto_accept_limit_mode(&legs)?;
        }

        let now = current_time_ms();
        let response_deadline_ms = self.config.response_deadline_ms;
        let record = RfqRecord {
            rfq_id,
            taker_wallet,
            taker_signer,
            builder_code_address: None,
            underlying: underlying.clone(),
            legs,
            legs_hash,
            taker_signature,
            taker_nonce,
            status: RfqStatus::Created,
            quotes: Vec::new(),
            created_at: now,
            deadline_at: now + response_deadline_ms,
            expires_at: now + self.config.rfq_lifetime_ms,
            auto_accept_limit,
            accepted_quote_id: None,
            execution_quote_id: None,
            execution_request_id: None,
            execution_fill_id: None,
            execution_timestamp_ms: None,
            eligible_qps: Vec::new(),
            responded_qps: HashSet::new(),
            rpi_auction: None,
        };

        // Insert atomically; reject if the caller reused an rfq_id. The
        // DB header row is persisted with `ON CONFLICT (rfq_id) DO
        // NOTHING` (see `persist_rfq_record_sync`), so overwriting the
        // in-memory entry while the durable header stayed bound to the
        // original RFQ would splice new quotes/execution onto stale
        // persisted metadata and break taker flow consistency. Fail
        // fast before any mutation.
        use dashmap::mapref::entry::Entry;
        match self.active_rfqs.entry(rfq_id) {
            Entry::Occupied(_) => {
                return Err(format!("RFQ {} already exists", rfq_id));
            }
            Entry::Vacant(slot) => {
                slot.insert(record.clone());
            }
        }

        // Persist RFQ record + legs to DB
        if let Some(ref handler) = self.db {
            let db_legs: Vec<(String, String, Decimal)> = record
                .legs
                .iter()
                .map(|l| {
                    let side = match l.side {
                        Side::Buy => "buy".to_string(),
                        Side::Sell => "sell".to_string(),
                    };
                    (l.instrument.clone(), side, l.size)
                })
                .collect();
            if let Err(e) = handler.persist_rfq_record_sync(
                &rfq_id,
                &taker_wallet,
                &underlying,
                "created",
                &record.taker_signature,
                record.taker_nonce,
                &legs_hash,
                &db_legs,
                record.expires_at,
            ) {
                info!("Failed to persist RFQ record to DB (non-fatal): {}", e);
            }
        }

        // Fan out to eligible QPs
        let eligible_qps = self.qp_cache.get_active_for_underlying(&underlying).await;
        let eligible_wallets: Vec<WalletAddress> = eligible_qps
            .iter()
            .filter(|_qp| {
                // Filter out MMP-frozen QPs
                // Note: is_frozen is async but we need sync here. Use try approach.
                true // MMP check is done async in the engine execution path
            })
            .map(|qp| qp.wallet)
            .collect();

        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            entry.eligible_qps = eligible_wallets.clone();
        }

        if !eligible_wallets.is_empty() {
            let auto_execute = record.auto_accept_limit.is_some();
            let revealed_limit = record
                .auto_accept_limit
                .and_then(|limit| self.config.reveal_limit_to_qps.then(|| limit.to_string()));
            let rfq_msg = QpOutboundMessage::RfqRequest {
                rfq_id: rfq_id.to_string(),
                legs: record
                    .legs
                    .iter()
                    .map(|l| QpRfqLeg {
                        instrument: l.instrument.clone(),
                        side: format!("{:?}", l.side).to_lowercase(),
                        size: l.size.to_string(),
                    })
                    .collect(),
                taker_wallet: taker_wallet.as_hex(),
                request_timestamp: now,
                response_deadline_ms,
                auto_accept_limit: revealed_limit.clone(),
                auto_execute,
                taker_limit_price: revealed_limit,
                reference_price: None,
                min_improvement_tick: None,
                auction_deadline_ms: None,
                requires_price_improvement: false,
            };

            self.qp_sessions
                .send_rfq_to_eligible(&eligible_wallets, rfq_msg);
        }

        // Transition to SENT_TO_QPS
        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            entry.status = RfqStatus::SentToQps;
        }
        self.persist_status(&rfq_id, RfqStatus::SentToQps);

        let record = self
            .active_rfqs
            .get(&rfq_id)
            .map(|r| r.clone())
            .ok_or("RFQ not found after creation")?;

        counter!(
            "ht_rfq_submit_total",
            "underlying" => underlying.clone(),
            "result" => "ok"
        )
        .increment(1);

        info!(
            "RFQ submitted: rfq_id={}, taker={}, underlying={}, legs={}, eligible_qps={}",
            rfq_id,
            taker_wallet.as_hex(),
            underlying,
            record.legs.len(),
            eligible_wallets.len(),
        );

        Ok(record)
    }

    pub async fn submit_rpi_auction(&self, request: SubmitRpiAuction) -> Result<RfqRecord, String> {
        if request.legs.len() != 1 {
            return Err("RPI auction requires exactly one leg".to_string());
        }
        if request.limit_price <= Decimal::ZERO {
            return Err("RPI limit price must be positive".to_string());
        }
        if request
            .reference_price
            .is_some_and(|reference_price| reference_price <= Decimal::ZERO)
        {
            return Err("RPI reference price must be positive when present".to_string());
        }
        if request.min_tick <= Decimal::ZERO {
            return Err("RPI min tick must be positive".to_string());
        }

        let underlying = request.legs[0]
            .instrument
            .split('-')
            .next()
            .ok_or("Invalid instrument format")?
            .to_string();

        let now = current_time_ms();
        let auction_ms = self.config.rpi_auction_ms;
        let record = RfqRecord {
            rfq_id: request.rfq_id,
            taker_wallet: request.taker_wallet,
            taker_signer: request.taker_signer,
            builder_code_address: request.builder_code_address,
            underlying: underlying.clone(),
            legs: request.legs,
            legs_hash: request.legs_hash,
            taker_signature: request.taker_signature,
            taker_nonce: request.taker_nonce,
            status: RfqStatus::Created,
            quotes: Vec::new(),
            created_at: now,
            deadline_at: now + auction_ms,
            expires_at: now + self.config.rfq_lifetime_ms,
            auto_accept_limit: None,
            accepted_quote_id: None,
            execution_quote_id: None,
            execution_request_id: None,
            execution_fill_id: None,
            execution_timestamp_ms: None,
            eligible_qps: Vec::new(),
            responded_qps: HashSet::new(),
            rpi_auction: Some(RpiAuctionContext {
                limit_price: request.limit_price,
                reference_price: request.reference_price,
                min_tick: request.min_tick,
            }),
        };

        use dashmap::mapref::entry::Entry;
        match self.active_rfqs.entry(record.rfq_id) {
            Entry::Occupied(_) => {
                return Err(format!("RFQ {} already exists", record.rfq_id));
            }
            Entry::Vacant(slot) => {
                slot.insert(record.clone());
            }
        }

        if let Some(ref handler) = self.db {
            let db_legs: Vec<(String, String, Decimal)> = record
                .legs
                .iter()
                .map(|l| {
                    let side = match l.side {
                        Side::Buy => "buy".to_string(),
                        Side::Sell => "sell".to_string(),
                    };
                    (l.instrument.clone(), side, l.size)
                })
                .collect();
            if let Err(e) = handler.persist_rfq_record_sync(
                &record.rfq_id,
                &record.taker_wallet,
                &underlying,
                "created",
                &record.taker_signature,
                record.taker_nonce,
                &record.legs_hash,
                &db_legs,
                record.expires_at,
            ) {
                info!("Failed to persist RPI RFQ record to DB (non-fatal): {}", e);
            }
        }

        let eligible_qps = self.qp_cache.get_active_for_underlying(&underlying).await;
        let connected_wallets: HashSet<WalletAddress> = self
            .qp_sessions
            .get_connected_wallets()
            .into_iter()
            .collect();
        let eligible_wallets: Vec<WalletAddress> = eligible_qps
            .iter()
            .map(|qp| qp.wallet)
            .filter(|wallet| connected_wallets.contains(wallet))
            .collect();

        if let Some(mut entry) = self.active_rfqs.get_mut(&record.rfq_id) {
            entry.eligible_qps = eligible_wallets.clone();
        }

        if !eligible_wallets.is_empty() {
            let rpi = record
                .rpi_auction
                .as_ref()
                .ok_or("RPI context missing after record creation")?;
            let rfq_msg = QpOutboundMessage::RfqRequest {
                rfq_id: record.rfq_id.to_string(),
                legs: record
                    .legs
                    .iter()
                    .map(|l| QpRfqLeg {
                        instrument: l.instrument.clone(),
                        side: format!("{:?}", l.side).to_lowercase(),
                        size: l.size.to_string(),
                    })
                    .collect(),
                taker_wallet: record.taker_wallet.as_hex(),
                request_timestamp: now,
                response_deadline_ms: auction_ms,
                auto_accept_limit: None,
                auto_execute: true,
                taker_limit_price: self
                    .config
                    .reveal_limit_to_qps
                    .then(|| rpi.limit_price.to_string()),
                reference_price: rpi.reference_price.map(|price| price.to_string()),
                min_improvement_tick: rpi.reference_price.map(|_| rpi.min_tick.to_string()),
                auction_deadline_ms: Some(auction_ms),
                requires_price_improvement: rpi.reference_price.is_some(),
            };
            self.qp_sessions
                .send_rfq_to_eligible(&eligible_wallets, rfq_msg);
        }

        if let Some(mut entry) = self.active_rfqs.get_mut(&record.rfq_id) {
            entry.status = RfqStatus::SentToQps;
        }
        self.persist_status(&record.rfq_id, RfqStatus::SentToQps);

        counter!(
            "ht_rpi_auction_total",
            "outcome" => "started",
            "underlying" => underlying.clone()
        )
        .increment(1);

        Ok(self
            .active_rfqs
            .get(&record.rfq_id)
            .map(|r| r.clone())
            .ok_or("RPI RFQ not found after creation")?)
    }

    /// Handle a firm quote response from a QP.
    ///
    /// No margin check here. Quotes are short-lived and non-binding until
    /// accepted, so margin state can change between quote time and acceptance
    /// (new fills, deposits, other RFQs landing). The authoritative SPAN
    /// margin check on both taker and QP runs in `plan_rfq_execution` at
    /// accept time, which cannot be skipped. Adding a check here would add
    /// engine load on every incoming quote for a result that's immediately
    /// stale.
    pub fn handle_qp_response(
        &self,
        rfq_id: Uuid,
        quote: QpQuote,
    ) -> Result<HandleQpResponseResult, String> {
        let mut entry = self.active_rfqs.get_mut(&rfq_id).ok_or("RFQ not found")?;

        // Validate state
        if entry.status != RfqStatus::SentToQps && entry.status != RfqStatus::QuotesReceived {
            return Err(format!(
                "RFQ in invalid state for quotes: {:?}",
                entry.status
            ));
        }

        let now = current_time_ms();
        if now > entry.deadline_at {
            return Err("RFQ response deadline has passed".to_string());
        }

        // Validate valid_for_ms
        if quote.valid_for_ms > self.config.max_valid_for_ms {
            return Err(format!(
                "valid_for_ms {} exceeds max {}",
                quote.valid_for_ms, self.config.max_valid_for_ms
            ));
        }

        if entry.taker_wallet == quote.qp_wallet {
            entry.responded_qps.insert(quote.qp_wallet);
            let snapshot = entry.clone();
            if entry.rpi_auction.is_some() {
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason: RFQ_SELF_TRADE_REJECTION_REASON.to_string(),
                });
            }
            return Err(RFQ_SELF_TRADE_REJECTION_REASON.to_string());
        }

        let rpi_auction = entry.rpi_auction.clone();
        if let Some(rpi) = rpi_auction.as_ref() {
            entry.responded_qps.insert(quote.qp_wallet);
            if let Err(reason) = validate_rpi_quote_shape(&entry.legs, &quote) {
                let snapshot = entry.clone();
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason,
                });
            }
            let Some(first_quote_leg) = quote.legs.first() else {
                let snapshot = entry.clone();
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason: "RPI quote missing quote leg".to_string(),
                });
            };
            if !quote_satisfies_rpi_auction(
                first_quote_leg.side,
                rpi.limit_price,
                rpi.reference_price,
                first_quote_leg.price,
                rpi.min_tick,
            ) {
                let snapshot = entry.clone();
                let (reason_label, reason) = rpi_quote_rejection(
                    first_quote_leg.side,
                    rpi.limit_price,
                    rpi.reference_price,
                    first_quote_leg.price,
                    rpi.min_tick,
                );
                counter!("ht_rpi_quote_rejected_total", "reason" => reason_label).increment(1);
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason,
                });
            }
        } else {
            entry.responded_qps.insert(quote.qp_wallet);
        }

        // Persist quote to DB
        if let Some(ref handler) = self.db {
            let db_legs: Vec<(String, String, Decimal, Decimal)> = quote
                .legs
                .iter()
                .map(|l| {
                    let side = match l.side {
                        Side::Buy => "buy".to_string(),
                        Side::Sell => "sell".to_string(),
                    };
                    (l.instrument.clone(), side, l.price, l.size)
                })
                .collect();
            let expires_at_ms = quote.received_at + quote.valid_for_ms;
            if let Err(e) = handler.persist_rfq_quote_sync(
                &quote.quote_id,
                &rfq_id,
                &quote.qp_wallet,
                quote.net_premium,
                quote.valid_for_ms,
                &quote.qp_signature,
                quote.qp_nonce,
                &db_legs,
                expires_at_ms,
            ) {
                info!("Failed to persist RFQ quote to DB (non-fatal): {}", e);
            }
        }

        // Store quote and transition; snapshot while holding the lock.
        let quote_id = quote.quote_id;
        let quote_net_premium = quote.net_premium;
        entry.quotes.push(quote);
        let transitioned_to_quotes_received = entry.status == RfqStatus::SentToQps;
        if transitioned_to_quotes_received {
            entry.status = RfqStatus::QuotesReceived;
        }

        if rpi_auction.is_some() {
            let snapshot = entry.clone();
            drop(entry);
            if transitioned_to_quotes_received {
                self.persist_status(&rfq_id, RfqStatus::QuotesReceived);
            }
            return Ok(HandleQpResponseResult::RpiQuoteStored(snapshot));
        }

        // Check auto-accept directionally. Buy RFQs authorize a max debit,
        // while sell RFQs authorize a min credit. Atomically transition to
        // Accepted under the DashMap lock so only one QP can win.
        let auto_accept_limit = entry.auto_accept_limit;
        let auto_execute = if let Some(limit) = auto_accept_limit {
            let within_limit = match quote_satisfies_auto_accept_limit(
                &entry.legs,
                quote_net_premium,
                limit,
                self.config.min_improvement_tick(),
            ) {
                Ok(within_limit) => within_limit,
                Err(reason) => {
                    warn!(
                        "RFQ {} auto-execute SKIPPED: {} net_premium={} limit={}",
                        rfq_id, reason, quote_net_premium, limit
                    );
                    false
                }
            };

            if within_limit
                && (entry.status == RfqStatus::QuotesReceived
                    || entry.status == RfqStatus::SentToQps)
            {
                entry.status = RfqStatus::Accepted;
                entry.accepted_quote_id = Some(quote_id);
                true
            } else {
                info!(
                    "RFQ {} auto-execute SKIPPED: net_premium={} limit={} status={:?}",
                    rfq_id, quote_net_premium, limit, entry.status,
                );
                false
            }
        } else {
            false
        };

        let snapshot = entry.clone();
        drop(entry);

        if auto_execute {
            self.persist_status(&rfq_id, RfqStatus::Accepted);
            counter!("ht_rfq_auto_execute_total", "result" => "triggered").increment(1);
            info!(
                "RFQ {} auto-executing: quote {} net_premium={} within limit",
                rfq_id, quote_id, quote_net_premium
            );
            Ok(HandleQpResponseResult::AutoExecuted {
                record: snapshot,
                accepted_quote_id: quote_id,
            })
        } else {
            if transitioned_to_quotes_received {
                self.persist_status(&rfq_id, RfqStatus::QuotesReceived);
            }
            if let Some(limit) = auto_accept_limit {
                Ok(HandleQpResponseResult::AutoExecuteSkipped {
                    record: snapshot,
                    quote_premium: quote_net_premium.abs(),
                    limit,
                })
            } else {
                Ok(HandleQpResponseResult::QuoteStored(snapshot))
            }
        }
    }

    pub fn record_qp_invalid_response(&self, rfq_id: Uuid, qp_wallet: WalletAddress) {
        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            if entry.rpi_auction.is_some() {
                entry.responded_qps.insert(qp_wallet);
            }
        }
    }

    /// Accept a quote. Uses get_mut() for atomic state transition (prevents double-accept).
    /// DashMap::get_mut() holds an exclusive shard lock for the key, so the read-check-write
    /// within the block is atomic with respect to other accesses to the same key.
    pub async fn accept_quote(
        &self,
        rfq_id: Uuid,
        quote_id: Uuid,
        taker_wallet: WalletAddress,
        taker_signature: String,
        taker_signer: WalletAddress,
        taker_nonce: u64,
    ) -> Result<RfqExecuteResult, String> {
        // Atomic check-and-transition using DashMap::entry
        let (cmd, was_accepted) = {
            let mut entry = self.active_rfqs.get_mut(&rfq_id).ok_or("RFQ not found")?;

            // Validate state atomically
            if entry.status != RfqStatus::QuotesReceived {
                return Err(format!(
                    "RFQ in invalid state for acceptance: {:?}",
                    entry.status
                ));
            }

            if entry.taker_wallet != taker_wallet {
                return Err("Taker wallet mismatch".to_string());
            }

            // Find the quote
            let quote = entry
                .quotes
                .iter()
                .find(|q| q.quote_id == quote_id)
                .ok_or("Quote not found")?
                .clone();

            // Check quote validity
            let now = current_time_ms();
            let quote_expires_at = quote.received_at + quote.valid_for_ms;
            if now > quote_expires_at {
                return Err("Quote has expired".to_string());
            }

            // Atomically transition to ACCEPTED (prevents double-accept)
            entry.status = RfqStatus::Accepted;
            entry.accepted_quote_id = Some(quote_id);
            let (request_id, fill_id, execution_timestamp_ms) =
                entry.execution_facts_for_quote(quote_id, now);

            // Build execution command
            let legs: Vec<RfqExecuteLeg> = quote
                .legs
                .iter()
                .map(|l| RfqExecuteLeg {
                    instrument: l.instrument.clone(),
                    taker_side: l.side,
                    price: l.price,
                    size: l.size,
                })
                .collect();

            let cmd = RfqExecuteCommand {
                request_id,
                fill_id,
                rfq_id: rfq_id.to_string(),
                quote_id: quote_id.to_string(),
                taker_wallet,
                qp_wallet: quote.qp_wallet,
                builder_code_address: entry.builder_code_address,
                timestamp_ms: execution_timestamp_ms,
                legs,
                net_premium: quote.net_premium,
                taker_signature,
                qp_signature: quote.qp_signature.clone(),
                taker_nonce: Some(entry.taker_nonce),
                taker_accept_nonce: Some(taker_nonce),
                taker_submit_signer: Some(entry.taker_signer),
                taker_accept_signer: Some(taker_signer),
            };

            (cmd, true)
        };

        if !was_accepted {
            return Err("Failed to accept quote".to_string());
        }
        self.persist_status(&rfq_id, RfqStatus::Accepted);

        // Dispatch to engine. If the engine channel is closed or the
        // response channel is dropped (e.g. engine restart mid-request),
        // the in-memory and durable RFQ state are still `Accepted` from
        // the atomic transition above. Revert to `QuotesReceived` on any
        // failure so the caller can retry instead of leaving the RFQ
        // permanently stuck in `Accepted` (which the state check at the
        // top of this function would then forever reject as "invalid
        // state for acceptance", and which `cleanup_expired` does NOT
        // transition out of).
        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
        let request = RfqExecuteRequest {
            command: cmd,
            response_tx,
        };

        let revert_to_quotes_received = |manager: &Self| {
            if let Some(mut entry) = manager.active_rfqs.get_mut(&rfq_id) {
                entry.status = RfqStatus::QuotesReceived;
                entry.accepted_quote_id = None;
            }
            manager.persist_status(&rfq_id, RfqStatus::QuotesReceived);
        };

        if self.rfq_sender.send(request).await.is_err() {
            revert_to_quotes_received(self);
            return Err("Engine channel closed".to_string());
        }

        let result = match response_rx.await {
            Ok(r) => r,
            Err(_) => {
                revert_to_quotes_received(self);
                return Err("Engine response channel closed".to_string());
            }
        };

        if let RfqExecuteResult::Failed { reason } = &result {
            if is_nonce_rejection_reason(reason) {
                revert_to_quotes_received(self);
                return Err(format!("RFQ execution rejected: {}", reason));
            }
        }

        // Update status based on result
        let terminal_status = match &result {
            RfqExecuteResult::Success { .. } => RfqStatus::Executed,
            RfqExecuteResult::Failed { .. } => RfqStatus::Failed,
        };
        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            entry.status = terminal_status;
        }
        self.persist_status(&rfq_id, terminal_status);

        Ok(result)
    }

    /// Execute an auto-accepted RFQ quote. Called from a spawned task after
    /// `handle_qp_response` returns `AutoExecuted`. The taker already
    /// authorized execution via the `SubmitAutoExecuteRfq` EIP-712
    /// signature, so no separate accept signature is needed.
    pub async fn auto_accept_quote(
        &self,
        rfq_id: Uuid,
        quote_id: Uuid,
        taker_signature: String,
    ) -> Result<RfqExecuteResult, String> {
        // Build the execution command from the already-Accepted record.
        let cmd = {
            let mut entry = self.active_rfqs.get_mut(&rfq_id).ok_or("RFQ not found")?;
            if entry.status != RfqStatus::Accepted {
                return Err(format!(
                    "RFQ in unexpected state for auto-accept execution: {:?}",
                    entry.status
                ));
            }

            let quote = entry
                .quotes
                .iter()
                .find(|q| q.quote_id == quote_id)
                .ok_or("Quote not found for auto-accept")?
                .clone();

            // Reject if the quote has expired between the instant it was
            // stored (handle_qp_response) and when we dispatch execution.
            let now = current_time_ms();
            let quote_expires_at = quote.received_at + quote.valid_for_ms;
            if now > quote_expires_at {
                // Revert to QuotesReceived so subsequent quotes can still
                // be accepted or the taker can manually pick one.
                entry.revert_auto_accept_to_quotes_received();
                drop(entry);
                self.persist_status(&rfq_id, RfqStatus::QuotesReceived);
                return Err("Quote has expired".to_string());
            }

            let legs: Vec<RfqExecuteLeg> = quote
                .legs
                .iter()
                .map(|l| RfqExecuteLeg {
                    instrument: l.instrument.clone(),
                    taker_side: l.side,
                    price: l.price,
                    size: l.size,
                })
                .collect();
            let (request_id, fill_id, execution_timestamp_ms) =
                entry.execution_facts_for_quote(quote_id, now);

            RfqExecuteCommand {
                request_id,
                fill_id,
                rfq_id: rfq_id.to_string(),
                quote_id: quote_id.to_string(),
                taker_wallet: entry.taker_wallet,
                qp_wallet: quote.qp_wallet,
                builder_code_address: entry.builder_code_address,
                timestamp_ms: execution_timestamp_ms,
                legs,
                net_premium: quote.net_premium,
                taker_signature,
                qp_signature: quote.qp_signature.clone(),
                taker_nonce: Some(entry.taker_nonce),
                taker_accept_nonce: None,
                taker_submit_signer: Some(entry.taker_signer),
                taker_accept_signer: None,
            }
        }; // DashMap read guard dropped here

        // Dispatch to engine
        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
        let request = RfqExecuteRequest {
            command: cmd,
            response_tx,
        };

        let revert_to_quotes_received = |manager: &Self| {
            if let Some(mut entry) = manager.active_rfqs.get_mut(&rfq_id) {
                entry.revert_auto_accept_to_quotes_received();
            }
            manager.persist_status(&rfq_id, RfqStatus::QuotesReceived);
        };

        if self.rfq_sender.send(request).await.is_err() {
            revert_to_quotes_received(self);
            return Err("Engine channel closed".to_string());
        }

        let result = match response_rx.await {
            Ok(r) => r,
            Err(_) => {
                revert_to_quotes_received(self);
                return Err("Engine response channel closed".to_string());
            }
        };

        if let RfqExecuteResult::Failed { reason } = &result {
            if is_nonce_rejection_reason(reason) {
                revert_to_quotes_received(self);
                return Err(format!("RFQ execution rejected: {}", reason));
            }
        }

        // Update status based on result
        let terminal_status = match &result {
            RfqExecuteResult::Success { .. } => RfqStatus::Executed,
            RfqExecuteResult::Failed { .. } => RfqStatus::Failed,
        };
        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            entry.status = terminal_status;
        }
        self.persist_status(&rfq_id, terminal_status);

        Ok(result)
    }

    pub fn record_qp_decline(&self, rfq_id: Uuid, qp_wallet: WalletAddress) {
        if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
            entry.responded_qps.insert(qp_wallet);
        }
    }

    pub async fn wait_for_rpi_auction_close(&self, rfq_id: Uuid) -> Result<RfqRecord, String> {
        loop {
            let record = self
                .active_rfqs
                .get(&rfq_id)
                .map(|r| r.clone())
                .ok_or_else(|| "RPI RFQ not found".to_string())?;
            if record.rpi_auction.is_none() {
                return Err("RFQ is not an RPI auction".to_string());
            }

            if should_close_rpi_auction(&record, current_time_ms()) {
                histogram!(
                    "ht_rpi_auction_duration_ms",
                    "outcome" => if record.quotes.is_empty() { "fallback_to_book" } else { "candidate_quotes" }
                )
                .record(current_time_ms().saturating_sub(record.created_at) as f64);
                return Ok(record);
            }

            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }

    pub fn rpi_candidate_quote_ids(&self, rfq_id: &Uuid) -> Result<Vec<Uuid>, String> {
        let record = self
            .active_rfqs
            .get(rfq_id)
            .map(|r| r.clone())
            .ok_or_else(|| "RPI RFQ not found".to_string())?;
        sorted_rpi_candidate_quote_ids(&record)
    }

    pub async fn execute_rpi_quote(
        &self,
        rfq_id: Uuid,
        quote_id: Uuid,
        taker_signature: String,
    ) -> Result<RfqExecuteResult, String> {
        let cmd = {
            let mut entry = self.active_rfqs.get_mut(&rfq_id).ok_or("RFQ not found")?;
            if entry.rpi_auction.is_none() {
                return Err("RFQ is not an RPI auction".to_string());
            }
            if entry.status != RfqStatus::SentToQps && entry.status != RfqStatus::QuotesReceived {
                return Err(format!(
                    "RFQ in invalid state for RPI execution: {:?}",
                    entry.status
                ));
            }

            let quote = entry
                .quotes
                .iter()
                .find(|q| q.quote_id == quote_id)
                .ok_or("Quote not found")?
                .clone();

            let now = current_time_ms();
            let quote_expires_at = quote.received_at + quote.valid_for_ms;
            if now > quote_expires_at {
                return Err("Quote has expired".to_string());
            }

            entry.status = RfqStatus::Accepted;
            entry.accepted_quote_id = Some(quote_id);
            let (request_id, fill_id, execution_timestamp_ms) =
                entry.execution_facts_for_quote(quote_id, now);

            let legs: Vec<RfqExecuteLeg> = quote
                .legs
                .iter()
                .map(|l| RfqExecuteLeg {
                    instrument: l.instrument.clone(),
                    taker_side: l.side,
                    price: l.price,
                    size: l.size,
                })
                .collect();

            RfqExecuteCommand {
                request_id,
                fill_id,
                rfq_id: rfq_id.to_string(),
                quote_id: quote_id.to_string(),
                taker_wallet: entry.taker_wallet,
                qp_wallet: quote.qp_wallet,
                builder_code_address: entry.builder_code_address,
                timestamp_ms: execution_timestamp_ms,
                legs,
                net_premium: quote.net_premium,
                taker_signature,
                qp_signature: quote.qp_signature.clone(),
                taker_nonce: Some(entry.taker_nonce),
                taker_accept_nonce: None,
                taker_submit_signer: Some(entry.taker_signer),
                taker_accept_signer: None,
            }
        };
        self.persist_status(&rfq_id, RfqStatus::Accepted);

        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
        let request = RfqExecuteRequest {
            command: cmd,
            response_tx,
        };

        let revert_to_quotes_received = |manager: &Self| {
            if let Some(mut entry) = manager.active_rfqs.get_mut(&rfq_id) {
                entry.status = RfqStatus::QuotesReceived;
                entry.clear_execution_selection();
            }
            manager.persist_status(&rfq_id, RfqStatus::QuotesReceived);
        };

        if self.rfq_sender.send(request).await.is_err() {
            revert_to_quotes_received(self);
            return Err("Engine channel closed".to_string());
        }

        let result = match response_rx.await {
            Ok(r) => r,
            Err(_) => {
                revert_to_quotes_received(self);
                return Err("Engine response channel closed".to_string());
            }
        };

        match &result {
            RfqExecuteResult::Success { .. } => {
                if let Some(mut entry) = self.active_rfqs.get_mut(&rfq_id) {
                    entry.status = RfqStatus::Executed;
                }
                self.persist_status(&rfq_id, RfqStatus::Executed);
                counter!("ht_rpi_auction_total", "outcome" => "filled").increment(1);
            }
            RfqExecuteResult::Failed { .. } => {
                revert_to_quotes_received(self);
            }
        }

        Ok(result)
    }

    pub fn mark_rpi_fallback_to_book(&self, rfq_id: &Uuid) {
        if let Some(mut entry) = self.active_rfqs.get_mut(rfq_id) {
            entry.status = if entry.quotes.is_empty() {
                RfqStatus::NoQuotes
            } else {
                RfqStatus::Failed
            };
        }
        let status = self
            .active_rfqs
            .get(rfq_id)
            .map(|entry| entry.status)
            .unwrap_or(RfqStatus::Failed);
        self.persist_status(rfq_id, status);
        counter!("ht_rpi_auction_total", "outcome" => "fallback_to_book").increment(1);
    }

    /// Get an RFQ record by ID.
    pub fn get_rfq(&self, rfq_id: &Uuid) -> Option<RfqRecord> {
        self.active_rfqs.get(rfq_id).map(|r| r.clone())
    }

    /// Get RFQ history for a wallet.
    pub fn get_history(&self, wallet: &WalletAddress) -> Vec<RfqRecord> {
        self.active_rfqs
            .iter()
            .filter(|r| r.taker_wallet == *wallet)
            .map(|r| r.clone())
            .collect()
    }

    /// Start a background cleanup task that expires stale RFQs.
    pub fn start_cleanup_task(
        self: &Arc<Self>,
        mut shutdown: tokio::sync::broadcast::Receiver<()>,
    ) {
        let manager = Arc::clone(self);
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
            loop {
                tokio::select! {
                    _ = shutdown.recv() => break,
                    _ = interval.tick() => {
                        manager.cleanup_expired();
                    }
                }
            }
        });
    }

    /// Expire stale RFQs and evict old terminal records.
    fn cleanup_expired(&self) {
        let now = current_time_ms();

        // Collect transitions inside the lock, then persist them to DB after
        // the `alter_all` closure returns so we don't hold the DashMap shard
        // lock across a sync DB call.
        let transitions = std::sync::Mutex::new(Vec::<(Uuid, RfqStatus)>::new());
        self.active_rfqs.alter_all(|rfq_id, mut record| {
            let new_status = match record.status {
                RfqStatus::SentToQps if now > record.deadline_at => {
                    if record.quotes.is_empty() {
                        Some(RfqStatus::NoQuotes)
                    } else {
                        Some(RfqStatus::QuotesReceived)
                    }
                }
                RfqStatus::QuotesReceived if now > record.expires_at => Some(RfqStatus::Expired),
                _ => None,
            };
            if let Some(status) = new_status {
                record.status = status;
                if let Ok(mut t) = transitions.lock() {
                    t.push((*rfq_id, status));
                }
            }
            record
        });

        if let Ok(t) = transitions.into_inner() {
            for (rfq_id, status) in t {
                self.persist_status(&rfq_id, status);
            }
        }

        // Evict terminal RFQs older than 5 minutes
        let eviction_threshold = now.saturating_sub(5 * 60 * 1000);
        self.active_rfqs.retain(|_, record| {
            !(record.status.is_terminal() && record.created_at < eviction_threshold)
        });
    }
}

/// Test-only handle that wraps an `RfqManager` for unit tests, providing
/// direct access to internal state without requiring a real database or
/// quote provider cache.
#[cfg(test)]
struct TestRfqManager {
    active_rfqs: Arc<DashMap<Uuid, RfqRecord>>,
    config: RfqConfig,
}

#[cfg(test)]
impl TestRfqManager {
    fn new() -> Self {
        Self {
            active_rfqs: Arc::new(DashMap::new()),
            config: RfqConfig::default(),
        }
    }

    fn insert_rfq(&self, record: RfqRecord) {
        self.active_rfqs.insert(record.rfq_id, record);
    }

    fn get_rfq(&self, rfq_id: &Uuid) -> Option<RfqRecord> {
        self.active_rfqs.get(rfq_id).map(|r| r.clone())
    }

    /// Mirrors `RfqManager::handle_qp_response` logic for testing without
    /// requiring the full manager (which needs DB, QP cache, etc.).
    fn handle_qp_response(
        &self,
        rfq_id: Uuid,
        quote: QpQuote,
    ) -> Result<HandleQpResponseResult, String> {
        let mut entry = self.active_rfqs.get_mut(&rfq_id).ok_or("RFQ not found")?;

        if entry.status != RfqStatus::SentToQps && entry.status != RfqStatus::QuotesReceived {
            return Err(format!(
                "RFQ in invalid state for quotes: {:?}",
                entry.status
            ));
        }

        let now = current_time_ms();
        if now > entry.deadline_at {
            return Err("RFQ response deadline has passed".to_string());
        }

        if quote.valid_for_ms > self.config.max_valid_for_ms {
            return Err(format!(
                "valid_for_ms {} exceeds max {}",
                quote.valid_for_ms, self.config.max_valid_for_ms
            ));
        }

        if entry.taker_wallet == quote.qp_wallet {
            entry.responded_qps.insert(quote.qp_wallet);
            let snapshot = entry.clone();
            if entry.rpi_auction.is_some() {
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason: RFQ_SELF_TRADE_REJECTION_REASON.to_string(),
                });
            }
            return Err(RFQ_SELF_TRADE_REJECTION_REASON.to_string());
        }

        let rpi_auction = entry.rpi_auction.clone();
        if let Some(rpi) = rpi_auction.as_ref() {
            entry.responded_qps.insert(quote.qp_wallet);
            if let Err(reason) = validate_rpi_quote_shape(&entry.legs, &quote) {
                let snapshot = entry.clone();
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason,
                });
            }
            let first_quote_leg = quote
                .legs
                .first()
                .ok_or_else(|| "RPI quote missing quote leg".to_string())?;
            if !quote_satisfies_rpi_auction(
                first_quote_leg.side,
                rpi.limit_price,
                rpi.reference_price,
                first_quote_leg.price,
                rpi.min_tick,
            ) {
                let snapshot = entry.clone();
                let (_, reason) = rpi_quote_rejection(
                    first_quote_leg.side,
                    rpi.limit_price,
                    rpi.reference_price,
                    first_quote_leg.price,
                    rpi.min_tick,
                );
                return Ok(HandleQpResponseResult::RpiQuoteRejected {
                    record: snapshot,
                    reason,
                });
            }
        } else {
            entry.responded_qps.insert(quote.qp_wallet);
        }

        let quote_id = quote.quote_id;
        let quote_net_premium = quote.net_premium;
        entry.quotes.push(quote);
        let transitioned_to_quotes_received = entry.status == RfqStatus::SentToQps;
        if transitioned_to_quotes_received {
            entry.status = RfqStatus::QuotesReceived;
        }

        if rpi_auction.is_some() {
            let snapshot = entry.clone();
            drop(entry);
            return Ok(HandleQpResponseResult::RpiQuoteStored(snapshot));
        }

        let auto_accept_limit = entry.auto_accept_limit;
        let auto_execute = if let Some(limit) = auto_accept_limit {
            let within_limit = quote_satisfies_auto_accept_limit(
                &entry.legs,
                quote_net_premium,
                limit,
                self.config.min_improvement_tick(),
            )
            .unwrap_or(false);

            if within_limit
                && (entry.status == RfqStatus::QuotesReceived
                    || entry.status == RfqStatus::SentToQps)
            {
                entry.status = RfqStatus::Accepted;
                entry.accepted_quote_id = Some(quote_id);
                true
            } else {
                false
            }
        } else {
            false
        };

        let snapshot = entry.clone();
        drop(entry);

        if auto_execute {
            Ok(HandleQpResponseResult::AutoExecuted {
                record: snapshot,
                accepted_quote_id: quote_id,
            })
        } else if let Some(limit) = auto_accept_limit {
            Ok(HandleQpResponseResult::AutoExecuteSkipped {
                record: snapshot,
                quote_premium: quote_net_premium.abs(),
                limit,
            })
        } else {
            Ok(HandleQpResponseResult::QuoteStored(snapshot))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rust_decimal_macros::dec;

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

    /// Create a test RfqRecord in `SentToQps` state (ready to receive quotes).
    fn make_test_rfq(
        rfq_id: Uuid,
        taker_wallet: WalletAddress,
        auto_accept_limit: Option<Decimal>,
    ) -> RfqRecord {
        make_test_rfq_with_side(rfq_id, taker_wallet, auto_accept_limit, Side::Buy)
    }

    fn make_test_rfq_with_side(
        rfq_id: Uuid,
        taker_wallet: WalletAddress,
        auto_accept_limit: Option<Decimal>,
        side: Side,
    ) -> RfqRecord {
        let now = current_time_ms();
        RfqRecord {
            rfq_id,
            taker_wallet,
            taker_signer: taker_wallet,
            builder_code_address: None,
            underlying: "ETH".to_string(),
            legs: vec![RfqLeg {
                instrument: "ETH-20260401-3000-C".to_string(),
                side,
                size: dec!(5),
            }],
            legs_hash: [0u8; 32],
            taker_signature: "0xtest_sig".to_string(),
            taker_nonce: 1,
            status: RfqStatus::SentToQps,
            quotes: Vec::new(),
            created_at: now,
            deadline_at: now + 30_000,
            expires_at: now + 60_000,
            auto_accept_limit,
            accepted_quote_id: None,
            execution_quote_id: None,
            execution_request_id: None,
            execution_fill_id: None,
            execution_timestamp_ms: None,
            eligible_qps: Vec::new(),
            responded_qps: HashSet::new(),
            rpi_auction: None,
        }
    }

    /// Create a test QpQuote with the given net_premium.
    fn make_test_quote(qp_wallet: WalletAddress, net_premium: Decimal) -> QpQuote {
        make_test_quote_with_leg(qp_wallet, net_premium, Side::Buy, dec!(150), dec!(5))
    }

    fn make_test_quote_with_leg(
        qp_wallet: WalletAddress,
        net_premium: Decimal,
        side: Side,
        price: Decimal,
        size: Decimal,
    ) -> QpQuote {
        QpQuote {
            quote_id: Uuid::now_v7(),
            qp_wallet,
            legs: vec![QuoteLeg {
                instrument: "ETH-20260401-3000-C".to_string(),
                side,
                price,
                size,
            }],
            net_premium,
            valid_for_ms: 25_000,
            qp_signature: "0xqp_sig".to_string(),
            qp_nonce: 1,
            received_at: current_time_ms(),
        }
    }

    fn make_test_rpi_rfq(rfq_id: Uuid, taker_wallet: WalletAddress, side: Side) -> RfqRecord {
        let mut record = make_test_rfq_with_side(rfq_id, taker_wallet, None, side);
        record.rpi_auction = Some(RpiAuctionContext {
            limit_price: dec!(100),
            reference_price: Some(dec!(100)),
            min_tick: dec!(0.0001),
        });
        record
    }

    fn make_test_rpi_rfq_without_book_reference(
        rfq_id: Uuid,
        taker_wallet: WalletAddress,
        side: Side,
    ) -> RfqRecord {
        let mut record = make_test_rpi_rfq(rfq_id, taker_wallet, side);
        record
            .rpi_auction
            .as_mut()
            .expect("test RPI context should exist")
            .reference_price = None;
        record
    }

    #[test]
    fn test_rfq_config_default() {
        let config = RfqConfig::default();
        assert_eq!(config.response_deadline_ms, 10_000);
        assert_eq!(config.rpi_auction_ms, 2_000);
        assert_eq!(config.min_improvement_ticks, 1);
        assert!(config.reveal_limit_to_qps);
        assert_eq!(config.max_valid_for_ms, 30_000);
        assert_eq!(config.rfq_lifetime_ms, 60_000);
        assert_eq!(config.max_legs, 10);
    }

    #[test]
    fn test_rfq_record_lifecycle() {
        let record = RfqRecord {
            rfq_id: Uuid::now_v7(),
            taker_wallet: test_wallet(1),
            taker_signer: test_wallet(1),
            builder_code_address: None,
            underlying: "ETH".to_string(),
            legs: vec![RfqLeg {
                instrument: "ETH-20260401-3000-C".to_string(),
                side: Side::Buy,
                size: dec!(5),
            }],
            legs_hash: [0u8; 32],
            taker_signature: "0x123".to_string(),
            taker_nonce: 1,
            status: RfqStatus::Created,
            quotes: Vec::new(),
            created_at: current_time_ms(),
            deadline_at: current_time_ms() + 5000,
            expires_at: current_time_ms() + 60000,
            auto_accept_limit: None,
            accepted_quote_id: None,
            execution_quote_id: None,
            execution_request_id: None,
            execution_fill_id: None,
            execution_timestamp_ms: None,
            eligible_qps: Vec::new(),
            responded_qps: HashSet::new(),
            rpi_auction: None,
        };

        assert_eq!(record.status, RfqStatus::Created);
        assert!(!record.status.is_terminal());
    }

    #[test]
    fn test_rfq_execution_facts_stable_for_quote_retries() {
        let rfq_id = Uuid::now_v7();
        let quote_id = Uuid::now_v7();
        let other_quote_id = Uuid::now_v7();
        let mut record = make_test_rfq(rfq_id, test_wallet(1), None);

        let first = record.execution_facts_for_quote(quote_id, 100);
        let retry = record.execution_facts_for_quote(quote_id, 200);
        assert_eq!(
            first, retry,
            "same accepted quote must reuse execution facts across retries"
        );

        let other_quote = record.execution_facts_for_quote(other_quote_id, 300);
        assert_ne!(
            first.0, other_quote.0,
            "different accepted quotes must not share a request_id"
        );
        assert_ne!(
            first.1, other_quote.1,
            "different accepted quotes must not share a fill_id"
        );
        assert_eq!(other_quote.2, 300);
    }

    #[test]
    fn test_auto_accept_rollback_clears_execution_selection() {
        let rfq_id = Uuid::now_v7();
        let quote_id = Uuid::now_v7();
        let mut record = make_test_rfq(rfq_id, test_wallet(1), Some(dec!(5.0)));

        record.status = RfqStatus::Accepted;
        record.accepted_quote_id = Some(quote_id);
        let _ = record.execution_facts_for_quote(quote_id, 100);

        record.revert_auto_accept_to_quotes_received();

        assert_eq!(record.status, RfqStatus::QuotesReceived);
        assert_eq!(record.auto_accept_limit, None);
        assert_eq!(record.accepted_quote_id, None);
        assert_eq!(record.execution_quote_id, None);
        assert_eq!(record.execution_request_id, None);
        assert_eq!(record.execution_fill_id, None);
        assert_eq!(record.execution_timestamp_ms, None);
    }

    #[test]
    fn test_rfq_status_terminal() {
        assert!(!RfqStatus::Created.is_terminal());
        assert!(!RfqStatus::SentToQps.is_terminal());
        assert!(!RfqStatus::QuotesReceived.is_terminal());
        assert!(!RfqStatus::Accepted.is_terminal());
        assert!(RfqStatus::NoQuotes.is_terminal());
        assert!(RfqStatus::Expired.is_terminal());
        assert!(RfqStatus::Executed.is_terminal());
        assert!(RfqStatus::Failed.is_terminal());
    }

    #[test]
    fn test_auto_execute_rfq() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        // Insert an RFQ with auto_accept_limit = 5.0
        let record = make_test_rfq(rfq_id, taker, Some(dec!(5.0)));
        manager.insert_rfq(record);

        // Taker buy quotes encode as negative premium. A 3.0 debit is within
        // a 5.0 max debit limit.
        let quote = make_test_quote(qp, dec!(-3.0));
        let quote_id = quote.quote_id;

        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        // Should be auto-executed
        match result {
            HandleQpResponseResult::AutoExecuted {
                record,
                accepted_quote_id,
            } => {
                assert_eq!(accepted_quote_id, quote_id);
                assert_eq!(record.status, RfqStatus::Accepted);
            }
            HandleQpResponseResult::QuoteStored(_) => {
                panic!("Expected AutoExecuted, got QuoteStored");
            }
            HandleQpResponseResult::AutoExecuteSkipped { .. } => {
                panic!("Expected AutoExecuted, got AutoExecuteSkipped");
            }
            HandleQpResponseResult::RpiQuoteStored(_)
            | HandleQpResponseResult::RpiQuoteRejected { .. } => {
                panic!("Expected AutoExecuted, got RPI result");
            }
        }

        // Verify the RFQ status is Accepted in the manager
        let stored = manager.get_rfq(&rfq_id).unwrap();
        assert_eq!(stored.status, RfqStatus::Accepted);

        // The auto_accept_quote method would be called by the server to
        // dispatch the execute command. We verify the record state is
        // correct for that dispatch.
        assert_eq!(stored.quotes.len(), 1);
        assert_eq!(stored.quotes[0].quote_id, quote_id);
    }

    #[test]
    fn test_manual_rfq_rejects_self_trade_quote() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);

        manager.insert_rfq(make_test_rfq(rfq_id, taker, None));

        let result = manager.handle_qp_response(rfq_id, make_test_quote(taker, dec!(-3.0)));

        assert_eq!(
            result.unwrap_err(),
            RFQ_SELF_TRADE_REJECTION_REASON.to_string()
        );
        let stored = manager.get_rfq(&rfq_id).unwrap();
        assert_eq!(stored.status, RfqStatus::SentToQps);
        assert_eq!(stored.quotes.len(), 0);
        assert!(stored.responded_qps.contains(&taker));
    }

    #[test]
    fn test_auto_execute_rfq_rejects_self_trade_quote() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);

        manager.insert_rfq(make_test_rfq(rfq_id, taker, Some(dec!(5.0))));

        let result = manager.handle_qp_response(rfq_id, make_test_quote(taker, dec!(-3.0)));

        assert_eq!(
            result.unwrap_err(),
            RFQ_SELF_TRADE_REJECTION_REASON.to_string()
        );
        let stored = manager.get_rfq(&rfq_id).unwrap();
        assert_eq!(stored.status, RfqStatus::SentToQps);
        assert_eq!(stored.accepted_quote_id, None);
        assert_eq!(stored.quotes.len(), 0);
        assert!(stored.responded_qps.contains(&taker));
    }

    #[test]
    fn test_auto_execute_rfq_outside_limit() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        // Insert an RFQ with auto_accept_limit = 2.0
        let record = make_test_rfq(rfq_id, taker, Some(dec!(2.0)));
        manager.insert_rfq(record);

        // A 3.0 debit is outside a 2.0 max debit limit.
        let quote = make_test_quote(qp, dec!(-3.0));

        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        // Should skip auto-execute because quote exceeds limit
        match result {
            HandleQpResponseResult::AutoExecuteSkipped {
                record,
                quote_premium,
                limit,
            } => {
                assert_eq!(record.status, RfqStatus::QuotesReceived);
                assert_eq!(record.quotes.len(), 1);
                assert_eq!(quote_premium, dec!(3.0));
                assert_eq!(limit, dec!(2.0));
            }
            HandleQpResponseResult::QuoteStored(_) => {
                panic!("Expected AutoExecuteSkipped, got QuoteStored");
            }
            HandleQpResponseResult::AutoExecuted { .. } => {
                panic!("Expected AutoExecuteSkipped, got AutoExecuted");
            }
            HandleQpResponseResult::RpiQuoteStored(_)
            | HandleQpResponseResult::RpiQuoteRejected { .. } => {
                panic!("Expected AutoExecuteSkipped, got RPI result");
            }
        }

        // Verify the RFQ status is QuotesReceived (not Accepted)
        let stored = manager.get_rfq(&rfq_id).unwrap();
        assert_eq!(stored.status, RfqStatus::QuotesReceived);
    }

    #[test]
    fn test_auto_execute_prevents_double_fill() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp1 = test_wallet(2);
        let qp2 = test_wallet(3);

        // Insert an RFQ with auto_accept_limit = 5.0
        let record = make_test_rfq(rfq_id, taker, Some(dec!(5.0)));
        manager.insert_rfq(record);

        // First QP responds with a 3.0 debit within the limit -> AutoExecuted
        let quote1 = make_test_quote(qp1, dec!(-3.0));
        let result1 = manager.handle_qp_response(rfq_id, quote1).unwrap();
        assert!(
            matches!(result1, HandleQpResponseResult::AutoExecuted { .. }),
            "First quote should auto-execute"
        );

        // Second QP responds -> should be rejected (RFQ already in Accepted state)
        let quote2 = make_test_quote(qp2, dec!(-2.0));
        let result2 = manager.handle_qp_response(rfq_id, quote2);
        assert!(
            result2.is_err(),
            "Second quote should be rejected because RFQ is already Accepted"
        );
        assert!(
            result2.unwrap_err().contains("invalid state for quotes"),
            "Error should indicate invalid state"
        );
    }

    #[test]
    fn test_auto_execute_sell_quote_below_limit_is_skipped() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        let record = make_test_rfq_with_side(rfq_id, taker, Some(dec!(1.9553)), Side::Sell);
        manager.insert_rfq(record);

        let quote = make_test_quote(qp, dec!(1.0));
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::AutoExecuteSkipped {
                record,
                quote_premium,
                limit,
            } => {
                assert_eq!(record.status, RfqStatus::QuotesReceived);
                assert_eq!(quote_premium, dec!(1.0));
                assert_eq!(limit, dec!(1.9553));
            }
            HandleQpResponseResult::QuoteStored(_) => {
                panic!("Expected AutoExecuteSkipped, got QuoteStored");
            }
            HandleQpResponseResult::AutoExecuted { .. } => {
                panic!("Sell quote below limit must not auto-execute");
            }
            HandleQpResponseResult::RpiQuoteStored(_)
            | HandleQpResponseResult::RpiQuoteRejected { .. } => {
                panic!("Expected AutoExecuteSkipped, got RPI result");
            }
        }
    }

    #[test]
    fn test_auto_execute_sell_quote_at_or_above_limit_executes() {
        let manager = TestRfqManager::new();

        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        let record = make_test_rfq_with_side(rfq_id, taker, Some(dec!(1.9553)), Side::Sell);
        manager.insert_rfq(record);

        let quote = make_test_quote(qp, dec!(1.96));
        let quote_id = quote.quote_id;
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::AutoExecuted {
                record,
                accepted_quote_id,
            } => {
                assert_eq!(accepted_quote_id, quote_id);
                assert_eq!(record.status, RfqStatus::Accepted);
            }
            HandleQpResponseResult::QuoteStored(_) => {
                panic!("Expected AutoExecuted, got QuoteStored");
            }
            HandleQpResponseResult::AutoExecuteSkipped { .. } => {
                panic!("Expected AutoExecuted, got AutoExecuteSkipped");
            }
            HandleQpResponseResult::RpiQuoteStored(_)
            | HandleQpResponseResult::RpiQuoteRejected { .. } => {
                panic!("Expected AutoExecuted, got RPI result");
            }
        }
    }

    #[test]
    fn test_auto_execute_mixed_side_limit_is_rejected() {
        let legs = vec![
            RfqLeg {
                instrument: "ETH-20260401-3000-C".to_string(),
                side: Side::Buy,
                size: dec!(1),
            },
            RfqLeg {
                instrument: "ETH-20260401-3000-P".to_string(),
                side: Side::Sell,
                size: dec!(1),
            },
        ];

        let err = quote_satisfies_auto_accept_limit(&legs, dec!(1), dec!(1), dec!(0.0001))
            .expect_err("mixed-side auto-execute RFQ should be rejected");

        assert!(err.contains("mixed-side packages require explicit quote acceptance"));
    }

    #[test]
    fn test_rpi_buy_accepts_limit_without_book_reference() {
        assert!(quote_qualifies_for_rpi(
            Side::Buy,
            dec!(100),
            None,
            dec!(100),
            dec!(0.0001),
        ));
        assert!(!quote_qualifies_for_rpi(
            Side::Buy,
            dec!(100),
            None,
            dec!(100.0001),
            dec!(0.0001),
        ));
    }

    #[test]
    fn test_rpi_price_improvement_buy_beats_executable_book() {
        assert!(quote_qualifies_for_rpi(
            Side::Buy,
            dec!(105),
            Some(dec!(100)),
            dec!(99.9999),
            dec!(0.0001),
        ));
        assert!(!quote_qualifies_for_rpi(
            Side::Buy,
            dec!(105),
            Some(dec!(100)),
            dec!(104.9999),
            dec!(0.0001),
        ));
    }

    #[test]
    fn test_rpi_price_improvement_sell_beats_executable_book() {
        assert!(quote_qualifies_for_rpi(
            Side::Sell,
            dec!(95),
            Some(dec!(100)),
            dec!(100.0001),
            dec!(0.0001),
        ));
        assert!(!quote_qualifies_for_rpi(
            Side::Sell,
            dec!(95),
            Some(dec!(100)),
            dec!(95.0001),
            dec!(0.0001),
        ));
    }

    #[test]
    fn test_rpi_rejection_prefers_limit_reason_over_reference_reason() {
        let (label, reason) = rpi_quote_rejection(
            Side::Buy,
            dec!(100),
            Some(dec!(99)),
            dec!(100.0001),
            dec!(0.0001),
        );

        assert_eq!(label, "outside_limit");
        assert!(reason.contains("does not satisfy limit"));
    }

    #[test]
    fn test_rpi_rejection_reports_missing_price_improvement() {
        let (label, reason) = rpi_quote_rejection(
            Side::Buy,
            dec!(105),
            Some(dec!(100)),
            dec!(100),
            dec!(0.0001),
        );

        assert_eq!(label, "no_price_improvement");
        assert!(reason.contains("does not improve reference"));
    }

    #[test]
    fn test_rpi_quote_rejects_price_without_improvement() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        manager.insert_rfq(make_test_rpi_rfq(rfq_id, taker, Side::Buy));

        let quote = make_test_quote_with_leg(qp, dec!(-500), Side::Buy, dec!(100), dec!(5));
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteRejected { record, reason } => {
                assert!(reason.contains("does not improve reference"));
                assert_eq!(record.quotes.len(), 0);
                assert!(record.responded_qps.contains(&qp));
            }
            other => panic!("Expected RpiQuoteRejected, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_quote_stores_limit_price_when_no_book_reference() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        manager.insert_rfq(make_test_rpi_rfq_without_book_reference(
            rfq_id,
            taker,
            Side::Buy,
        ));

        let quote = make_test_quote_with_leg(qp, dec!(-500), Side::Buy, dec!(100), dec!(5));
        let quote_id = quote.quote_id;
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteStored(record) => {
                assert_eq!(record.status, RfqStatus::QuotesReceived);
                assert_eq!(record.quotes.len(), 1);
                assert_eq!(record.quotes[0].quote_id, quote_id);
                assert!(record.responded_qps.contains(&qp));
            }
            other => panic!("Expected RpiQuoteStored, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_quote_rejects_price_outside_limit_without_book_reference() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        manager.insert_rfq(make_test_rpi_rfq_without_book_reference(
            rfq_id,
            taker,
            Side::Buy,
        ));

        let quote =
            make_test_quote_with_leg(qp, dec!(-500.0005), Side::Buy, dec!(100.0001), dec!(5));
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteRejected { record, reason } => {
                assert!(reason.contains("does not satisfy limit"));
                assert_eq!(record.quotes.len(), 0);
                assert!(record.responded_qps.contains(&qp));
            }
            other => panic!("Expected RpiQuoteRejected, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_auction_close_boundary_allows_deadline_millisecond() {
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let mut record = make_test_rpi_rfq(rfq_id, taker, Side::Buy);
        record.eligible_qps = vec![test_wallet(2)];
        record.deadline_at = 1_000;

        assert!(!should_close_rpi_auction(&record, 999));
        assert!(!should_close_rpi_auction(&record, 1_000));
        assert!(should_close_rpi_auction(&record, 1_001));
    }

    #[test]
    fn test_rpi_quote_rejects_self_trade_and_counts_response() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);

        manager.insert_rfq(make_test_rpi_rfq(rfq_id, taker, Side::Buy));

        let quote =
            make_test_quote_with_leg(taker, dec!(-499.9995), Side::Buy, dec!(99.9999), dec!(5));
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteRejected { record, reason } => {
                assert_eq!(reason, RFQ_SELF_TRADE_REJECTION_REASON);
                assert_eq!(record.status, RfqStatus::SentToQps);
                assert_eq!(record.quotes.len(), 0);
                assert!(record.responded_qps.contains(&taker));
            }
            other => panic!("Expected RpiQuoteRejected, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_quote_rejects_partial_size() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        manager.insert_rfq(make_test_rpi_rfq(rfq_id, taker, Side::Buy));

        let quote =
            make_test_quote_with_leg(qp, dec!(-499.9995), Side::Buy, dec!(99.9999), dec!(4));
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteRejected { record, reason } => {
                assert!(reason.contains("does not fully cover requested size"));
                assert_eq!(record.quotes.len(), 0);
                assert!(record.responded_qps.contains(&qp));
            }
            other => panic!("Expected RpiQuoteRejected, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_quote_stores_full_size_price_improved_candidate() {
        let manager = TestRfqManager::new();
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let qp = test_wallet(2);

        manager.insert_rfq(make_test_rpi_rfq(rfq_id, taker, Side::Buy));

        let quote =
            make_test_quote_with_leg(qp, dec!(-499.9995), Side::Buy, dec!(99.9999), dec!(5));
        let quote_id = quote.quote_id;
        let result = manager.handle_qp_response(rfq_id, quote).unwrap();

        match result {
            HandleQpResponseResult::RpiQuoteStored(record) => {
                assert_eq!(record.status, RfqStatus::QuotesReceived);
                assert_eq!(record.quotes.len(), 1);
                assert_eq!(record.quotes[0].quote_id, quote_id);
                assert!(record.responded_qps.contains(&qp));
            }
            other => panic!("Expected RpiQuoteStored, got {:?}", other),
        }
    }

    #[test]
    fn test_rpi_candidate_sort_buy_best_price_then_arrival() {
        let rfq_id = Uuid::now_v7();
        let taker = test_wallet(1);
        let mut record = make_test_rpi_rfq(rfq_id, taker, Side::Buy);
        let older_best = Uuid::now_v7();
        let newer_best = Uuid::now_v7();
        let worse = Uuid::now_v7();
        record.quotes = vec![
            QpQuote {
                quote_id: worse,
                qp_wallet: test_wallet(2),
                legs: vec![QuoteLeg {
                    instrument: "ETH-20260401-3000-C".to_string(),
                    side: Side::Buy,
                    price: dec!(99.90),
                    size: dec!(5),
                }],
                net_premium: dec!(-499.50),
                valid_for_ms: 25_000,
                qp_signature: "0xqp_sig".to_string(),
                qp_nonce: 1,
                received_at: 3,
            },
            QpQuote {
                quote_id: newer_best,
                qp_wallet: test_wallet(3),
                legs: vec![QuoteLeg {
                    instrument: "ETH-20260401-3000-C".to_string(),
                    side: Side::Buy,
                    price: dec!(99.80),
                    size: dec!(5),
                }],
                net_premium: dec!(-499),
                valid_for_ms: 25_000,
                qp_signature: "0xqp_sig".to_string(),
                qp_nonce: 1,
                received_at: 2,
            },
            QpQuote {
                quote_id: older_best,
                qp_wallet: test_wallet(4),
                legs: vec![QuoteLeg {
                    instrument: "ETH-20260401-3000-C".to_string(),
                    side: Side::Buy,
                    price: dec!(99.80),
                    size: dec!(5),
                }],
                net_premium: dec!(-499),
                valid_for_ms: 25_000,
                qp_signature: "0xqp_sig".to_string(),
                qp_nonce: 1,
                received_at: 1,
            },
        ];

        let sorted = sorted_rpi_candidate_quote_ids(&record).unwrap();

        assert_eq!(sorted, vec![older_best, newer_best, worse]);
    }
}