hypercall/rsm/unified_engine/

recovery.rs

//! Replay and recovery logic for UnifiedEngine.

use super::*;
use crate::engine_enums_ext::ReplayCommandExt;
use crate::rsm::apply::{CommandEnvelope, EngineCommand};
use hypercall_db::JournalReplayReader;
use std::collections::BTreeMap;

type ReplayFillEvent = (hypercall_types::Fill, hypercall_types::FillAccounting);
type ReplayFillEventsByCommand = BTreeMap<i64, Vec<ReplayFillEvent>>;

#[cfg(test)]
fn test_source_hash(byte: u8) -> alloy::primitives::FixedBytes<32> {
    alloy::primitives::FixedBytes::from([byte; 32])
}

fn replay_state_command_type(command_type: &str) -> Option<crate::nats::CommandType> {
    match command_type {
        "ExpireMarket" => Some(crate::nats::CommandType::MarketAction),
        "TickExpiry" => Some(crate::nats::CommandType::TickExpiry),
        "PriceUpdate" => Some(crate::nats::CommandType::PriceUpdate),
        "IvUpdate" => Some(crate::nats::CommandType::IvUpdate),
        "LiquidationState" => Some(crate::nats::CommandType::LiquidationState),
        "TierUpdate" => Some(crate::nats::CommandType::TierUpdate),
        "HypercorePositionUpdate" => Some(crate::nats::CommandType::HypercorePositionUpdate),
        "MmpConfigUpdate" => Some(crate::nats::CommandType::MmpConfigUpdate),
        "TradingModeUpdate" => Some(crate::nats::CommandType::TradingModeUpdate),
        "DepositUpdate" => Some(crate::nats::CommandType::DepositUpdate),
        "OptionDepositUpdate" => Some(crate::nats::CommandType::OptionDepositUpdate),
        "OptionWithdrawalUpdate" => Some(crate::nats::CommandType::OptionWithdrawalUpdate),
        "CashWithdrawalUpdate" => Some(crate::nats::CommandType::CashWithdrawalUpdate),
        "LiquidationBonusUpdate" => Some(crate::nats::CommandType::LiquidationBonusUpdate),
        "ApproveAgent" => Some(crate::nats::CommandType::ApproveAgent),
        "RevokeAgent" => Some(crate::nats::CommandType::RevokeAgent),
        "NonceAdvance" => Some(crate::nats::CommandType::NonceAdvance),
        "HypercoreEquityUpdate" => Some(crate::nats::CommandType::HypercoreEquityUpdate),
        "SetPmSettlementPoolConfig" => Some(crate::nats::CommandType::SetPmSettlementPoolConfig),
        "RecordPmVaultDeposit" => Some(crate::nats::CommandType::RecordPmVaultDeposit),
        "RequestPmVaultWithdrawal" => Some(crate::nats::CommandType::RequestPmVaultWithdrawal),
        "AccruePmSettlementInterest" => Some(crate::nats::CommandType::AccruePmSettlementInterest),
        "ApplyPmSettlementRepayment" => Some(crate::nats::CommandType::ApplyPmSettlementRepayment),
        "JournalPmRecoveryPlan" => Some(crate::nats::CommandType::JournalPmRecoveryPlan),
        "MarkPmRecoveryActionSubmitted" => {
            Some(crate::nats::CommandType::MarkPmRecoveryActionSubmitted)
        }
        "ResolvePmRecoveryAction" => Some(crate::nats::CommandType::ResolvePmRecoveryAction),
        _ => None,
    }
}

fn decode_state_command_for_replay(cmd: &hypercall_db::ReplayCommand) -> Option<EngineCommand> {
    let command_type = replay_state_command_type(&cmd.command_type)?;
    let command_wire_version = replay_command_wire_version(&cmd.command_data);
    Some(
        crate::nats::deserialize::deserialize_command_for_replay(
            command_type,
            command_wire_version,
            &cmd.command_data,
        )
        .unwrap_or_else(|e| {
            panic!(
                "CRITICAL_FAILURE: Failed to decode {} for replay \
                     (command_id={}, command_data_len={}): {}",
                cmd.command_type,
                cmd.command_id,
                cmd.command_data.len(),
                e
            )
        }),
    )
}

fn replay_command_wire_version(command_data: &[u8]) -> u8 {
    if replay_command_data_is_named_payload(command_data) {
        crate::nats::COMMAND_WIRE_VERSION_V1
    } else {
        crate::nats::LEGACY_COMMAND_WIRE_VERSION
    }
}

fn replay_command_data_is_named_payload(command_data: &[u8]) -> bool {
    command_data.first() == Some(&hypercall_types::WIRE_FORMAT_VERSION)
        && command_data
            .get(1)
            .is_some_and(|marker| matches!(*marker, 0x80..=0x8f | 0xde | 0xdf))
}

fn replay_envelope_timestamp(command: &EngineCommand) -> u64 {
    match command {
        EngineCommand::PriceUpdate { timestamp_ms, .. }
        | EngineCommand::IvUpdate { timestamp_ms, .. }
        | EngineCommand::HypercorePositionUpdate { timestamp_ms, .. }
        | EngineCommand::TradingModeUpdate { timestamp_ms, .. }
        | EngineCommand::DepositUpdate { timestamp_ms, .. }
        | EngineCommand::OptionDepositUpdate { timestamp_ms, .. }
        | EngineCommand::OptionWithdrawalUpdate { timestamp_ms, .. }
        | EngineCommand::CashWithdrawalUpdate { timestamp_ms, .. }
        | EngineCommand::LiquidationBonusUpdate { timestamp_ms, .. }
        | EngineCommand::ApproveAgent { timestamp_ms, .. }
        | EngineCommand::RevokeAgent { timestamp_ms, .. }
        | EngineCommand::NonceAdvance { timestamp_ms, .. }
        | EngineCommand::HypercoreEquityUpdate { timestamp_ms, .. } => *timestamp_ms,
        EngineCommand::SetPmSettlementPoolConfig(command) => command.timestamp_ms,
        EngineCommand::RecordPmVaultDeposit(command) => command.timestamp_ms,
        EngineCommand::RequestPmVaultWithdrawal(command) => command.timestamp_ms,
        EngineCommand::AccruePmSettlementInterest(command) => command.timestamp_ms,
        EngineCommand::ApplyPmSettlementRepayment(command) => command.timestamp_ms,
        EngineCommand::JournalPmRecoveryPlan(command) => command.timestamp_ms,
        EngineCommand::MarkPmRecoveryActionSubmitted(command) => command.timestamp_ms,
        EngineCommand::ResolvePmRecoveryAction(command) => command.timestamp_ms,
        EngineCommand::MarketAction(command) => command.message.timestamp,
        EngineCommand::TickExpiry { now_ms, .. } => *now_ms,
        EngineCommand::LiquidationState(message) => message.timestamp,
        _ => 0,
    }
}

pub(crate) struct StartupExpiredOrderCancel {
    pub order_id: u64,
    pub wallet: WalletAddress,
    pub symbol: String,
    pub price: Decimal,
    pub size: Decimal,
    pub side: hypercall_types::Side,
    pub tif: hypercall_types::TimeInForce,
    pub is_perp: bool,
    pub underlying: Option<String>,
    pub reduce_only: Option<bool>,
    pub nonce: Option<u64>,
    pub signature: Option<String>,
    pub mmp_enabled: bool,
    pub filled_size: Decimal,
    pub client_id: Option<String>,
    pub timestamp: u64,
}

impl UnifiedEngine {
    pub(super) fn load_markets_from_db(&mut self) {
        if let Some(ref handler) = self.ctx.db {
            info!("Loading instruments from database...");
            match handler.get_all_instruments_sync() {
                Ok(instruments) => {
                    info!("Loading {} instruments from database", instruments.len());
                    let mut loaded_count = 0;
                    let mut expired_on_startup = 0;
                    let mut scheduled_for_expiry = 0;
                    let mut startup_pending_symbols: Vec<String> = Vec::new();

                    // Startup expiry classification. A clock failure must not
                    // silently classify every expired instrument as live, so
                    // unix_now_secs panics instead of falling back.
                    let now = crate::shared::clock::unix_now_secs();

                    for instrument in instruments {
                        // Parse option type
                        let option_type = instrument.option_type;

                        // Rebuild expiry maps for restart-safety
                        // Convert YYYYMMDD expiry to Unix timestamp
                        let expiry_timestamp = crate::rsm::margin_manager::expiry_date_to_timestamp(
                            &instrument.underlying,
                            instrument.expiry as u64,
                        );

                        if expiry_timestamp <= now {
                            // Already expired at startup, always block trading in-memory.
                            self.ctx
                                .expired_instruments
                                .insert(instrument.id.clone(), true);
                            let trading_mode = hypercall_types::TradingModes::from_db_str(
                                &instrument.trading_mode,
                            );
                            self.ctx
                                .instrument_trading_modes
                                .insert(instrument.id.clone(), trading_mode);
                            expired_on_startup += 1;

                            match instrument.status {
                                hypercall_types::api_models::InstrumentStatus::Active => {
                                    startup_pending_symbols.push(instrument.id.clone());
                                    warn!(
                                        "Instrument {} expired while offline (expiry_ts={}) with status Active - \
                                         queuing startup DB transition to EXPIRED_PENDING_PRICE for settlement retry.",
                                        instrument.id, expiry_timestamp
                                    );
                                }
                                hypercall_types::api_models::InstrumentStatus::ExpiredPendingPrice => {
                                    info!(
                                        "Instrument {} expired while offline (expiry_ts={}) and already ExpiredPendingPrice - \
                                         settlement retry will resume from persisted state.",
                                        instrument.id, expiry_timestamp
                                    );
                                }
                                hypercall_types::api_models::InstrumentStatus::Settled => {
                                    debug!(
                                        "Instrument {} expired while offline (expiry_ts={}) and already Settled - \
                                         no startup status transition needed.",
                                        instrument.id, expiry_timestamp
                                    );
                                }
                            }

                            // Emit metric for monitoring
                            metrics::gauge!("ht_settlement_instruments_pending_price")
                                .increment(1.0);
                        } else {
                            // Not yet expired - create orderbook and schedule for expiry
                            let orderbook = OrderBook::with_symbol(
                                instrument.expiry as u64,
                                instrument.strike,
                                option_type,
                                instrument.id.clone(),
                            );

                            self.ctx.orderbooks.insert(instrument.id.clone(), orderbook);
                            let trading_mode = hypercall_types::TradingModes::from_db_str(
                                &instrument.trading_mode,
                            );
                            self.ctx
                                .instrument_trading_modes
                                .insert(instrument.id.clone(), trading_mode);
                            debug!("Loaded orderbook for symbol: {}", instrument.id);
                            loaded_count += 1;

                            // Schedule for expiry
                            self.expiry_manager
                                .expiry_schedules
                                .entry(expiry_timestamp)
                                .or_default()
                                .push(instrument.id.clone());
                            scheduled_for_expiry += 1;
                        }
                    }

                    if !startup_pending_symbols.is_empty() {
                        if let Err(e) = handler
                            .transition_active_instruments_to_expired_pending_sync(
                                &startup_pending_symbols,
                            )
                        {
                            panic!(
                                "CRITICAL_FAILURE: Failed startup transition of expired ACTIVE instruments to EXPIRED_PENDING_PRICE for {:?}: {}. \
                                 Persisted state must match in-memory expiry state before continuing.",
                                startup_pending_symbols, e
                            );
                        }
                        info!(
                            "Startup transitioned {} expired ACTIVE instruments to EXPIRED_PENDING_PRICE",
                            startup_pending_symbols.len()
                        );
                    }

                    // Observability: log rebuild summary
                    info!(
                        "Expiry rebuild complete: {} instruments loaded, {} expired at startup, {} transitioned to EXPIRED_PENDING_PRICE, {} scheduled for expiry",
                        loaded_count,
                        expired_on_startup,
                        startup_pending_symbols.len(),
                        scheduled_for_expiry
                    );

                    // Emit metrics for observability
                    metrics::gauge!("ht_expiry_rebuild_loaded").set(loaded_count as f64);
                    metrics::gauge!("ht_expiry_rebuild_pending_price")
                        .set(expired_on_startup as f64);
                    metrics::gauge!("ht_expiry_rebuild_scheduled").set(scheduled_for_expiry as f64);

                    info!(
                        "Successfully loaded {} orderbooks from database",
                        loaded_count
                    );

                    if loaded_count != self.ctx.orderbooks.len() {
                        warn!(
                            "Mismatch: loaded {} instruments but have {} orderbooks",
                            loaded_count,
                            self.ctx.orderbooks.len()
                        );
                    }

                    info!(
                        "UnifiedEngine sync status remains CatchingUp (journal replay and post-startup reconciliation pending)"
                    );
                }
                Err(e) => {
                    error!("FATAL: Failed to load instruments from database: {}", e);
                    panic!(
                        "Cannot start UnifiedEngine without loading instruments: {}",
                        e
                    );
                }
            }
        } else {
            // Allow running without diesel_handler if no database is configured
            info!("No diesel_handler configured, skipping market loading from database");
        }
    }

    /// Load WAL checkpoint metadata and prepare replay bounds.
    pub(super) fn load_orderbook_snapshots(&mut self) {
        if self.ctx.db.is_none() {
            self.replay_checkpoint = hypercall_journal::checkpoint::WalCheckpointMetadata::ZERO;
            self.ctx
                .l2_update_seq
                .store(self.replay_checkpoint.last_l2_seq, Ordering::SeqCst);
            return;
        }

        let checkpoint_path =
            hypercall_journal::checkpoint::checkpoint_path_from_config(self.wal_path.as_ref());
        let checkpoint = hypercall_journal::checkpoint::read_checkpoint(&checkpoint_path)
            .unwrap_or_else(|e| {
                panic!(
                    "CRITICAL_FAILURE: Failed to load WAL checkpoint metadata from {}: {}",
                    checkpoint_path.display(),
                    e
                )
            });

        self.replay_checkpoint = checkpoint;
        self.ctx
            .l2_update_seq
            .store(checkpoint.last_l2_seq, Ordering::SeqCst);
        info!(
            "Loaded WAL checkpoint for recovery: wal_offset={}, last_command_id={}, last_l2_seq={}",
            checkpoint.wal_offset, checkpoint.last_command_id, checkpoint.last_l2_seq
        );
    }

    /// Check restored orderbooks for crosses and log warnings.
    /// Crossed orderbooks after restore indicate a bug in the matching engine
    /// or snapshot corruption - the matching engine should never allow bid >= ask.
    pub(super) fn check_restored_orderbooks_for_crosses(&self) {
        let mut crossed_count = 0;
        let mut crossed_details = Vec::new();

        for (symbol, orderbook) in &self.ctx.orderbooks {
            if orderbook.is_crossed() {
                crossed_count += 1;
                let best_bid = orderbook.get_best_bid();
                let best_ask = orderbook.get_best_ask();
                crossed_details.push(format!("{}: bid={:?} ask={:?}", symbol, best_bid, best_ask));
            }
        }

        if crossed_count > 0 {
            error!(
                "🚨 CRITICAL: {} orderbooks have CROSSED state after restore! \
                This indicates a matching engine bug - orders should have executed. \
                Details: {:?}",
                crossed_count, crossed_details
            );
            panic!(
                "CRITICAL_FAILURE: {} crossed orderbooks after restore: {:?}",
                crossed_count, crossed_details
            );
        }
    }

    /// Replay commands from the journal to reconstruct orderbook state.
    ///
    /// Recovery runs in two bounded windows:
    /// 1) Base reconstruction: commands (0, checkpoint.last_command_id]
    /// 2) Delta replay: commands (checkpoint.last_command_id, +inf)
    pub(super) fn replay_commands_from_journal(&mut self) {
        let snapshot_seq = self.ctx.l2_update_seq.load(Ordering::SeqCst);
        let checkpoint_command_id = self.replay_checkpoint.last_command_id;
        const REPLAY_CHUNK_SIZE: i64 = 5_000;

        if self.ctx.db.is_none() {
            debug!("No diesel handler, skipping journal replay");
            return;
        }

        let mut replay_count = 0;
        let mut max_replayed_order_id: u64 = 0;
        if checkpoint_command_id > 0 && !self.snapshot_loaded {
            info!(
                "Recovery base reconstruction: replaying journal commands up to checkpoint command_id {}",
                checkpoint_command_id
            );
            self.replay_command_window(
                0,
                Some(checkpoint_command_id),
                REPLAY_CHUNK_SIZE,
                &mut replay_count,
                &mut max_replayed_order_id,
            );
        } else if self.snapshot_loaded {
            info!(
                "Skipping base reconstruction — engine state snapshot loaded (last_command_id={})",
                checkpoint_command_id
            );
        }

        info!(
            "Recovery delta replay: replaying commands after checkpoint command_id {}",
            checkpoint_command_id
        );
        self.replay_command_window(
            checkpoint_command_id,
            None,
            REPLAY_CHUNK_SIZE,
            &mut replay_count,
            &mut max_replayed_order_id,
        );

        // Pass 4 (ghost order reconciliation) is deferred to
        // run_post_startup_reconciliation(), which runs ~5s after the engine
        // loop starts. This gives the journal batcher time to materialize
        // pending terminal statuses into order_infos.
        // Running the expensive get_terminal_order_ids_sync query here
        // (before flush completes) would produce stale results
        // AND double the startup time by querying the same large table twice.

        // After replay, ensure next_order_id is higher than any order_id seen
        // in the replayed commands. Without this, the engine could reassign order IDs
        // that were used by replayed commands (especially rejected/filled/cancelled
        // orders that don't appear in the orderbook snapshot), causing the MM to
        // detect "phantom regressions" where a terminal order_id reappears as open.
        if max_replayed_order_id >= self.ctx.next_order_id {
            let old = self.ctx.next_order_id;
            self.ctx.next_order_id = max_replayed_order_id + 1;
            info!(
                "Advanced next_order_id after replay: {} -> {} (max replayed order_id was {})",
                old, self.ctx.next_order_id, max_replayed_order_id
            );
        }

        // After replay, update the engine's L2 sequence to match the database.
        // The replayed commands already emitted L2 updates with sequences, so
        // we query the max sequence from the events table.
        // Re-borrow handler for the final query.
        let max_seq_result = {
            let Some(ref handler) = self.ctx.db else {
                warn!(
                    "No diesel handler for max L2 seq query. L2 seq remains at {}",
                    snapshot_seq
                );
                return;
            };
            let replay: &dyn JournalReplayReader = handler;
            replay.get_max_l2_seq_from_events_sync()
        };

        match max_seq_result {
            Ok(max_seq) => {
                if max_seq > snapshot_seq {
                    self.ctx.l2_update_seq.store(max_seq, Ordering::SeqCst);
                    info!(
                        "Journal replay complete: {} commands replayed, L2 seq {} -> {}",
                        replay_count, snapshot_seq, max_seq
                    );
                } else {
                    info!(
                        "Journal replay complete: {} commands replayed, L2 seq unchanged at {}",
                        replay_count, snapshot_seq
                    );
                }
            }
            Err(e) => {
                panic!(
                    "CRITICAL_FAILURE: Failed to get max L2 seq after replay: {}. \
                     Cannot safely continue with stale sequence {}.",
                    e, snapshot_seq
                );
            }
        }

        // Sync L2 snapshot baselines for all orderbooks after replay.
        //
        // During replay, orders are added/removed from orderbooks but
        // emit_orderbook_events() is never called, so last_bid_snapshot
        // and last_ask_snapshot remain empty. Without this sync, the
        // first cancel after restart produces an empty L2 diff
        // (empty→empty), and the CancelOrder command gets no L2Update
        // event in engine_events — making it invisible to the replay
        // query on the NEXT restart (JOIN on event_type = 'L2Update').
        for ob in self.ctx.orderbooks.values_mut() {
            ob.sync_l2_snapshot_baseline();
        }

        self.flush_startup_expired_order_cancels();
    }

    fn replay_command_window(
        &mut self,
        start_command_id: i64,
        end_command_id: Option<i64>,
        chunk_size: i64,
        replay_count: &mut usize,
        max_replayed_order_id: &mut u64,
    ) {
        let mut cursor = start_command_id;
        loop {
            let commands = {
                let replay: &dyn JournalReplayReader = self
                    .ctx
                    .db
                    .as_ref()
                    .expect("diesel handler required for replay");
                replay
                    .get_replay_commands_after_command_id_sync(cursor, end_command_id, chunk_size)
                    .unwrap_or_else(|e| {
                        panic!(
                            "CRITICAL_FAILURE: Failed to query replay commands after command_id {} \
                             (end={:?}, limit={}): {}",
                            cursor, end_command_id, chunk_size, e
                        )
                    })
            };

            if commands.is_empty() {
                break;
            }

            let chunk_end_command_id = commands
                .last()
                .expect("replay command chunk unexpectedly empty")
                .command_id;
            if chunk_end_command_id <= cursor {
                panic!(
                    "CRITICAL_FAILURE: Replay cursor did not advance (cursor={}, chunk_end={})",
                    cursor, chunk_end_command_id
                );
            }

            let mut fill_events_by_command =
                self.replay_fill_events_by_command_for_window(cursor, chunk_end_command_id);
            for cmd in commands {
                self.replay_command_row(&cmd, replay_count, max_replayed_order_id);
                if let Some(fill_events) = fill_events_by_command.remove(&cmd.command_id) {
                    self.apply_fill_events_for_replay(&fill_events);
                }
            }
            if !fill_events_by_command.is_empty() {
                let command_ids = fill_events_by_command.keys().copied().collect::<Vec<_>>();
                panic!(
                    "CRITICAL_FAILURE: Fill replay found OrderFilled events for command_ids {:?} \
                     without matching replay commands in window ({}..={}). Persisted journal is inconsistent.",
                    command_ids, cursor, chunk_end_command_id
                );
            }

            self.replay_order_update_events_for_window(cursor, chunk_end_command_id);
            cursor = chunk_end_command_id;
        }
    }

    fn replay_command_row(
        &mut self,
        cmd: &hypercall_db::ReplayCommand,
        replay_count: &mut usize,
        max_replayed_order_id: &mut u64,
    ) {
        // RFQ command rows: replay is event-driven, not command-driven. The
        // journaled OrderFilled events are picked up by the fill-event replay
        // pass. Replay must not re-run matching or validation for an accepted
        // RFQ, but it must restore any nonces burned by the original command.
        if cmd.command_type == "RfqExecute" {
            if let Ok(rfq_cmd) = rmp_serde::from_slice::<hypercall_runtime_api::RfqExecuteCommand>(
                &cmd.command_data[1..],
            ) {
                if let Some(nonce) = rfq_cmd.taker_nonce {
                    let nonce_signer = rfq_cmd.taker_submit_signer();
                    self.ctx
                        .nonce_sets
                        .entry(nonce_signer)
                        .or_insert_with(|| {
                            hypercall_engine::BoundedNonceSet::new(
                                hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                            )
                        })
                        .insert(nonce);
                }
                if let Some(nonce) = rfq_cmd.taker_accept_nonce {
                    let nonce_signer = rfq_cmd.taker_accept_signer();
                    self.ctx
                        .nonce_sets
                        .entry(nonce_signer)
                        .or_insert_with(|| {
                            hypercall_engine::BoundedNonceSet::new(
                                hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                            )
                        })
                        .insert(nonce);
                }
            }
            *replay_count += 1;
            return;
        }

        if let Some(command) = decode_state_command_for_replay(cmd) {
            if Self::is_expiry_replay_command(&command) {
                self.replay_expiry_command_row(cmd, command);
                *replay_count += 1;
                return;
            }
            let disposition =
                crate::rsm::restart_components::replay_disposition_for_command(&command);
            assert_eq!(
                disposition,
                hypercall_recovery::ReplayDisposition::Applied,
                "CRITICAL_FAILURE: decoded restart-owned command {} has no component replay owner",
                cmd.command_type
            );
            if self.apply_legacy_agent_auth_replay_command(&command) {
                *replay_count += 1;
                return;
            }
            let timestamp_ms = replay_envelope_timestamp(&command);
            let output = self
                .apply(CommandEnvelope::new(timestamp_ms, command))
                .unwrap_or_else(|e| {
                    panic!(
                        "CRITICAL_FAILURE: Replay failed for {} command {} \
                         (request_id={}): {}",
                        cmd.command_type, cmd.command_id, cmd.request_id, e
                    )
                });
            self.apply_replayed_expiry_effects(&output.expiry_effects)
                .unwrap_or_else(|e| {
                    panic!(
                        "CRITICAL_FAILURE: Replay failed to apply expiry runtime effects for {} \
                         command {} (request_id={}): {}",
                        cmd.command_type, cmd.command_id, cmd.request_id, e
                    )
                });
            self.apply_pm_settlement_projection_effects_sync(
                &output.pm_settlement_effects,
                &cmd.request_id,
            )
            .unwrap_or_else(|e| {
                panic!(
                    "CRITICAL_FAILURE: Replay failed to apply PM settlement projection effects for {} \
                     command {} (request_id={}): {}",
                    cmd.command_type, cmd.command_id, cmd.request_id, e
                )
            });
            *replay_count += 1;
            return;
        }

        let decoded_response = cmd.decode_response();
        if let Some(resp) = decoded_response.as_ref() {
            if let Some(order_id) = resp.order_id {
                *max_replayed_order_id = (*max_replayed_order_id).max(order_id);
            }
        }

        let order_action = cmd.decode_command();

        match self.process_order_for_replay(&order_action, decoded_response.as_ref()) {
            Ok(_) => {
                *replay_count += 1;
                debug!(
                    "Replayed command {} (request_id: {})",
                    cmd.command_id, cmd.request_id
                );
            }
            Err(e) => {
                panic!(
                    "CRITICAL_FAILURE: Replay failed for command {} (request_id: {}): {}",
                    cmd.command_id, cmd.request_id, e
                );
            }
        }
    }

    fn is_expiry_replay_command(command: &EngineCommand) -> bool {
        matches!(
            command,
            EngineCommand::MarketAction(market_command)
                if matches!(
                    market_command.message.action,
                    hypercall_types::MarketAction::ExpireMarket
                )
        ) || matches!(command, EngineCommand::TickExpiry { .. })
    }

    fn replay_expiry_command_row(
        &mut self,
        cmd: &hypercall_db::ReplayCommand,
        command: EngineCommand,
    ) {
        let timestamp_ms = replay_envelope_timestamp(&command);
        let output = self
            .apply(CommandEnvelope::new(timestamp_ms, command))
            .unwrap_or_else(|error| {
                panic!(
                    "CRITICAL_FAILURE: Replay failed for expiry command {} \
                     (request_id={}): {}",
                    cmd.command_id, cmd.request_id, error
                )
            });

        self.apply_replayed_expiry_effects(&output.expiry_effects)
            .unwrap_or_else(|error| {
                panic!(
                    "CRITICAL_FAILURE: Failed to apply replayed expiry effects while \
                     replaying command {} (request_id={}): {}",
                    cmd.command_id, cmd.request_id, error
                )
            });

        self.apply_pm_settlement_projection_effects_sync(
            &output.pm_settlement_effects,
            &cmd.request_id,
        )
        .unwrap_or_else(|error| {
            panic!(
                "CRITICAL_FAILURE: Failed to apply replayed PM settlement projection effects while \
                 replaying command {} (request_id={}): {}",
                cmd.command_id, cmd.request_id, error
            )
        });

        if !output.events.is_empty() {
            self.startup_replayed_events
                .push((cmd.request_id.clone(), output.events));
        }
    }

    pub(super) fn apply_legacy_agent_auth_replay_command(
        &mut self,
        command: &EngineCommand,
    ) -> bool {
        // Temporary replay shim for WAL rows written before agent-auth commands
        // required signed nonces. Remove this after the pre-nonce WAL retention
        // window has expired and all restart replay ranges are guaranteed to
        // contain only nonce-bearing ApproveAgent/RevokeAgent commands.
        match command {
            EngineCommand::ApproveAgent {
                wallet,
                agent,
                expires_at_ms,
                nonce: None,
                ..
            } => {
                self.ctx
                    .agent_authorizations
                    .entry(*wallet)
                    .or_default()
                    .insert(*agent, *expires_at_ms);
                debug!(
                    wallet = %wallet,
                    agent = %agent,
                    expires_at_ms = ?expires_at_ms,
                    "Replayed legacy ApproveAgent command without nonce"
                );
                true
            }
            EngineCommand::RevokeAgent {
                wallet,
                agent,
                nonce: None,
                ..
            } => {
                if let Some(agents) = self.ctx.agent_authorizations.get_mut(wallet) {
                    agents.remove(agent);
                    if agents.is_empty() {
                        self.ctx.agent_authorizations.remove(wallet);
                    }
                }
                debug!(
                    wallet = %wallet,
                    agent = %agent,
                    "Replayed legacy RevokeAgent command without nonce"
                );
                true
            }
            _ => false,
        }
    }

    fn replay_fill_events_by_command_for_window(
        &mut self,
        start_command_id: i64,
        end_command_id: i64,
    ) -> ReplayFillEventsByCommand {
        let rows = {
            let replay: &dyn JournalReplayReader = self
                .ctx
                .db
                .as_ref()
                .expect("diesel handler required for fill replay");
            replay
                .get_portfolio_events_for_command_range_sync(start_command_id + 1, end_command_id)
                .unwrap_or_else(|e| {
                    panic!(
                        "CRITICAL_FAILURE: Failed to query OrderFilled events for replay window \
                         ({}..={}): {}",
                        start_command_id, end_command_id, e
                    )
                })
        };
        if rows.is_empty() {
            return BTreeMap::new();
        }

        let mut fill_events_by_command: ReplayFillEventsByCommand = BTreeMap::new();
        for (idx, row) in rows.into_iter().enumerate() {
            if row.event_type != hypercall_db::EventType::OrderFilled {
                continue;
            }
            fill_events_by_command
                .entry(row.command_id)
                .or_default()
                .push(Self::decode_fill_event(&row.event_data, idx));
        }
        let max_fill_trade_id = fill_events_by_command
            .values()
            .flat_map(|fill_events| fill_events.iter().map(|(fill, _)| fill.trade_id))
            .max()
            .unwrap_or(0);
        if max_fill_trade_id >= self.ctx.next_trade_id {
            let old = self.ctx.next_trade_id;
            self.ctx.next_trade_id = max_fill_trade_id + 1;
            info!(
                "Advanced next_trade_id after replay: {} -> {} (max replayed trade_id was {})",
                old, self.ctx.next_trade_id, max_fill_trade_id
            );
        }
        fill_events_by_command
    }

    pub(super) fn apply_fill_events_for_replay(
        &mut self,
        fill_events: &[(hypercall_types::Fill, hypercall_types::FillAccounting)],
    ) {
        if fill_events.is_empty() {
            return;
        }

        let fills: Vec<_> = fill_events.iter().map(|(fill, _)| fill.clone()).collect();
        self.apply_fills_for_replay(&fills);

        // Update engine_positions from replayed fills so margin checks
        // and expiry settlement see the correct position state after restart.
        for (fill, accounting) in fill_events {
            for (wallet, delta) in [
                (fill.taker_wallet_address, accounting.taker_net_cash_delta),
                (fill.maker_wallet_address, accounting.maker_net_cash_delta),
            ] {
                if delta == rust_decimal::Decimal::ZERO {
                    continue;
                }
                let balance_after = self.ctx.balance_ledger.balance(&wallet) + delta;
                let update = hypercall_types::BalanceUpdate {
                    balance_update_seq: self.ctx.balance_ledger.next_balance_update_seq(),
                    wallet,
                    delta,
                    balance_after,
                    reason: hypercall_types::BalanceUpdateReason::OptionFillPremium,
                    reference_id: Some(fill.trade_id.to_string()),
                    source_command_id: None,
                    timestamp_ms: fill.timestamp,
                };
                self.ctx
                    .balance_ledger
                    .apply_balance_update(&update)
                    .unwrap_or_else(|error| {
                        panic!(
                            "CRITICAL: failed to apply replayed fill balance update: {}",
                            error
                        )
                    });
            }

            use crate::rsm::engine_deps::apply_fill_to_positions;
            use hypercall_types::to_human_readable_decimal;
            let human_size = to_human_readable_decimal(&fill.symbol, fill.size);
            let is_buy = matches!(fill.taker_side, hypercall_types::Side::Buy);
            let taker_signed = if is_buy { human_size } else { -human_size };
            apply_fill_to_positions(
                &mut self.ctx.engine_positions,
                fill.taker_wallet_address,
                fill.symbol.clone(),
                taker_signed,
                fill.price,
            );
            apply_fill_to_positions(
                &mut self.ctx.engine_positions,
                fill.maker_wallet_address,
                fill.symbol.clone(),
                -taker_signed,
                fill.price,
            );
        }
    }

    fn replay_order_update_events_for_window(
        &mut self,
        start_command_id: i64,
        end_command_id: i64,
    ) {
        let rows = {
            let replay: &dyn JournalReplayReader = self
                .ctx
                .db
                .as_ref()
                .expect("diesel handler required for order update replay");
            replay
                .get_order_update_events_for_command_range_sync(start_command_id, end_command_id)
                .unwrap_or_else(|e| {
                    panic!(
                        "CRITICAL_FAILURE: Failed to query OrderUpdate events for replay window \
                         ({}..={}): {}",
                        start_command_id, end_command_id, e
                    )
                })
        };
        if rows.is_empty() {
            return;
        }

        let updates = Self::decode_order_update_events(&rows);
        self.apply_cancel_events_for_replay(&updates);
    }

    fn decode_fill_events(
        raw_rows: &[Vec<u8>],
    ) -> Vec<(hypercall_types::Fill, hypercall_types::FillAccounting)> {
        raw_rows
            .iter()
            .enumerate()
            .map(|(idx, raw)| Self::decode_fill_event(raw, idx))
            .collect()
    }

    fn decode_fill_event(
        raw: &[u8],
        idx: usize,
    ) -> (hypercall_types::Fill, hypercall_types::FillAccounting) {
        match hypercall_types::EngineMessage::deserialize_from_wire(
            crate::shared::topics::TOPIC_FILLS,
            raw,
        ) {
            Ok(hypercall_types::EngineMessage::OrderFilled { fill, accounting }) => {
                (fill, accounting)
            }
            Ok(other) => {
                panic!(
                    "CRITICAL_FAILURE: Expected OrderFilled event at index {} during fill replay, got {}. \
                     Persisted data is corrupt; refusing to continue.",
                    idx,
                    other.type_name()
                );
            }
            Err(e) => {
                panic!(
                    "CRITICAL_FAILURE: Failed to deserialize OrderFilled event at index {}: {}. \
                     Persisted data is corrupt; refusing to continue.",
                    idx, e
                );
            }
        }
    }

    fn decode_order_update_events(
        raw_rows: &[Vec<u8>],
    ) -> Vec<hypercall_types::OrderUpdateMessage> {
        raw_rows
            .iter()
            .enumerate()
            .map(|(idx, raw)| {
                if raw.len() < 2 {
                    panic!(
                        "CRITICAL_FAILURE: OrderUpdate event data too short at index {} ({} bytes). \
                         Persisted data is corrupt.",
                        idx,
                        raw.len()
                    );
                }
                let version = raw[0];
                rmp_serde::from_slice::<hypercall_types::OrderUpdateMessage>(&raw[1..])
                    .unwrap_or_else(|e| {
                        panic!(
                            "CRITICAL_FAILURE: Failed to deserialize OrderUpdate event at index {} \
                             (version={}): {}. Persisted data is corrupt; refusing to continue.",
                            idx, version, e
                        )
                    })
            })
            .collect()
    }

    /// Post-startup ghost order reconciliation.
    ///
    /// Runs ~5s after engine start, giving the journal batcher time to flush
    /// pending terminal statuses to order_infos. Queries for any
    /// orders in the orderbook that have a terminal status and removes them.
    /// This is the sole safeguard against ghost orders from the outbox pattern
    /// crash scenario (SIGKILL between journal fsync and outbox publish).
    pub(super) fn run_post_startup_reconciliation(&mut self) {
        let all_order_ids: Vec<i64> = self
            .ctx
            .orderbooks
            .values()
            .flat_map(|ob| ob.get_all_order_ids())
            .map(|id| id as i64)
            .collect();

        if all_order_ids.is_empty() {
            info!("Post-startup reconciliation: no orders in orderbook, skipping");
            return;
        }

        if let Some(ref handler) = self.ctx.db {
            match handler.get_terminal_order_ids_sync(&all_order_ids) {
                Ok(terminal_ids) if !terminal_ids.is_empty() => {
                    let mut removed = 0;
                    for order_id in &terminal_ids {
                        let oid = *order_id as u64;
                        for ob in self.ctx.orderbooks.values_mut() {
                            if ob.has_order(oid) {
                                ob.cancel_order(oid);
                                removed += 1;
                                break;
                            }
                        }
                        self.ctx.order_index.remove_order_by_id(oid);
                    }
                    if removed > 0 {
                        warn!(
                            "Post-startup reconciliation: removed {} ghost orders that were \
                             terminal in order_infos but present in orderbook after replay.",
                            removed
                        );
                        // Publish updated snapshot so API reflects the corrected state
                        self.publish_snapshot();
                    }
                }
                Ok(_) => {
                    info!(
                        "Post-startup reconciliation: checked {} orders, all non-terminal — no ghost orders",
                        all_order_ids.len()
                    );
                }
                Err(e) => {
                    panic!(
                        "CRITICAL_FAILURE: Post-startup reconciliation query failed: {}. \
                         Ghost orders may remain in orderbook; refusing to serve live traffic.",
                        e
                    );
                }
            }
        }
    }

    /// Process an order for replay purposes (no journaling, no response).
    ///
    /// This is used during startup to replay commands from the journal after
    /// loading a snapshot. The commands are already persisted, so we just need
    /// to apply their state changes.
    ///
    /// `response_json` is the original response from when this command was first
    /// processed. It tells us the final order state (filled, open, partially_filled,
    /// cancelled) so we can reconstruct the correct resting quantity without
    /// re-running matching.
    pub(super) fn process_order_for_replay(
        &mut self,
        msg: &OrderActionMessage,
        response: Option<&OrderUpdateMessage>,
    ) -> Result<(), String> {
        if let Some(nonce) = msg.info.nonce {
            let signer = msg.api_wallet_address.unwrap_or(msg.wallet);
            self.ctx
                .nonce_sets
                .entry(signer)
                .or_insert_with(|| {
                    hypercall_engine::BoundedNonceSet::new(
                        hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                    )
                })
                .insert(nonce);
        }
        match msg.action {
            OrderAction::CreateOrder => self.replay_create_order(msg, response),
            OrderAction::CancelOrder => self.replay_cancel_order(msg),
            OrderAction::ReplaceOrder => self.replay_replace_order(msg, response),
        }
    }

    /// Replay a create order command using the journal's response to determine
    /// what resting state to restore.
    ///
    /// Instead of re-running matching (which would require determinism of all
    /// inputs: timestamps, fees, oracle state, etc.), we read the response_json
    /// from the journal to determine the order's final status and filled_size.
    /// This tells us exactly what resting quantity belongs in the orderbook:
    ///
    /// - FILLED → don't add (fully consumed)
    /// - CANCELED / REJECTED → don't add
    /// - OPEN → add with full original quantity
    /// - PARTIALLY_FILLED → add with (original_size - filled_size)
    fn replay_create_order(
        &mut self,
        msg: &OrderActionMessage,
        response: Option<&OrderUpdateMessage>,
    ) -> Result<(), String> {
        let symbol = &msg.info.symbol;

        // Perp orders are validated and sent to HyperCore; they never rest in
        // the internal orderbook. Skip replay entirely.
        if symbol.ends_with("-PERP") || msg.info.is_perp {
            return Ok(());
        }

        // No orderbook for this symbol — the instrument may have expired while
        // offline, been removed, or the snapshot predates the instrument.
        // If the journal response shows a resting order (Open/PartiallyFilled),
        // we must cancel it in the DB; otherwise the persisted order state
        // retains a stale OPEN row with no engine-side counterpart.
        //
        // We intentionally do NOT gate on expired_instruments here: the snapshot
        // restore can overwrite that map with older state, so an instrument that
        // expired during downtime may not appear in it. The absence of an
        // orderbook is sufficient — any resting order without an orderbook is
        // orphaned.
        if !self.ctx.orderbooks.contains_key(symbol) {
            let resp = match response {
                Some(r) => r,
                None => {
                    return Err(format!(
                        "Replay create for missing-orderbook instrument {} has no response; \
                         cannot determine order state for reconciliation",
                        symbol
                    ));
                }
            };

            match resp.status {
                OrderUpdateStatus::Open | OrderUpdateStatus::PartiallyFilled => {
                    let order_id = match (msg.info.order_id, resp.order_id) {
                        (Some(cmd_id), Some(resp_id)) if cmd_id != resp_id => {
                            return Err(format!(
                                "Replay create for missing-orderbook instrument {} has mismatched \
                                 order_id (command={}, response={})",
                                symbol, cmd_id, resp_id
                            ));
                        }
                        (Some(id), _) | (_, Some(id)) => id,
                        (None, None) => {
                            return Err(format!(
                                "Replay create for missing-orderbook instrument {} has no order_id \
                                 in command or response",
                                symbol
                            ));
                        }
                    };
                    let filled_size = match resp.status {
                        OrderUpdateStatus::PartiallyFilled => {
                            if resp.filled_size < dec!(0) || resp.filled_size >= msg.info.size {
                                return Err(format!(
                                    "Replay create for missing-orderbook instrument {} has invalid \
                                     partially-filled payload: size={}, filled={}",
                                    symbol, msg.info.size, resp.filled_size
                                ));
                            }
                            resp.filled_size
                        }
                        _ => dec!(0),
                    };
                    warn!(
                        "Replay: order {} on missing-orderbook instrument {} was {:?} — \
                         collecting for startup DB cancellation",
                        order_id, symbol, resp.status
                    );
                    self.startup_expired_order_cancels
                        .push(StartupExpiredOrderCancel {
                            order_id,
                            wallet: msg.wallet,
                            symbol: symbol.clone(),
                            price: msg.info.price,
                            size: msg.info.size,
                            side: msg.info.side,
                            tif: msg.info.tif,
                            is_perp: msg.info.is_perp,
                            underlying: msg.info.underlying.clone(),
                            reduce_only: msg.info.reduce_only,
                            nonce: msg.info.nonce,
                            signature: msg.info.signature.clone(),
                            mmp_enabled: msg.info.mmp_enabled,
                            filled_size,
                            client_id: msg.info.client_id.clone(),
                            timestamp: msg.timestamp,
                        });
                }
                OrderUpdateStatus::Acked => {
                    return Err(format!(
                        "Replay create for missing-orderbook instrument {} has non-final Acked \
                         status; refusing to skip reconciliation",
                        symbol
                    ));
                }
                // Filled, Canceled, Rejected — terminal, safe to skip
                _ => {}
            }

            return Ok(());
        }

        let response = match response {
            Some(resp) => resp,
            None => {
                return Err(format!(
                    "Replay create for symbol {} has no response; refusing to guess order state",
                    symbol
                ));
            }
        };

        // Determine resting quantity from the journal response.
        // Deserialize into OrderUpdateMessage to use proper enum types
        // instead of fragile string matching.
        let resting_quantity = match response.status {
            OrderUpdateStatus::Filled => {
                // Taker was fully filled — don't add to book.
                // Maker fills are applied by Pass 2 (exact fill events
                // from engine_events, same DB transaction as commands).
                debug!(
                    "Replay create for {} was fully filled, not adding to orderbook",
                    symbol
                );
                return Ok(());
            }
            OrderUpdateStatus::Canceled | OrderUpdateStatus::Rejected => {
                debug!(
                    "Replay create for {} was {:?}, not adding to orderbook",
                    symbol, response.status
                );
                return Ok(());
            }
            OrderUpdateStatus::Open => {
                // No fills occurred — full quantity rests
                msg.info.size
            }
            OrderUpdateStatus::PartiallyFilled => {
                if response.filled_size < dec!(0) || response.filled_size >= msg.info.size {
                    return Err(format!(
                        "Replay create for symbol {} has invalid partially-filled payload: size={}, filled={}",
                        symbol, msg.info.size, response.filled_size
                    ));
                }

                // Taker was partially filled — add remaining to book.
                // Maker fills are applied by Pass 2 (exact fill events).
                let remaining = msg.info.size - response.filled_size;
                debug!(
                    "Replay create for {} partially filled: original={}, filled={}, resting={}",
                    symbol, msg.info.size, response.filled_size, remaining
                );
                remaining
            }
            OrderUpdateStatus::Acked => {
                return Err(format!(
                    "Replay create for symbol {} has non-final Acked status; refusing to infer resting quantity",
                    symbol
                ));
            }
        };

        let order_id = match (msg.info.order_id, response.order_id) {
            (Some(command_order_id), Some(response_order_id)) => {
                if command_order_id != response_order_id {
                    return Err(format!(
                        "Replay create for symbol {} has mismatched order_id (command={}, response={})",
                        symbol, command_order_id, response_order_id
                    ));
                }
                command_order_id
            }
            (Some(order_id), None) | (None, Some(order_id)) => order_id,
            (None, None) => {
                return Err(format!(
                    "Replay create for symbol {} with status {:?} has no order_id in command or response",
                    symbol, response.status
                ));
            }
        };

        // Validate order isn't already present.
        // Use a scoped immutable borrow so we can call &mut self methods below.
        {
            let orderbook = self.ctx.orderbooks.get(symbol).unwrap();

            if orderbook.has_order(order_id) {
                debug!(
                    "Order {} already exists in orderbook during replay, skipping",
                    order_id
                );
                return Ok(());
            }
        }

        let timestamp = msg.timestamp;
        let price = msg.info.price;
        let side = msg.info.side;
        let wallet = msg.wallet;

        // Re-borrow orderbook mutably now that fill inference is done
        let orderbook = self.ctx.orderbooks.get_mut(symbol).unwrap();
        orderbook.add_order_with_metadata(
            order_id,
            price,
            resting_quantity,
            side,
            wallet,
            timestamp,
            msg.info.client_id.clone(),
            msg.info.mmp_enabled,
            msg.info.size, // original_size from the journal entry
        );

        // Update order index: remove any snapshot-loaded stub, then add with replay metadata
        self.ctx.order_index.remove_order(&wallet, order_id);
        self.ctx.order_index.add_order(
            &wallet,
            hypercall_engine::order_index::OrderSummary {
                order_id,
                symbol: symbol.clone(),
                side,
                price,
                original_size: msg.info.size,
                remaining_size: resting_quantity,
                is_perp: msg.info.is_perp,
                mmp_enabled: msg.info.mmp_enabled,
                client_id: msg.info.client_id.clone(),
                created_at: timestamp as i64,
            },
        );

        debug!(
            "Replayed create order {}: symbol={}, resting_qty={}, side={:?}",
            order_id, symbol, resting_quantity, side
        );

        // NOTE: We do NOT increment L2 sequence here because:
        // 1. The original order processing already emitted L2 updates
        // 2. The cache will replay those L2 updates from the event stream
        // 3. The engine's L2 seq will be updated after all replays complete
        // Incrementing here would cause engine/cache seq divergence.

        Ok(())
    }

    /// Replay a cancel order command.
    fn replay_cancel_order(&mut self, msg: &OrderActionMessage) -> Result<(), String> {
        let symbol = &msg.info.symbol;
        let wallet = &msg.wallet;

        // Skip commands for instruments that no longer have an orderbook
        // (expired instruments are not loaded during startup).
        if !self.ctx.orderbooks.contains_key(symbol) {
            debug!(
                "Skipping replay cancel for expired/removed instrument {}",
                symbol
            );
            return Ok(());
        }

        // Get the orderbook
        let orderbook = self.ctx.orderbooks.get_mut(symbol).unwrap();

        // Try to find the order by order_id or client_id
        let order_id = if let Some(oid) = msg.info.order_id {
            oid
        } else if let Some(ref client_id) = msg.info.client_id {
            // Look up by client_id via order_index (synchronous, no cache lag)
            match self
                .ctx
                .order_index
                .get_order_by_client_id(wallet, client_id)
            {
                Some((oid, _)) => oid,
                None => {
                    debug!(
                        "Order with client_id {} not found during replay cancel, may already be cancelled",
                        client_id
                    );
                    return Ok(());
                }
            }
        } else {
            return Err("Cancel order has neither order_id nor client_id".to_string());
        };

        // Replay cancel is authoritative. Scrub this order_id from the book
        // even if snapshot metadata drift left it at the wrong level or side.
        if orderbook.cancel_order_for_replay(order_id).is_none() {
            debug!(
                "Order {} not present in replay cancel scan for symbol {}, may already be cancelled/filled",
                order_id, symbol
            );
        }

        // Update order index
        self.ctx.order_index.remove_order(wallet, order_id);

        // NOTE: We do NOT increment L2 sequence here - see replay_create_order comment.

        Ok(())
    }

    /// Recovery-only cancel helper.
    ///
    /// Tries likely symbols first, then scans every orderbook as a fallback.
    /// Replay is authoritative by order_id, so it is better to scrub a ghost
    /// from an unexpected book than to leave a stale resting order behind.
    fn cancel_order_for_replay_with_fallback(
        &mut self,
        order_id: u64,
        candidate_symbols: Vec<String>,
    ) -> Option<String> {
        let mut attempted_symbols = Vec::new();

        for symbol in candidate_symbols {
            if attempted_symbols.iter().any(|seen| seen == &symbol) {
                continue;
            }
            attempted_symbols.push(symbol.clone());

            if let Some(orderbook) = self.ctx.orderbooks.get_mut(&symbol) {
                if orderbook.cancel_order_for_replay(order_id).is_some() {
                    return Some(symbol);
                }
            }
        }

        let remaining_symbols: Vec<String> = self
            .ctx
            .orderbooks
            .keys()
            .filter(|symbol| !attempted_symbols.iter().any(|seen| seen == *symbol))
            .cloned()
            .collect();

        for symbol in remaining_symbols {
            if let Some(orderbook) = self.ctx.orderbooks.get_mut(&symbol) {
                if orderbook.cancel_order_for_replay(order_id).is_some() {
                    error!(
                        "RECOVERY_INVARIANT: order {} was found in unexpected orderbook {} during replay cancel fallback",
                        order_id, symbol
                    );
                    return Some(symbol);
                }
            }
        }

        None
    }

    /// Replay a replace order: cancel the old order and create the new one.
    ///
    /// For replay, `msg.info.order_id` is the cancel target (not patched by journal).
    /// The new order's assigned ID comes from `response.order_id`.
    fn replay_replace_order(
        &mut self,
        msg: &OrderActionMessage,
        response: Option<&OrderUpdateMessage>,
    ) -> Result<(), String> {
        let wallet = &msg.wallet;

        // Phase 1: Always attempt to cancel the old order.
        // This is idempotent: if the order was already cancelled/filled by a
        // prior replay command, nothing happens. We must always try because
        // even a Rejected replace may have succeeded at the cancel phase
        // (e.g., cancel succeeded but the new order was rejected for margin).
        let cancel_order_id = msg
            .info
            .order_id
            .ok_or("Replace order missing order_id for cancel target".to_string())?;

        let mut candidate_symbols = Vec::new();
        if let Some(index_symbol) = self
            .ctx
            .order_index
            .get_order_symbol(wallet, cancel_order_id)
            .map(|s| s.to_string())
        {
            candidate_symbols.push(index_symbol);
        }

        if self
            .cancel_order_for_replay_with_fallback(cancel_order_id, candidate_symbols)
            .is_none()
        {
            debug!(
                "Replay replace: cancel target {} not present in any replay cancel scan",
                cancel_order_id
            );
        }
        self.ctx.order_index.remove_order(wallet, cancel_order_id);

        // Phase 2: Only create the new order if the response shows it succeeded.
        // Rejected means the new order was not placed (either cancel target was
        // not found, or cancel succeeded but create failed for margin/validation).
        if let Some(resp) = response {
            if resp.status == OrderUpdateStatus::Rejected {
                debug!(
                    "Replay replace: skipping create phase (rejected) for wallet {}, cancel_target={}",
                    wallet, cancel_order_id
                );
                return Ok(());
            }
        }

        let mut create_msg = msg.clone();
        create_msg.action = OrderAction::CreateOrder;
        if let Some(resp) = response {
            create_msg.info.order_id = resp.order_id;
        } else {
            create_msg.info.order_id = None;
        }

        self.replay_create_order(&create_msg, response)
    }

    /// Apply fill events from the journal to resting maker orders.
    ///
    /// During `replay_create_order`, each command's `response_json` only
    /// reflects the order's state at the time *that* command was processed.
    /// A maker order whose response said "Open" may have been filled later
    /// by a subsequent taker order. This method applies those fills to both
    /// the orderbook and order_index, closing the gap.
    ///
    /// `fills` should contain all `OrderFilled` events from engine_events
    /// for commands in the replay window (after the snapshot L2 sequence).
    pub(super) fn apply_fills_for_replay(&mut self, fills: &[hypercall_types::Fill]) {
        if fills.is_empty() {
            return;
        }

        let mut applied = 0;
        let mut skipped = 0;

        for fill in fills {
            // Update the order_index (handles partial/full fill, converts raw→human)
            let fully_filled = self.ctx.order_index.fill_order(
                &fill.maker_wallet_address,
                fill.maker_order_id,
                fill.size,
            );

            // Update the orderbook
            if let Some(orderbook) = self.ctx.orderbooks.get_mut(&fill.symbol) {
                if orderbook.has_order(fill.maker_order_id) {
                    orderbook.reduce_order_quantity(fill.maker_order_id, fill.size);
                    applied += 1;
                } else {
                    skipped += 1;
                }
            } else {
                // No orderbook for this symbol. Expected for expired instruments;
                // suspicious otherwise. We warn rather than panic because
                // expired_instruments can be stale after snapshot restore (see
                // replay_create_order comment), so we can't hard-gate on it.
                if self.ctx.expired_instruments.contains_key(&fill.symbol) {
                    debug!(
                        "Replay fill for maker order {} on expired instrument {} \
                         — no orderbook, fill applied to order_index only",
                        fill.maker_order_id, fill.symbol
                    );
                } else {
                    warn!(
                        "Replay fill for maker order {} references non-existent \
                         orderbook {} (not in expired_instruments) — fill applied \
                         to order_index but not orderbook",
                        fill.maker_order_id, fill.symbol
                    );
                }
                skipped += 1;
            }

            if fully_filled {
                debug!(
                    "Replay fill: maker order {} fully filled (trade_id={})",
                    fill.maker_order_id, fill.trade_id
                );
            } else {
                debug!(
                    "Replay fill: maker order {} partially filled by {} (trade_id={})",
                    fill.maker_order_id, fill.size, fill.trade_id
                );
            }
        }

        if applied > 0 || skipped > 0 {
            info!(
                "Applied {} replay fills ({} skipped - order not in book)",
                applied, skipped
            );
        }
    }

    pub(super) fn flush_startup_expired_order_cancels(&mut self) {
        if self.startup_expired_order_cancels.is_empty() {
            return;
        }

        let count = self.startup_expired_order_cancels.len();
        warn!(
            "Startup recovery: {} orders on expired instruments need DB cancellation",
            count
        );

        for cancel in &self.startup_expired_order_cancels {
            warn!(
                "  order_id={}, symbol={}, wallet={}, side={:?}, price={}, size={}, filled={}",
                cancel.order_id,
                cancel.symbol,
                cancel.wallet,
                cancel.side,
                cancel.price,
                cancel.size,
                cancel.filled_size,
            );
        }

        if let Some(ref handler) = self.ctx.db {
            let event_handler =
                crate::db_handler::DieselEventHandler::with_pool_no_migrations(handler.pool());
            let chunk_size = 5000;
            let mut total_rows = 0usize;
            for (i, chunk) in self
                .startup_expired_order_cancels
                .chunks(chunk_size)
                .enumerate()
            {
                match event_handler
                    .batch_cancel_expired_orders_sync(chunk, "Instrument expired during offline")
                {
                    Ok(rows) => {
                        total_rows += rows;
                        info!(
                            "Startup recovery: cancelled chunk {}, {} rows written ({} in chunk, {} total so far)",
                            i + 1, rows, chunk.len(), total_rows
                        );
                    }
                    Err(e) => {
                        error!(
                            "Startup recovery: failed to cancel chunk {} ({} orders): {}. \
                             Remaining orders will show stale OPEN status.",
                            i + 1,
                            chunk.len(),
                            e
                        );
                    }
                }
            }
            info!(
                "Startup recovery: {} candidates, {} total rows cancelled",
                count, total_rows
            );
            metrics::counter!("ht_recovery_expired_order_cancels_total")
                .increment(total_rows as u64);
        }
        self.startup_expired_order_cancels.clear();
    }

    /// Apply side-effect cancels from OrderUpdate events during replay (Pass 3).
    ///
    /// During live processing, some commands produce cancel side effects:
    /// - MMP-triggered cancels: A taker fill triggers MMP, which cancels all
    ///   MMP-enabled orders for that wallet+underlying as a side effect.
    /// - Self-trade prevention: A taker's order is cancelled to prevent self-trade,
    ///   and the matching maker may also be cancelled.
    ///
    /// These cancels are captured as OrderUpdate events in the taker's journal entry
    /// but are NOT separate journal commands. Pass 1 only processes commands, so these
    /// side-effect cancels are lost during replay. This pass re-applies them.
    pub(super) fn apply_cancel_events_for_replay(
        &mut self,
        order_updates: &[hypercall_types::OrderUpdateMessage],
    ) {
        if order_updates.is_empty() {
            return;
        }

        let mut applied = 0;
        let mut skipped = 0;

        for update in order_updates {
            // Only process terminal statuses
            let is_terminal = matches!(
                update.status,
                hypercall_types::OrderUpdateStatus::Canceled
                    | hypercall_types::OrderUpdateStatus::Filled
                    | hypercall_types::OrderUpdateStatus::Rejected
            );
            if !is_terminal {
                continue;
            }

            let Some(order_id) = update.order_id else {
                warn!(
                    "Replay Pass 3: terminal OrderUpdate (status={:?}, wallet={}) \
                     has no order_id — cannot reconcile, skipping",
                    update.status, update.wallet_address
                );
                continue;
            };

            let wallet = &update.wallet_address;
            let symbol = &update.info.symbol;

            if let Some(orderbook) = self.ctx.orderbooks.get_mut(symbol) {
                if orderbook.cancel_order_for_replay(order_id).is_some() {
                    applied += 1;
                    debug!(
                        "Replay Pass 3: removed order {} (status={:?}, wallet={}, symbol={})",
                        order_id, update.status, wallet, symbol
                    );
                } else {
                    skipped += 1;
                }
            }

            self.ctx.order_index.remove_order(wallet, order_id);
        }

        if applied > 0 || skipped > 0 {
            info!(
                "Replay Pass 3: applied {} side-effect cancels ({} skipped - not in book)",
                applied, skipped
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::rsm::MarginMode;
    use hypercall_db::SettlementWriter;
    use hypercall_engine::FillAccounting;
    use hypercall_types::wallet_address::test_wallet;
    use hypercall_types::FillSource;
    use hypercall_types::Side;
    use rust_decimal::Decimal;
    use rust_decimal_macros::dec;

    #[test]
    fn decode_fill_events_accepts_stored_fill_payload() {
        let fill = hypercall_types::Fill {
            trade_id: 77,
            taker_order_id: 101,
            maker_order_id: 102,
            symbol: "BTC-20261231-100000-C".to_string(),
            price: dec!(250),
            size: dec!(100000000),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(2),
            fee: Decimal::ZERO,
            is_taker: true,
            timestamp: 1_700_000_000_000,
            builder_code_address: None,
            builder_code_fee: None,
            source: FillSource::Orderbook,
            taker_realized_pnl: Some(dec!(10)),
            maker_realized_pnl: Some(dec!(-10)),
            underlying_notional: Some(dec!(6500000)),
        };
        let message = hypercall_types::EngineMessage::OrderFilled {
            accounting: FillAccounting::from_fill(&fill),
            fill: fill.clone(),
        };
        let wire = message.serialize_inner().expect("serialize OrderFilled");

        let decoded = UnifiedEngine::decode_fill_events(&[wire]);
        let (decoded_fill, decoded_accounting) = decoded.first().expect("decoded fill");

        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded_fill.trade_id, fill.trade_id);
        assert_eq!(decoded_fill.taker_order_id, fill.taker_order_id);
        assert_eq!(decoded_fill.maker_order_id, fill.maker_order_id);
        assert_eq!(decoded_fill.symbol, fill.symbol);
        assert_eq!(decoded_fill.price, fill.price);
        assert_eq!(decoded_fill.size, fill.size);
        assert_eq!(decoded_fill.taker_side, fill.taker_side);
        assert_eq!(decoded_fill.taker_wallet_address, fill.taker_wallet_address);
        assert_eq!(decoded_fill.maker_wallet_address, fill.maker_wallet_address);
        assert_eq!(decoded_fill.taker_realized_pnl, fill.taker_realized_pnl);
        assert_eq!(decoded_fill.maker_realized_pnl, fill.maker_realized_pnl);
        assert_eq!(*decoded_accounting, FillAccounting::from_fill(&fill));
    }

    #[test]
    fn state_command_replay_decodes_hypercore_equity_update() {
        let wallet = test_wallet(12);
        let mut command_data = vec![1];
        command_data.extend(
            rmp_serde::to_vec(&(wallet, dec!(1234), 300u64)).expect("serialize replay payload"),
        );
        let cmd = hypercall_db::ReplayCommand {
            command_id: 42,
            request_id: "req".to_string(),
            command_type: "HypercoreEquityUpdate".to_string(),
            command_data,
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&cmd).expect("decoded state command");

        assert_eq!(replay_envelope_timestamp(&decoded), 300);
        assert_eq!(
            crate::rsm::restart_components::replay_disposition_for_command(&decoded),
            hypercall_recovery::ReplayDisposition::Applied
        );
        assert!(matches!(
            decoded,
            EngineCommand::HypercoreEquityUpdate {
                wallet: actual_wallet,
                account_value,
                timestamp_ms: 300
            } if actual_wallet == wallet && account_value == dec!(1234)
        ));
    }

    #[test]
    fn state_command_replay_decodes_price_update() {
        let mut command_data = vec![1];
        command_data.extend(
            rmp_serde::to_vec(&crate::rsm::apply::PriceUpdatePayload {
                underlying: "BTC".to_string(),
                spot_price: dec!(95000),
                timestamp_ms: 123,
            })
            .expect("serialize PriceUpdate replay payload"),
        );
        let cmd = hypercall_db::ReplayCommand {
            command_id: 43,
            request_id: "req".to_string(),
            command_type: "PriceUpdate".to_string(),
            command_data,
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&cmd).expect("decoded state command");

        assert_eq!(replay_envelope_timestamp(&decoded), 123);
        assert!(matches!(
            decoded,
            EngineCommand::PriceUpdate {
                underlying,
                spot_price,
                timestamp_ms: 123
            } if underlying == "BTC" && spot_price == dec!(95000)
        ));
    }

    #[test]
    fn state_command_replay_decodes_expiry_commands() {
        let market = hypercall_types::Market {
            symbol: "BTC-20260619-110000-C".to_string(),
            underlying: "BTC".to_string(),
            expiry: 20260619,
            strike: dec!(110000),
            option_type: hypercall_types::OptionType::Call,
        };
        let expire = hypercall_db::ReplayCommand {
            command_id: 44,
            request_id: "expire-market".to_string(),
            command_type: "ExpireMarket".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::MarketActionCommand::with_expiry_context(
                    hypercall_types::MarketActionMessage {
                        market: market.clone(),
                        action: hypercall_types::MarketAction::ExpireMarket,
                        timestamp: 459,
                    },
                    crate::rsm::apply::TickExpiryContext {
                        due_expiries: vec![crate::rsm::apply::TickExpiryDueGroup {
                            expiry_ts: 20260619,
                            symbols: vec![market.symbol.clone()],
                        }],
                        pending_settlements: Vec::new(),
                        settlement_prices: vec![crate::rsm::apply::TickExpirySettlementPrice {
                            underlying: "BTC".to_string(),
                            expiry_ts: 20260619,
                            price: dec!(120000),
                        }],
                        margin_modes: Vec::new(),
                        pm_settlements: Vec::new(),
                    },
                ),
            ),
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&expire).expect("decoded expiry command");

        assert_eq!(replay_envelope_timestamp(&decoded), 459);
        assert!(UnifiedEngine::is_expiry_replay_command(&decoded));
        assert!(matches!(
            decoded,
            EngineCommand::MarketAction(command)
                if command.message.action == hypercall_types::MarketAction::ExpireMarket
                    && command.expiry_context.is_some()
        ));

        let tick = hypercall_db::ReplayCommand {
            command_id: 45,
            request_id: "tick-expiry".to_string(),
            command_type: "TickExpiry".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(&(
                460u64,
                crate::rsm::apply::TickExpiryContext::empty(),
            )),
            response_data: None,
        };

        let decoded_tick = decode_state_command_for_replay(&tick).expect("decoded tick command");

        assert_eq!(replay_envelope_timestamp(&decoded_tick), 460);
        assert!(UnifiedEngine::is_expiry_replay_command(&decoded_tick));
        assert!(matches!(decoded_tick, EngineCommand::TickExpiry { .. }));
    }

    #[tokio::test]
    async fn replay_command_row_tick_expiry_never_reconciles_from_durable_settlement() {
        let context = crate::rsm::unified_engine::tests::setup_test_context().await;
        let diesel_handler = std::sync::Arc::new(
            DatabaseHandler::new(&context.database_url).expect("diesel handler should connect"),
        );

        let wallet = test_wallet(156);
        let symbol = "BTC-20261231-100000-C".to_string();
        let expiry_ts = hypercall_types::expiry_date_to_timestamp("BTC", 20261231) as u64;
        let reference_price = dec!(105000);
        let settlement_price = dec!(5000);
        let position_size = dec!(1);
        let settlement_entry_price = dec!(100000);
        let settlement_value = settlement_price * position_size;
        let cost_basis = settlement_entry_price * position_size;
        let net_pnl = settlement_value - cost_basis;

        diesel_handler
            .try_apply_settlement_sync(
                &wallet,
                &symbol,
                position_size,
                settlement_price,
                settlement_value,
                MarginMode::Standard,
                1_798_675_200_000,
                Some(settlement_entry_price),
                Some(cost_basis),
                Some(net_pnl),
            )
            .expect("durable settlement should be present before replay");

        let (engine_event_tx, _engine_event_rx) = tokio::sync::mpsc::unbounded_channel();
        let (shutdown_tx, _) = tokio::sync::broadcast::channel::<()>(1);
        let (mut engine, _order_tx, _market_tx) =
            UnifiedEngineBuilder::new(crate::rsm::unified_engine::tests::create_test_config())
                .with_database(&context.database_url)
                .allow_no_database_for_tests()
                .with_mock_journal()
                .build(engine_event_tx, shutdown_tx);

        engine
            .expiry_manager
            .schedule_expiry(symbol.clone(), expiry_ts);
        engine.ctx.engine_positions.insert(
            (wallet, symbol.clone()),
            crate::rsm::engine_deps::EnginePosition {
                quantity: position_size,
                entry_price: settlement_entry_price,
            },
        );
        engine.ctx.balance_ledger.set(wallet, dec!(10000));

        let cmd = hypercall_db::ReplayCommand {
            command_id: 156,
            request_id: "tick-expiry-divergent-cash".to_string(),
            command_type: "TickExpiry".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(&(
                1_798_675_200_000_u64,
                crate::rsm::apply::TickExpiryContext {
                    due_expiries: vec![crate::rsm::apply::TickExpiryDueGroup {
                        expiry_ts: expiry_ts as i64,
                        symbols: vec![symbol],
                    }],
                    pending_settlements: Vec::new(),
                    settlement_prices: vec![crate::rsm::apply::TickExpirySettlementPrice {
                        underlying: "BTC".to_string(),
                        expiry_ts: expiry_ts as i64,
                        price: reference_price,
                    }],
                    margin_modes: vec![crate::rsm::apply::TickExpiryWalletMarginMode {
                        wallet,
                        margin_mode: MarginMode::Standard,
                        pm_settlement_required: false,
                    }],
                    pm_settlements: Vec::new(),
                },
            )),
            response_data: None,
        };

        let mut replay_count = 0;
        let mut max_replayed_order_id = 0;
        engine.replay_command_row(&cmd, &mut replay_count, &mut max_replayed_order_id);

        assert_eq!(
            engine.ctx.balance_ledger.balance(&wallet),
            dec!(15000),
            "replay must keep deterministic engine cash and never reconcile from durable settlement rows"
        );
    }

    #[test]
    fn state_command_replay_ignores_non_state_commands() {
        let cmd = hypercall_db::ReplayCommand {
            command_id: 46,
            request_id: "req".to_string(),
            command_type: "CreateOrder".to_string(),
            command_data: vec![1],
            response_data: None,
        };

        assert!(decode_state_command_for_replay(&cmd).is_none());
    }

    #[test]
    fn replay_command_row_applies_legacy_agent_auth_without_nonce() {
        let (mut engine, _order_tx, _market_tx) =
            crate::rsm::unified_engine::tests::build_replay_test_engine();
        let wallet = test_wallet(52);
        let agent = test_wallet(53);
        let mut replay_count = 0;
        let mut max_replayed_order_id = 0;

        let approve = hypercall_db::ReplayCommand {
            command_id: 45,
            request_id: "legacy-approve-agent".to_string(),
            command_type: "ApproveAgent".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(&(wallet, agent, None::<u64>)),
            response_data: None,
        };

        engine.replay_command_row(&approve, &mut replay_count, &mut max_replayed_order_id);

        assert_eq!(replay_count, 1);
        assert_eq!(max_replayed_order_id, 0);
        assert_eq!(
            engine
                .ctx
                .agent_authorizations
                .get(&wallet)
                .and_then(|agents| agents.get(&agent)),
            Some(&None),
            "legacy persisted ApproveAgent replay row without nonce must restore auth"
        );
        let revoke = hypercall_db::ReplayCommand {
            command_id: 46,
            request_id: "legacy-revoke-agent".to_string(),
            command_type: "RevokeAgent".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(&(wallet, agent)),
            response_data: None,
        };

        engine.replay_command_row(&revoke, &mut replay_count, &mut max_replayed_order_id);

        assert_eq!(replay_count, 2);
        assert!(
            !engine.ctx.agent_authorizations.contains_key(&wallet),
            "legacy persisted RevokeAgent replay row without nonce must clear auth"
        );
    }

    #[test]
    fn state_command_replay_decodes_named_agent_auth_payloads() {
        let wallet = test_wallet(54);
        let agent = test_wallet(55);
        let approve = hypercall_db::ReplayCommand {
            command_id: 47,
            request_id: "approve-agent".to_string(),
            command_type: "ApproveAgent".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::ApproveAgentPayload {
                    wallet,
                    agent,
                    expires_at_ms: Some(999),
                    nonce: Some(123),
                    timestamp_ms: 456,
                },
            ),
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&approve).expect("decoded approve");

        assert!(matches!(
            decoded,
            EngineCommand::ApproveAgent {
                wallet: decoded_wallet,
                agent: decoded_agent,
                expires_at_ms: Some(999),
                nonce: Some(123),
                timestamp_ms: 456,
            } if decoded_wallet == wallet && decoded_agent == agent
        ));

        let revoke = hypercall_db::ReplayCommand {
            command_id: 48,
            request_id: "revoke-agent".to_string(),
            command_type: "RevokeAgent".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::RevokeAgentPayload {
                    wallet,
                    agent,
                    nonce: Some(124),
                    timestamp_ms: 457,
                },
            ),
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&revoke).expect("decoded revoke");

        assert!(matches!(
            decoded,
            EngineCommand::RevokeAgent {
                wallet: decoded_wallet,
                agent: decoded_agent,
                nonce: Some(124),
                timestamp_ms: 457,
            } if decoded_wallet == wallet && decoded_agent == agent
        ));
    }

    #[test]
    fn state_command_replay_decodes_named_balance_update_payloads() {
        let wallet = test_wallet(56);
        let cmd = hypercall_db::ReplayCommand {
            command_id: 49,
            request_id: "deposit-update".to_string(),
            command_type: "DepositUpdate".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::DepositUpdatePayload {
                    wallet,
                    amount: dec!(50),
                    timestamp_ms: 458,
                    sequence: 125,
                    source_event_hash: test_source_hash(125),
                },
            ),
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&cmd).expect("decoded deposit update");

        assert!(matches!(
            decoded,
            EngineCommand::DepositUpdate {
                wallet: decoded_wallet,
                amount,
                timestamp_ms: 458,
                sequence: Some(125),
                source_event_hash,
            } if decoded_wallet == wallet && amount == dec!(50) && source_event_hash == test_source_hash(125)
        ));
    }

    #[test]
    fn state_command_replay_decodes_legacy_deposit_balance_payloads() {
        let wallet = test_wallet(57);
        let cmd = hypercall_db::ReplayCommand {
            command_id: 50,
            request_id: "legacy-deposit-update".to_string(),
            command_type: "DepositUpdate".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::BalanceCommandPayload {
                    wallet,
                    amount: dec!(50),
                    balance_after: dec!(150),
                    timestamp_ms: 459,
                    sequence: Some(126),
                },
            ),
            response_data: None,
        };

        let decoded = decode_state_command_for_replay(&cmd).expect("decoded legacy deposit update");

        match decoded {
            EngineCommand::DepositUpdate {
                wallet: decoded_wallet,
                amount,
                timestamp_ms,
                sequence,
                source_event_hash,
            } => {
                assert_eq!(decoded_wallet, wallet);
                assert_eq!(amount, dec!(50));
                assert_eq!(timestamp_ms, 459);
                assert_eq!(sequence, Some(126));
                assert_ne!(source_event_hash, alloy::primitives::FixedBytes::<32>::ZERO);
            }
            other => panic!("expected DepositUpdate, got {other:?}"),
        }
    }

    #[test]
    #[should_panic(expected = "Legacy replay DepositUpdate balance payload missing sequence")]
    fn state_command_replay_rejects_legacy_deposit_balance_payloads_without_sequence() {
        let wallet = test_wallet(58);
        let cmd = hypercall_db::ReplayCommand {
            command_id: 51,
            request_id: "legacy-deposit-update-no-sequence".to_string(),
            command_type: "DepositUpdate".to_string(),
            command_data: hypercall_types::serialize_to_wire_bytes(
                &crate::rsm::apply::BalanceCommandPayload {
                    wallet,
                    amount: dec!(50),
                    balance_after: dec!(150),
                    timestamp_ms: 460,
                    sequence: None,
                },
            ),
            response_data: None,
        };

        let _ = decode_state_command_for_replay(&cmd);
    }
}