hypercall/rsm/unified_engine/

runtime.rs

//! Runtime loop and flush coordination for UnifiedEngine.

use super::*;
use crate::shared::order_types::ParsedSymbol;
use crate::vol_oracle::{VolPoint, VolProviderKind, VolSurfaceSnapshot};
use hypercall_db::{
    DirectiveOutboxReader, OrderWriter, PmSettlementProjectionSyncWriter, SettlementWriter,
};
use rust_decimal::prelude::ToPrimitive;
use std::collections::BTreeSet;

const STANDBY_SETTLEMENT_OBSERVE_ATTEMPTS: usize = 8;
const STANDBY_SETTLEMENT_OBSERVE_INITIAL_BACKOFF_MS: u64 = 25;
const STANDBY_SETTLEMENT_OBSERVE_MAX_BACKOFF_MS: u64 = 500;

fn synthesize_fixed_surface_snapshot(
    orderbooks: &std::collections::HashMap<String, hypercall_engine::OrderBook>,
    vol_oracle: &crate::vol_oracle::SharedVolOracle,
    statuses: &[crate::vol_oracle::VolOracleStatus],
    underlying: &str,
    now_ms: i64,
) -> Option<VolSurfaceSnapshot> {
    let has_ready_fixed = statuses.iter().any(|status| {
        status.underlying == underlying && status.provider == VolProviderKind::Fixed && status.ready
    });
    if !has_ready_fixed {
        return None;
    }

    let mut strike_points = Vec::new();
    let mut seen = BTreeSet::new();
    for symbol in orderbooks.keys() {
        let Ok(parsed) = ParsedSymbol::from_symbol(symbol) else {
            continue;
        };
        if parsed.underlying != underlying {
            continue;
        }

        let Some(strike) = parsed.strike.to_f64() else {
            panic!(
                "STATE_CORRUPTION: strike {} for {} is not representable as f64",
                parsed.strike, symbol
            );
        };
        let expiry_ts =
            hypercall_types::expiry_date_to_timestamp(&parsed.underlying, parsed.expiry) as i64;
        let dedupe_key = (expiry_ts, parsed.strike.to_string());
        if !seen.insert(dedupe_key) {
            continue;
        }

        let iv = vol_oracle.get_iv(underlying, strike, expiry_ts).ok()?;
        strike_points.push(VolPoint {
            strike,
            expiry: expiry_ts,
            iv,
            timestamp: now_ms,
        });
    }

    if strike_points.is_empty() {
        return None;
    }

    let expiries = strike_points
        .iter()
        .map(|point| point.expiry)
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();

    Some(VolSurfaceSnapshot {
        underlying: underlying.to_string(),
        last_update_ts_ms: Some(now_ms),
        expiries,
        strike_points,
        delta_curves: Vec::new(),
        atm_vols: Vec::new(),
        spot_price: None,
    })
}

fn max_listed_option_expiry_ts_ms(
    orderbooks: &std::collections::HashMap<String, hypercall_engine::OrderBook>,
    underlying: &str,
) -> Result<i64, String> {
    orderbooks
        .keys()
        .filter_map(|symbol| ParsedSymbol::from_symbol(symbol).ok())
        .filter(|parsed| parsed.underlying == underlying)
        .map(|parsed| {
            hypercall_types::expiry_date_to_timestamp_checked(&parsed.underlying, parsed.expiry)
                .map_err(|error| {
                    format!(
                        "invalid listed option expiry {} for {}: {}",
                        parsed.expiry, parsed.underlying, error
                    )
                })
                .and_then(|expiry_ts| {
                    i64::try_from(expiry_ts)
                        .map_err(|_| "listed option expiry exceeds i64".to_string())?
                        .checked_mul(1_000)
                        .ok_or_else(|| "listed option expiry ms overflow".to_string())
                })
        })
        .try_fold(None, |max: Option<i64>, expiry_ts_ms| {
            let expiry_ts_ms = expiry_ts_ms?;
            Ok::<_, String>(Some(
                max.map_or(expiry_ts_ms, |current| current.max(expiry_ts_ms)),
            ))
        })?
        .ok_or_else(|| format!("no listed option markets for underlying {underlying}"))
}

impl UnifiedEngine {
    /// Fail before mutation when an external durable operation enters the
    /// engine without a UUID-backed journal identity.
    fn require_external_durable_mutation_uuid(command_type: &str, request_id: &str) -> uuid::Uuid {
        uuid::Uuid::parse_str(request_id).unwrap_or_else(|error| {
            panic!(
                "RUNTIME_INVARIANT: external engine command {} request_id {} is not a UUID: {}",
                command_type, request_id, error
            )
        })
    }

    fn parse_directive_domain_status(
        value: &str,
    ) -> Result<crate::directive_outbox::DirectiveDomainStatus, String> {
        match value {
            "accepted" => Ok(crate::directive_outbox::DirectiveDomainStatus::Accepted),
            "rejected" => Ok(crate::directive_outbox::DirectiveDomainStatus::Rejected),
            "pending_chain_effect" => {
                Ok(crate::directive_outbox::DirectiveDomainStatus::PendingChainEffect)
            }
            "completed" => Ok(crate::directive_outbox::DirectiveDomainStatus::Completed),
            "failed" => Ok(crate::directive_outbox::DirectiveDomainStatus::Failed),
            other => Err(format!("unknown directive domain_status {}", other)),
        }
    }

    fn parse_directive_delivery_status(
        value: &str,
    ) -> Result<crate::directive_outbox::DirectiveDeliveryStatus, String> {
        match value {
            "pending" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Pending),
            "broadcasted" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Broadcasted),
            "included" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Included),
            "finalized" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Finalized),
            "reverted" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Reverted),
            "expired" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::Expired),
            "dead_lettered" => Ok(crate::directive_outbox::DirectiveDeliveryStatus::DeadLettered),
            other => Err(format!("unknown directive delivery_status {}", other)),
        }
    }

    fn persisted_withdrawal_directive_status(
        &self,
        request_id: &str,
    ) -> Result<
        Option<(
            crate::directive_outbox::DirectiveDomainStatus,
            crate::directive_outbox::DirectiveDeliveryStatus,
        )>,
        String,
    > {
        let Some(db) = self.ctx.db.as_ref() else {
            warn!(
                request_id = %request_id,
                "Cannot read persisted directive status without DB handle; returning pending retry receipt"
            );
            return Ok(None);
        };
        let row = db
            .get_directive_status_sync(request_id)
            .map_err(|error| format!("failed to read directive status for {}: {}", request_id, error))?
            .unwrap_or_else(|| {
                panic!(
                    "JOURNAL_FATAL: withdrawal request_id {} is journaled but missing directive_outbox row",
                    request_id
                )
            });

        Ok(Some((
            Self::parse_directive_domain_status(&row.domain_status)?,
            Self::parse_directive_delivery_status(&row.delivery_status)?,
        )))
    }

    async fn external_option_command_already_journaled(
        &self,
        request_id: &str,
        command_type: &str,
    ) -> bool {
        let request_uuid = Self::require_external_durable_mutation_uuid(command_type, request_id);
        let Some(journal_writer) = self.journal_writer.clone() else {
            return false;
        };
        let command_type = command_type.to_string();
        let result =
            tokio::task::spawn_blocking(move || journal_writer.get_by_request_id(&request_uuid))
                .await
                .unwrap_or_else(|error| {
                    panic!(
                        "JOURNAL_FATAL: idempotency lookup task failed for {}: {}",
                        request_id, error
                    )
                })
                .unwrap_or_else(|error| {
                    panic!(
                        "JOURNAL_FATAL: idempotency lookup failed for {}: {}",
                        request_id, error
                    )
                });

        match result {
            Some(record) if record.command_type == command_type => true,
            Some(record) => {
                panic!(
                    "JOURNAL_FATAL: request_id {} already used for {}, not {}",
                    request_id, record.command_type, command_type
                )
            }
            None => false,
        }
    }

    async fn journal_external_option_position_command(
        &self,
        env: &crate::rsm::apply::CommandEnvelope,
        request_id: &str,
        command_type_enum: hypercall_db_diesel::engine_enums::CommandType,
        outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
    ) {
        use crate::journal::{JournalEntry, JournalMessage};
        use crate::rsm::apply::EngineCommand;
        use hypercall_db_diesel::engine_enums::DbUuid;

        let request_uuid =
            Self::require_external_durable_mutation_uuid(env.command.command_type(), request_id);
        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();
        let command_data = match &env.command {
            EngineCommand::OptionDepositUpdate {
                request_id,
                wallet,
                symbol,
                quantity,
                timestamp_ms,
            } => hypercall_types::serialize_to_wire_bytes(&(
                request_id,
                wallet,
                symbol,
                quantity,
                timestamp_ms,
            )),
            EngineCommand::OptionWithdrawalUpdate {
                request_id,
                wallet,
                account,
                signer,
                rsm_signer,
                symbol,
                quantity,
                nonce,
                action,
                timestamp_ms,
            } => hypercall_types::serialize_to_wire_bytes(&(
                request_id,
                wallet,
                account,
                signer,
                rsm_signer,
                symbol,
                quantity,
                nonce,
                action,
                timestamp_ms,
            )),
            other => panic!(
                "RUNTIME_INVARIANT: journal_external_option_position_command called for {}",
                other.command_type()
            ),
        };

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let entry = JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends,
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: Some(ack_tx),
                request_uuid: DbUuid(request_uuid),
                command_type_enum: Some(command_type_enum),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            };
            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "CRITICAL_FAILURE: journal commit_ack dropped for {} request_id {}. \
                     Durability boundary is unknown after state mutation.",
                    env.command.command_type(),
                    request_id
                );
            }
            return;
        }

        let Some(ref journal_writer) = self.journal_writer else {
            panic!(
                "JOURNAL_FATAL: no journal configured for {} request_id {}",
                env.command.command_type(),
                request_id
            );
        };
        if !outbox_appends.is_empty() {
            panic!(
                "JOURNAL_FATAL: synchronous journal path cannot persist directive outbox appends for {} request_id {}",
                env.command.command_type(),
                request_id
            );
        }
        let balance_updates = Vec::new();
        journal_writer
            .append_transition_with_fill_side_effects(
                env.received_ts_ms,
                &command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                &[],
                &balance_updates,
                DbUuid(request_uuid),
                Some(command_type_enum),
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                )
            });
    }

    async fn journal_external_balance_update_command(
        &self,
        env: &crate::rsm::apply::CommandEnvelope,
        request_id: &str,
        outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
        balance_updates: Vec<hypercall_types::BalanceUpdate>,
    ) {
        use crate::journal::{JournalEntry, JournalMessage};
        use crate::rsm::apply::{BalanceCommandPayload, DepositUpdatePayload, EngineCommand};
        use hypercall_db_diesel::engine_enums::{CommandType, DbUuid};

        let request_uuid =
            Self::require_external_durable_mutation_uuid(env.command.command_type(), request_id);
        let (command_data, command_type_enum) = match &env.command {
            EngineCommand::DepositUpdate {
                wallet,
                amount,
                timestamp_ms,
                sequence,
                source_event_hash,
            } => {
                let Some(sequence) = sequence else {
                    panic!(
                        "RUNTIME_INVARIANT: DepositUpdate {} missing durable sequence",
                        request_id
                    );
                };
                (
                    hypercall_types::serialize_to_wire_bytes(&DepositUpdatePayload {
                        wallet: *wallet,
                        amount: *amount,
                        timestamp_ms: *timestamp_ms,
                        sequence: *sequence,
                        source_event_hash: source_event_hash.clone(),
                    }),
                    CommandType::DepositUpdate,
                )
            }
            EngineCommand::LiquidationBonusUpdate {
                wallet,
                amount,
                balance_after,
                timestamp_ms,
                sequence,
            } => (
                hypercall_types::serialize_to_wire_bytes(&BalanceCommandPayload {
                    wallet: *wallet,
                    amount: *amount,
                    balance_after: *balance_after,
                    timestamp_ms: *timestamp_ms,
                    sequence: *sequence,
                }),
                CommandType::LiquidationBonusUpdate,
            ),
            _ => {
                panic!(
                    "RUNTIME_INVARIANT: journal_external_balance_update_command called for {}",
                    env.command.command_type()
                );
            }
        };
        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let entry = JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends,
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: None,
                balance_updates,
                created_at: Instant::now(),
                commit_ack: Some(ack_tx),
                request_uuid: DbUuid(request_uuid),
                command_type_enum: Some(command_type_enum),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            };
            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "CRITICAL_FAILURE: journal commit_ack dropped for {} request_id {}. \
                     Durability boundary is unknown after state mutation.",
                    env.command.command_type(),
                    request_id
                );
            }
            return;
        }

        let Some(ref journal_writer) = self.journal_writer else {
            panic!(
                "JOURNAL_FATAL: no journal configured for {} request_id {}",
                env.command.command_type(),
                request_id
            );
        };
        if !outbox_appends.is_empty() {
            panic!(
                "JOURNAL_FATAL: synchronous journal path cannot persist directive outbox appends for {} request_id {}",
                env.command.command_type(),
                request_id
            );
        }
        journal_writer
            .append_transition_with_fill_side_effects(
                env.received_ts_ms,
                &command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                &[],
                &balance_updates,
                DbUuid(request_uuid),
                Some(command_type_enum),
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                )
            });
    }

    fn pm_settlement_request_id(command: &crate::rsm::apply::EngineCommand) -> Option<uuid::Uuid> {
        match command {
            crate::rsm::apply::EngineCommand::SetPmSettlementPoolConfig(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::ApplyPmSettlementRepayment(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::JournalPmRecoveryPlan(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::MarkPmRecoveryActionSubmitted(command) => {
                Some(command.request_id)
            }
            crate::rsm::apply::EngineCommand::ResolvePmRecoveryAction(command) => {
                Some(command.request_id)
            }
            _ => None,
        }
    }

    fn preflight_pm_settlement_admin_command(
        &self,
        command: &crate::rsm::apply::EngineCommand,
    ) -> Result<(), String> {
        let mut pm_state = self.ctx.pm_settlement_state.clone();
        match command.clone() {
            crate::rsm::apply::EngineCommand::SetPmSettlementPoolConfig(command) => {
                pm_state.apply_set_config(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(command) => {
                let engine_max_expiry_ts_ms =
                    max_listed_option_expiry_ts_ms(&self.ctx.orderbooks, &command.underlying)?;
                if command.max_listed_expiry_ts_ms != engine_max_expiry_ts_ms {
                    return Err(format!(
                        "RecordPmVaultDeposit max_listed_expiry_ts_ms {} does not match engine max listed expiry {} for {}",
                        command.max_listed_expiry_ts_ms,
                        engine_max_expiry_ts_ms,
                        command.underlying
                    ));
                }
                pm_state.apply_record_vault_deposit(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(command) => {
                pm_state.apply_request_vault_withdrawal(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(command) => {
                pm_state.apply_accrue_interest(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::ApplyPmSettlementRepayment(command) => {
                pm_state.apply_repayment(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::JournalPmRecoveryPlan(command) => {
                pm_state.apply_journal_recovery_plan(command).map(|_| ())
            }
            crate::rsm::apply::EngineCommand::MarkPmRecoveryActionSubmitted(command) => pm_state
                .apply_mark_recovery_action_submitted(command)
                .map(|_| ()),
            crate::rsm::apply::EngineCommand::ResolvePmRecoveryAction(command) => {
                pm_state.apply_resolve_recovery_action(command).map(|_| ())
            }
            other => Err(format!(
                "unsupported PM settlement admin command {}",
                other.command_type()
            )),
        }
    }

    fn stamp_pm_settlement_admin_command(
        command: crate::rsm::apply::EngineCommand,
        timestamp_ms: u64,
    ) -> crate::rsm::apply::EngineCommand {
        match command {
            crate::rsm::apply::EngineCommand::SetPmSettlementPoolConfig(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::SetPmSettlementPoolConfig(command)
            }
            crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(command)
            }
            crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(command)
            }
            crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(command)
            }
            crate::rsm::apply::EngineCommand::ApplyPmSettlementRepayment(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::ApplyPmSettlementRepayment(command)
            }
            crate::rsm::apply::EngineCommand::JournalPmRecoveryPlan(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::JournalPmRecoveryPlan(command)
            }
            crate::rsm::apply::EngineCommand::MarkPmRecoveryActionSubmitted(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::MarkPmRecoveryActionSubmitted(command)
            }
            crate::rsm::apply::EngineCommand::ResolvePmRecoveryAction(mut command) => {
                command.timestamp_ms = timestamp_ms;
                crate::rsm::apply::EngineCommand::ResolvePmRecoveryAction(command)
            }
            command => command,
        }
    }

    async fn journal_pm_settlement_admin_command(
        &self,
        env: &crate::rsm::apply::CommandEnvelope,
        request_uuid: uuid::Uuid,
    ) {
        use crate::engine_enums_ext::CommandTypeExt;
        use crate::journal::{JournalEntry, JournalMessage};
        use hypercall_db_diesel::engine_enums::{CommandType, DbUuid};

        let command_type_enum = CommandType::from_command(&env.command);
        let command_data = match &env.command {
            crate::rsm::apply::EngineCommand::SetPmSettlementPoolConfig(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::ApplyPmSettlementRepayment(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::JournalPmRecoveryPlan(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::MarkPmRecoveryActionSubmitted(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            crate::rsm::apply::EngineCommand::ResolvePmRecoveryAction(command) => {
                hypercall_types::serialize_to_wire_bytes(command)
            }
            _ => {
                panic!(
                    "RUNTIME_INVARIANT: journal_pm_settlement_admin_command called for {}",
                    env.command.command_type()
                );
            }
        };
        let request_id = request_uuid.to_string();
        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let entry = JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends: Vec::new(),
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: Some(ack_tx),
                request_uuid: DbUuid(request_uuid),
                command_type_enum: Some(command_type_enum),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            };
            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "CRITICAL_FAILURE: journal commit_ack dropped for {} request_id {}. \
                     Durability boundary is unknown after state mutation.",
                    env.command.command_type(),
                    request_id
                );
            }
            return;
        }

        let Some(ref journal_writer) = self.journal_writer else {
            panic!(
                "JOURNAL_FATAL: no journal configured for {} request_id {}",
                env.command.command_type(),
                request_id
            );
        };
        journal_writer
            .append_transition_with_fill_side_effects(
                env.received_ts_ms,
                &command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                &[],
                &[],
                DbUuid(request_uuid),
                Some(command_type_enum),
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for request_id {}: {}",
                    env.command.command_type(),
                    request_id,
                    error
                )
            });
    }

    async fn journal_agent_auth_command(&self, env: &crate::rsm::apply::CommandEnvelope) {
        use crate::journal::{JournalEntry, JournalMessage};
        use crate::rsm::apply::{ApproveAgentPayload, EngineCommand, RevokeAgentPayload};
        use hypercall_db_diesel::engine_enums::{CommandType, DbUuid};

        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();

        let (command_data, command_type_enum) = match &env.command {
            EngineCommand::ApproveAgent {
                wallet,
                agent,
                expires_at_ms,
                nonce,
                timestamp_ms,
            } => (
                hypercall_types::serialize_to_wire_bytes(&ApproveAgentPayload {
                    wallet: *wallet,
                    agent: *agent,
                    expires_at_ms: *expires_at_ms,
                    nonce: *nonce,
                    timestamp_ms: *timestamp_ms,
                }),
                CommandType::ApproveAgent,
            ),
            EngineCommand::RevokeAgent {
                wallet,
                agent,
                nonce,
                timestamp_ms,
            } => (
                hypercall_types::serialize_to_wire_bytes(&RevokeAgentPayload {
                    wallet: *wallet,
                    agent: *agent,
                    nonce: *nonce,
                    timestamp_ms: *timestamp_ms,
                }),
                CommandType::RevokeAgent,
            ),
            other => panic!(
                "RUNTIME_INVARIANT: journal_agent_auth_command called for {}",
                other.command_type()
            ),
        };

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let entry = JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends: Vec::new(),
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: Some(ack_tx),
                request_uuid: DbUuid(uuid::Uuid::new_v4()),
                command_type_enum: Some(command_type_enum),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            };
            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to journal {}: {}",
                    env.command.command_type(),
                    error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "CRITICAL_FAILURE: journal commit_ack dropped for {}. \
                     Durability boundary is unknown after state mutation.",
                    env.command.command_type()
                );
            }
            return;
        }

        let Some(ref journal_writer) = self.journal_writer else {
            panic!(
                "JOURNAL_FATAL: no journal configured for {}",
                env.command.command_type()
            );
        };
        journal_writer
            .append_transition(
                env.received_ts_ms,
                &command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                DbUuid(uuid::Uuid::new_v4()),
                Some(command_type_enum),
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal {}: {}",
                    env.command.command_type(),
                    error
                )
            });
    }

    fn agent_auth_journal_available(&self) -> bool {
        self.journal_batch_sender.is_some()
            || self
                .journal_writer
                .as_ref()
                .is_some_and(|writer| writer.is_durable())
    }

    pub(crate) fn withdrawal_directive_journal_unavailable_error(
        &self,
        command_type: &str,
        request_id: &str,
    ) -> Option<String> {
        if self.journal_batch_sender.is_some() {
            return None;
        }
        Some(format!(
            "{command_type} requires ENGINE_JOURNAL_ENABLED=true because directive outbox appends \
             must be persisted atomically with the state transition; request_id {request_id} was not applied"
        ))
    }

    async fn journal_external_cash_withdrawal_command(
        &self,
        env: &crate::rsm::apply::CommandEnvelope,
        request_id: &str,
        outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
        wallet: hypercall_types::WalletAddress,
        amount: rust_decimal::Decimal,
        balance_after: rust_decimal::Decimal,
        timestamp_ms: u64,
        balance_updates: Vec<hypercall_types::BalanceUpdate>,
    ) {
        use crate::journal::engine_journal_batcher::{
            JournalCashWithdrawalSideEffect, JournalEntry, JournalMessage,
        };
        use crate::rsm::apply::EngineCommand;
        use hypercall_db_diesel::engine_enums::{CommandType, DbUuid};

        let request_uuid =
            Self::require_external_durable_mutation_uuid("CashWithdrawalUpdate", request_id);
        let EngineCommand::CashWithdrawalUpdate {
            request_id: ref rid,
            wallet: w,
            account: a,
            destination: d,
            signer: s,
            rsm_signer: rs,
            amount: amt,
            amount_wei: aw,
            nonce: n,
            timestamp_ms: ts,
        } = env.command
        else {
            panic!(
                "RUNTIME_INVARIANT: journal_external_cash_withdrawal_command called for {}",
                env.command.command_type()
            );
        };
        let command_data = hypercall_types::serialize_to_wire_bytes(&(
            rid, &w, &a, &d, &s, &rs, &amt, &aw, &n, &ts,
        ));
        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();

        let side_effect = JournalCashWithdrawalSideEffect {
            wallet,
            request_id: request_id.to_string(),
            amount,
            balance_after,
            timestamp_ms,
        };

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let entry = JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends,
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: Some(side_effect),
                balance_updates,
                created_at: std::time::Instant::now(),
                commit_ack: Some(ack_tx),
                request_uuid: DbUuid(request_uuid),
                command_type_enum: Some(CommandType::CashWithdrawalUpdate),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            };
            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to journal CashWithdrawalUpdate for request_id {}: {}",
                    request_id, error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "CRITICAL_FAILURE: journal commit_ack dropped for CashWithdrawalUpdate request_id {}. \
                     Durability boundary is unknown after state mutation.",
                    request_id
                );
            }
            return;
        }

        let Some(ref journal_writer) = self.journal_writer else {
            panic!(
                "JOURNAL_FATAL: no journal configured for CashWithdrawalUpdate request_id {}",
                request_id
            );
        };
        if !outbox_appends.is_empty() {
            panic!(
                "JOURNAL_FATAL: synchronous journal path cannot persist directive outbox appends for CashWithdrawalUpdate request_id {}",
                request_id
            );
        }
        journal_writer
            .append_transition_with_fill_side_effects(
                env.received_ts_ms,
                &command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                &[],
                &balance_updates,
                DbUuid(request_uuid),
                Some(CommandType::CashWithdrawalUpdate),
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal CashWithdrawalUpdate for request_id {}: {}",
                    request_id, error
                )
            });
    }

    async fn prepare_tick_expiry_env(
        &self,
        now_ms: u64,
    ) -> Result<crate::rsm::apply::CommandEnvelope, EngineError> {
        let context = self
            .expiry_manager
            .prepare_tick_expiry_context(now_ms, &self.ctx, &self.margin_manager)
            .await
            .map_err(EngineError::Internal)?;
        self.validate_tick_expiry_nats_payload(now_ms, &context)?;
        Ok(crate::rsm::apply::CommandEnvelope::new(
            now_ms,
            crate::rsm::apply::EngineCommand::TickExpiry { now_ms, context },
        ))
    }

    fn replay_owned_expiry_command_payload(
        env: &crate::rsm::apply::CommandEnvelope,
    ) -> Option<(
        hypercall_db_diesel::engine_enums::CommandType,
        Vec<u8>,
        &'static str,
    )> {
        match &env.command {
            crate::rsm::apply::EngineCommand::MarketAction(command)
                if matches!(
                    command.message.action,
                    hypercall_types::MarketAction::ExpireMarket
                ) =>
            {
                Some((
                    hypercall_db_diesel::engine_enums::CommandType::ExpireMarket,
                    hypercall_types::serialize_to_wire_bytes(command),
                    "ExpireMarket",
                ))
            }
            crate::rsm::apply::EngineCommand::TickExpiry { now_ms, context }
                if !context.due_expiries.is_empty() || !context.pending_settlements.is_empty() =>
            {
                Some((
                    hypercall_db_diesel::engine_enums::CommandType::TickExpiry,
                    hypercall_types::serialize_to_wire_bytes(&(*now_ms, context)),
                    "TickExpiry",
                ))
            }
            _ => None,
        }
    }

    async fn journal_replay_owned_expiry_command(
        &self,
        env: &crate::rsm::apply::CommandEnvelope,
        balance_updates: &[hypercall_types::BalanceUpdate],
    ) {
        use crate::journal::JournalMessage;

        if let Some(ref batch_sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            let Some((entry, command_label)) = Self::replay_owned_expiry_journal_entry(
                env,
                balance_updates,
                hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::now_v7()),
                Some(ack_tx),
            ) else {
                return;
            };

            if let Err(error) = batch_sender.send(JournalMessage::Entry(entry)).await {
                panic!(
                    "JOURNAL_FATAL: failed to enqueue {} for replay journal: {}",
                    command_label, error
                );
            }
            if ack_rx.await.is_err() {
                panic!(
                    "JOURNAL_FATAL: replay journal commit ack dropped for {}",
                    command_label
                );
            }
            return;
        }

        let Some((entry, command_label)) = Self::replay_owned_expiry_journal_entry(
            env,
            balance_updates,
            hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::now_v7()),
            None,
        ) else {
            return;
        };

        let Some(ref journal_writer) = self.journal_writer else {
            if self.ctx.db.is_some() {
                panic!("JOURNAL_FATAL: no journal configured for {}", command_label);
            }
            return;
        };
        journal_writer
            .append_transition_with_fill_side_effects(
                entry.received_ts_ms,
                &entry.command_data,
                None,
                None,
                &Default::default(),
                &Default::default(),
                0,
                &[],
                &[],
                &entry.balance_updates,
                entry.request_uuid,
                entry.command_type_enum,
            )
            .unwrap_or_else(|error| {
                panic!(
                    "JOURNAL_FATAL: failed to journal {} for replay: {}",
                    command_label, error
                )
            });
    }

    fn replay_owned_expiry_journal_entry(
        env: &crate::rsm::apply::CommandEnvelope,
        balance_updates: &[hypercall_types::BalanceUpdate],
        request_uuid: hypercall_db_diesel::engine_enums::DbUuid,
        commit_ack: Option<tokio::sync::oneshot::Sender<()>>,
    ) -> Option<(crate::journal::JournalEntry, &'static str)> {
        let (command_type_enum, command_data, command_label) =
            Self::replay_owned_expiry_command_payload(env)?;
        #[cfg(feature = "rsm-state")]
        let command_identity_hash = env.command.identity_hash();

        Some((
            crate::journal::JournalEntry {
                received_ts_ms: env.received_ts_ms,
                command_data,
                response_data: None,
                order_id: None,
                pre_digest: Default::default(),
                post_digest: Default::default(),
                duration_ms: 0,
                events: Vec::new(),
                outbox_appends: Vec::new(),
                fill_side_effects: Vec::new(),
                cash_withdrawal_side_effect: None,
                balance_updates: balance_updates.to_vec(),
                created_at: Instant::now(),
                commit_ack,
                request_uuid,
                command_type_enum: Some(command_type_enum),
                #[cfg(feature = "rsm-state")]
                command_identity_hash,
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: None,
            },
            command_label,
        ))
    }

    fn validate_tick_expiry_nats_payload(
        &self,
        now_ms: u64,
        context: &crate::rsm::apply::TickExpiryContext,
    ) -> Result<(), EngineError> {
        if self.nats_publisher.is_none() {
            return Ok(());
        }

        let command_data = hypercall_types::serialize_to_wire_bytes(&(now_ms, context));
        let total_payload_len = crate::nats::COMMAND_PAYLOAD_PREFIX_LEN + command_data.len();
        if total_payload_len > super::MAX_EXPIRY_NATS_PAYLOAD_BYTES {
            return Err(EngineError::Internal(format!(
                "TickExpiry NATS payload is {} bytes, exceeding {} byte limit. due_groups={}, pending_groups={}, settlement_prices={}, margin_modes={}",
                total_payload_len,
                super::MAX_EXPIRY_NATS_PAYLOAD_BYTES,
                context.due_expiries.len(),
                context.pending_settlements.len(),
                context.settlement_prices.len(),
                context.margin_modes.len()
            )));
        }

        Ok(())
    }

    fn validate_manual_market_expiry_nats_payload(
        &self,
        command: &crate::rsm::apply::MarketActionCommand,
    ) -> Result<(), EngineError> {
        if self.nats_publisher.is_none() || command.expiry_context.is_none() {
            return Ok(());
        }

        let command_data = hypercall_types::serialize_to_wire_bytes(command);
        let total_payload_len = crate::nats::COMMAND_PAYLOAD_PREFIX_LEN + command_data.len();
        if total_payload_len > super::MAX_EXPIRY_NATS_PAYLOAD_BYTES {
            let context = command.expiry_context.as_ref().expect("checked above");
            return Err(EngineError::Internal(format!(
                "ExpireMarket NATS payload is {} bytes, exceeding {} byte limit for {}. due_groups={}, settlement_prices={}, margin_modes={}",
                total_payload_len,
                super::MAX_EXPIRY_NATS_PAYLOAD_BYTES,
                command.message.market.symbol,
                context.due_expiries.len(),
                context.settlement_prices.len(),
                context.margin_modes.len()
            )));
        }

        Ok(())
    }

    async fn prepare_manual_market_expiry_context(
        &self,
        message: &MarketActionMessage,
    ) -> Result<crate::rsm::apply::TickExpiryContext, EngineError> {
        let expiry_ts = crate::rsm::margin_manager::expiry_date_to_timestamp(
            &message.market.underlying,
            message.market.expiry,
        );
        let settlement_price = self
            .expiry_manager
            .get_settlement_price(&self.ctx.deps, &message.market.underlying, expiry_ts as i64)
            .await;
        let settlement_prices = match settlement_price {
            Some(price) => {
                vec![crate::rsm::apply::TickExpirySettlementPrice {
                    underlying: message.market.underlying.clone(),
                    expiry_ts: expiry_ts as i64,
                    price,
                }]
            }
            None => Vec::new(),
        };

        let mut margin_modes = Vec::new();
        if !settlement_prices.is_empty() {
            for ((wallet, symbol), position) in &self.ctx.engine_positions {
                if symbol != &message.market.symbol
                    || position.quantity == rust_decimal::Decimal::ZERO
                {
                    continue;
                }
                let margin_mode = self
                    .ctx
                    .deps
                    .wallet_margin_modes
                    .get(wallet)
                    .copied()
                    .ok_or_else(|| {
                        EngineError::Internal(format!(
                            "Missing margin mode for {} while preparing manual expiry for {}",
                            wallet, message.market.symbol
                        ))
                    })?;
                margin_modes.push(crate::rsm::apply::TickExpiryWalletMarginMode {
                    wallet: *wallet,
                    margin_mode,
                    pm_settlement_required: margin_mode
                        == hypercall_margin::margin_mode::MarginMode::Portfolio
                        && self.ctx.deps.portfolio_margin_pool_enabled
                        && self
                            .ctx
                            .deps
                            .portfolio_margin_settlement_allowlist
                            .contains(wallet),
                });
            }
        }
        margin_modes.sort_by(|left, right| left.wallet.as_bytes().cmp(right.wallet.as_bytes()));
        margin_modes.dedup_by_key(|mode| mode.wallet);

        let due_expiries = vec![crate::rsm::apply::TickExpiryDueGroup {
            expiry_ts: expiry_ts as i64,
            symbols: vec![message.market.symbol.clone()],
        }];
        let pending_settlements = Vec::new();
        let pm_settlements = self
            .expiry_manager
            .prepare_pm_settlements(
                &due_expiries,
                &pending_settlements,
                &settlement_prices,
                &margin_modes,
                message.timestamp,
                &self.ctx,
                &self.margin_manager,
            )
            .map_err(EngineError::Internal)?;

        Ok(crate::rsm::apply::TickExpiryContext {
            due_expiries,
            pending_settlements,
            settlement_prices,
            margin_modes,
            pm_settlements,
        })
    }

    async fn apply_expiry_effects_and_events(
        &mut self,
        output: &crate::rsm::apply::ApplyOutput,
        req_id: &str,
    ) -> Result<(), EngineError> {
        for effect in &output.expiry_effects {
            self.apply_expiry_effect(effect)?;
        }
        self.apply_pm_settlement_projection_effects_sync(&output.pm_settlement_effects, req_id)
            .map_err(EngineError::Internal)?;
        self.apply_replayed_events_sync(&output.events, req_id)
            .await;
        Ok(())
    }

    async fn publish_tick_expiry_balance_updates(
        &self,
        output: &crate::rsm::apply::ApplyOutput,
        context: &'static str,
    ) {
        if output.balance_updates.is_empty() {
            return;
        }

        debug!(
            balance_update_count = output.balance_updates.len(),
            context, "Publishing TickExpiry settlement balance updates"
        );
        self.publish_balance_updates_to_nats(&output.balance_updates)
            .await;
    }

    async fn apply_market_effects(
        &self,
        output: &crate::rsm::apply::ApplyOutput,
    ) -> Result<(), EngineError> {
        for effect in &output.market_effects {
            match effect {
                crate::rsm::apply::MarketEffect::SaveMarketAndInstrument {
                    underlying,
                    expiry,
                    instrument,
                } => {
                    if let Some(handler) = self.ctx.db.as_ref() {
                        if let Err(error) =
                            handler.save_market_and_instrument_sync(underlying, *expiry, instrument)
                        {
                            panic!(
                                "CRITICAL_FAILURE: Failed to persist market {} to database: {}. \
                                 Market will be lost on restart. Restart required.",
                                instrument.id, error
                            );
                        }
                    }
                }
                crate::rsm::apply::MarketEffect::DeleteMarketAndInstrument { symbol } => {
                    if let Some(handler) = self.ctx.db.as_ref() {
                        if let Err(error) = handler.delete_market_and_instrument_sync(symbol) {
                            panic!(
                                "CRITICAL_FAILURE: Failed to delete market {} from database: {}. \
                                 In-memory and persisted state would diverge.",
                                symbol, error
                            );
                        }
                    }
                }
                crate::rsm::apply::MarketEffect::RegisterSettlement {
                    underlying,
                    symbol,
                    expiry_ts,
                    twap_window_seconds,
                } => {
                    let Some(oracle) = self.ctx.deps.mark_price_oracles.get(underlying) else {
                        warn!(
                            "Missing mark price oracle for {} while registering settlement for {} expiry {} window {}s",
                            underlying, symbol, expiry_ts, twap_window_seconds
                        );
                        continue;
                    };
                    oracle
                        .register_settlement(*expiry_ts, *twap_window_seconds)
                        .await;
                    debug!(
                        "Registered TWAP settlement for {} at expiry {}",
                        symbol, expiry_ts
                    );
                }
            }
        }
        Ok(())
    }

    pub(crate) fn apply_pm_settlement_projection_effects_sync(
        &self,
        effects: &[crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect],
        _request_id: &str,
    ) -> Result<(), String> {
        if effects.is_empty() {
            return Ok(());
        }
        let handler = self
            .ctx
            .db
            .as_ref()
            .ok_or_else(|| "PM settlement projection write requires a database".to_string())?;
        let writes = Self::pm_settlement_projection_writes_from_effects(effects)?;
        handler
            .apply_pm_settlement_projection_writes_sync(&writes)
            .map_err(|error| format!("PM settlement projection write failed: {error}"))
    }

    pub(crate) fn pm_settlement_projection_writes_from_effects(
        effects: &[crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect],
    ) -> Result<Vec<hypercall_db::PmSettlementProjectionWrite>, String> {
        let mut writes = Vec::with_capacity(effects.len());
        for effect in effects {
            match effect {
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::PoolUpsert(pool_state) => {
                    let capacity = pool_state.pool_available_usdc
                        + pool_state.active_timing_bridge_usdc
                        + pool_state.active_settlement_debt_usdc;
                    let config = pool_state.config.as_ref();
                    let updated_at_ms = i64::try_from(pool_state.updated_at_ms)
                        .map_err(|_| "PM settlement pool projection timestamp exceeds i64".to_string())?;
                    writes.push(hypercall_db::PmSettlementProjectionWrite::Pool(
                        hypercall_db::PmSettlementPoolProjectionWrite {
                            underlying: pool_state.underlying.clone(),
                            config_version: Self::pm_projection_version_i32(
                                "config_version",
                                pool_state.config_version,
                            )?,
                            policy_version: Self::pm_projection_version_i32(
                                "policy_version",
                                pool_state.policy_version,
                            )?,
                            pool_available_usdc: pool_state.pool_available_usdc,
                            pool_target_usdc: pool_state.pool_target_usdc,
                            pool_capacity_usdc: capacity,
                            pool_utilization: pool_state.utilization,
                            active_timing_bridge_usdc: pool_state.active_timing_bridge_usdc,
                            active_settlement_debt_usdc: pool_state.active_settlement_debt_usdc,
                            target_short_oi_notional_multiplier: config
                                .map(|c| c.target_short_oi_notional_multiplier),
                            utilization_kink: config.map(|c| c.utilization_kink),
                            apr_at_kink: config.map(|c| c.apr_at_kink),
                            max_apr: config.map(|c| c.max_apr),
                            normal_utilization_cap: config.map(|c| c.normal_utilization_cap),
                            crisis_utilization_cap: config.map(|c| c.crisis_utilization_cap),
                            bridge_window_ms: config.map(|c| c.bridge_window_ms),
                            projection_seq: updated_at_ms,
                            updated_at_ms,
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::AccountUpsert(account) => {
                    let updated_at_ms = i64::try_from(account.updated_at_ms)
                        .map_err(|_| "PM settlement account projection timestamp exceeds i64".to_string())?;
                    writes.push(hypercall_db::PmSettlementProjectionWrite::Account(
                        hypercall_db::PmSettlementAccountProjectionWrite {
                            wallet: account.wallet,
                            underlying: account.underlying.clone(),
                            status: format!("{:?}", account.status),
                            timing_bridge_principal_usdc: account.bridge_principal_usdc,
                            settlement_debt_principal_usdc: account.debt_principal_usdc,
                            accrued_interest_usdc: account.accrued_interest_usdc,
                            interest_cursor_ms: account.last_interest_accrual_ms,
                            bridge_deadline_ms: account.bridge_deadline_ms,
                            active_recovery_plan_id: account.active_recovery_plan_id.clone(),
                            policy_version: Self::pm_projection_version_i32(
                                "policy_version",
                                account.policy_version,
                            )?,
                            projection_seq: updated_at_ms,
                            updated_at_ms,
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::InterestEvent(event) => {
                    writes.push(hypercall_db::PmSettlementProjectionWrite::InterestEvent(
                        hypercall_db::PmSettlementInterestEventProjectionWrite {
                            request_id: event.request_id,
                            wallet: event.wallet,
                            underlying: event.underlying.clone(),
                            from_ms: event.accrual_start_ms,
                            to_ms: event.accrual_end_ms,
                            utilization: event.utilization,
                            apr: event.apr,
                            interest_usdc: event.interest_amount_usdc,
                            policy_version: Self::pm_projection_version_i32(
                                "policy_version",
                                event.policy_version,
                            )?,
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::RepaymentEvent(event) => {
                    writes.push(hypercall_db::PmSettlementProjectionWrite::RepaymentEvent(
                        hypercall_db::PmSettlementRepaymentEventProjectionWrite {
                            request_id: event.request_id,
                            wallet: event.wallet,
                            underlying: event.underlying.clone(),
                            amount_usdc: event.amount_usdc,
                            interest_paid_usdc: event.interest_paid_usdc,
                            principal_paid_usdc: event.principal_paid_usdc,
                            reason: event.reason.clone(),
                            source_event_id: event.source_event_id.clone(),
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::VaultDepositUpsert(deposit) => {
                    let projection_seq = i64::try_from(deposit.updated_at_ms)
                        .map_err(|_| "PM vault deposit projection timestamp exceeds i64".to_string())?;
                    let created_at_ms = i64::try_from(deposit.created_at_ms)
                        .map_err(|_| "PM vault deposit created_at timestamp exceeds i64".to_string())?;
                    let chain_id = i64::try_from(deposit.chain_id)
                        .map_err(|_| "PM vault deposit chain_id exceeds i64".to_string())?;
                    let log_index = i32::try_from(deposit.log_index)
                        .map_err(|_| "PM vault deposit log_index exceeds i32".to_string())?;
                    writes.push(hypercall_db::PmSettlementProjectionWrite::VaultDeposit(
                        hypercall_db::PmVaultDepositProjectionWrite {
                            deposit_id: deposit.deposit_id,
                            depositor: deposit.depositor,
                            underlying: deposit.underlying.clone(),
                            principal_usdc: deposit.principal_usdc,
                            remaining_usdc: deposit.remaining_usdc,
                            withdrawn_usdc: deposit.withdrawn_usdc,
                            reserved_withdrawal_usdc: deposit.reserved_withdrawal_usdc,
                            chain_id,
                            source_contract_address: deposit.source_contract_address,
                            tx_hash: deposit.tx_hash.clone(),
                            log_index,
                            max_listed_expiry_ts_ms: deposit.max_listed_expiry_ts_ms,
                            settlement_grace_ms: deposit.settlement_grace_ms,
                            lock_until_ms: deposit.lock_until_ms,
                            status: match deposit.status {
                                crate::rsm::portfolio_margin::settlement_state::PmVaultDepositStatus::Active => {
                                    hypercall_db::PmVaultDepositProjectionStatus::Active
                                }
                                crate::rsm::portfolio_margin::settlement_state::PmVaultDepositStatus::PartiallyReserved => {
                                    hypercall_db::PmVaultDepositProjectionStatus::PartiallyReserved
                                }
                                crate::rsm::portfolio_margin::settlement_state::PmVaultDepositStatus::Reserved => {
                                    hypercall_db::PmVaultDepositProjectionStatus::Reserved
                                }
                            },
                            projection_seq,
                            created_at_ms,
                            updated_at_ms: projection_seq,
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::VaultWithdrawalUpsert(withdrawal) => {
                    let projection_seq = i64::try_from(withdrawal.updated_at_ms)
                        .map_err(|_| "PM vault withdrawal projection timestamp exceeds i64".to_string())?;
                    let requested_at_ms = i64::try_from(withdrawal.requested_at_ms)
                        .map_err(|_| "PM vault withdrawal requested_at timestamp exceeds i64".to_string())?;
                    writes.push(hypercall_db::PmSettlementProjectionWrite::VaultWithdrawal(
                        hypercall_db::PmVaultWithdrawalProjectionWrite {
                            withdrawal_id: withdrawal.withdrawal_id,
                            deposit_id: withdrawal.deposit_id,
                            depositor: withdrawal.depositor,
                            underlying: withdrawal.underlying.clone(),
                            amount_usdc: withdrawal.amount_usdc,
                            lock_until_ms: withdrawal.lock_until_ms,
                            status: match withdrawal.status {
                                crate::rsm::portfolio_margin::settlement_state::PmVaultWithdrawalStatus::Reserved => {
                                    hypercall_db::PmVaultWithdrawalProjectionStatus::Reserved
                                }
                            },
                            projection_seq,
                            requested_at_ms,
                            updated_at_ms: projection_seq,
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::EventUpsert(event) => {
                    let event_key = format!(
                        "{}:{}:{}:{}:{}",
                        event.event_key.wallet,
                        event.event_key.market_id,
                        event.event_key.expiry_ts_ms,
                        event.event_key.margin_mode,
                        event.event_key.settlement_event_sequence
                    );
                    writes.push(hypercall_db::PmSettlementProjectionWrite::Event(
                        hypercall_db::PmSettlementEventProjectionWrite {
                            event_key,
                            wallet: event.event_key.wallet,
                            underlying: event.underlying.clone(),
                            event_type: event.status.clone(),
                            amount_usdc: event.amount_usdc,
                            request_id: event.request_id,
                            input_digest: event.input_digest.clone(),
                        },
                    ));
                }
                crate::rsm::portfolio_margin::settlement_state::PmSettlementProjectionEffect::RecoveryPlanUpsert(plan) => {
                    let updated_at_ms = i64::try_from(plan.updated_at_ms)
                        .map_err(|_| "PM recovery plan projection timestamp exceeds i64".to_string())?;
                    let mut actions = Vec::with_capacity(plan.actions.len());
                    for action in &plan.actions {
                        let action_updated_at_ms = i64::try_from(action.updated_at_ms)
                            .map_err(|_| "PM recovery action projection timestamp exceeds i64".to_string())?;
                        actions.push(hypercall_db::PmRecoveryActionProjectionWrite {
                            plan_id: plan.plan_id.clone(),
                            action_id: format!("{}:{}", plan.plan_id, action.action_index),
                            action_type: action.action_type.clone(),
                            status: action.status.clone(),
                            target: action.target.clone(),
                            attempt: Self::pm_projection_version_i32("attempt", action.attempt)?,
                            external_id: action.submitted_external_id.clone(),
                            external_kind: action.external_kind.clone(),
                            result_external_id: action.result_external_id.clone(),
                            result: action.result.clone(),
                            expected_usdc_recovered: action.expected_usdc_recovered,
                            expected_obligation_reduced: action.expected_obligation_reduced,
                            expected_impact_usdc: action.expected_impact_usdc,
                            recovered_usdc: action.recovered_usdc,
                            liability_reduction_usdc: action.liability_reduction_usdc,
                            projection_seq: action_updated_at_ms,
                            updated_at_ms: action_updated_at_ms,
                        });
                    }
                    writes.push(hypercall_db::PmSettlementProjectionWrite::RecoveryPlan(
                        hypercall_db::PmRecoveryPlanProjectionWrite {
                            plan_id: plan.plan_id.clone(),
                            wallet: plan.wallet,
                            underlying: plan.underlying.clone(),
                            status: plan.status.clone(),
                            trigger: plan.trigger.clone(),
                            reason: plan.reason.clone(),
                            policy_version: Self::pm_projection_version_i32(
                                "policy_version",
                                plan.policy_version,
                            )?,
                            recovery_priority_version: Self::pm_projection_version_i32(
                                "recovery_priority_version",
                                plan.recovery_priority_version,
                            )?,
                            target_reduction_usdc: plan.target_reduction_usdc,
                            expected_usdc_recovered: plan.expected_usdc_recovered,
                            expected_obligation_reduced: plan.expected_obligation_reduced,
                            expected_impact_usdc: plan.expected_impact_usdc,
                            post_plan_utilization: plan.post_plan_utilization,
                            projection_seq: updated_at_ms,
                            updated_at_ms,
                            actions,
                        },
                    ));
                }
            }
        }
        Ok(writes)
    }

    fn pm_projection_version_i32(field: &str, value: u32) -> Result<i32, String> {
        i32::try_from(value).map_err(|_| {
            format!("PM settlement projection {field} {value} exceeds signed 32-bit integer range")
        })
    }

    pub(crate) fn apply_expiry_effect(
        &mut self,
        effect: &crate::rsm::apply::ExpiryEffect,
    ) -> Result<(), EngineError> {
        match effect {
            crate::rsm::apply::ExpiryEffect::UpdateInstrumentStatus { symbols, status } => {
                let Some(handler) = self.ctx.db.as_ref() else {
                    panic!(
                        "CRITICAL_FAILURE: Missing database handler for instrument status update {} on {:?}. \
                         Engine state has advanced without a durable expiry sink. Restart required.",
                        status, symbols
                    );
                };
                let update_result = if status == "EXPIRED_PENDING_PRICE" {
                    handler.transition_active_instruments_to_expired_pending_sync(symbols)
                } else {
                    handler.update_instrument_status_sync(symbols, status)
                };
                if let Err(error) = update_result {
                    panic!(
                        "CRITICAL_FAILURE: Failed to update instrument status to {} for {:?}: {}. \
                         Memory state and database are now inconsistent. Restart required.",
                        status, symbols, error
                    );
                }
                if status == "SETTLED" {
                    for symbol in symbols {
                        match crate::shared::ParsedSymbol::from_symbol(symbol) {
                            Ok(parsed) => {
                                crate::observability::record_settlement(&parsed.underlying, true);
                            }
                            Err(error) => {
                                warn!(
                                    "Failed to parse settled symbol {} for settlement metric: {}",
                                    symbol, error
                                );
                            }
                        }
                    }
                }
            }
            crate::rsm::apply::ExpiryEffect::BatchCancelOrdersForSettlement {
                order_ids,
                now_ms,
            } => {
                let Some(handler) = self.ctx.db.as_ref() else {
                    panic!(
                        "CRITICAL_FAILURE: Missing database handler for {} settlement order cancels at {}. \
                         Engine state has advanced without a durable expiry sink. Restart required.",
                        order_ids.len(),
                        now_ms
                    );
                };
                for chunk in order_ids.chunks(5000) {
                    if let Err(error) =
                        handler.batch_cancel_orders_for_settlement_sync(chunk, *now_ms as i64)
                    {
                        panic!(
                            "CRITICAL_FAILURE: Failed to persist {} settlement order cancels: {}. \
                             Memory state and database are now inconsistent. Restart required.",
                            chunk.len(),
                            error
                        );
                    }
                }
            }
            crate::rsm::apply::ExpiryEffect::CancelOrphanedOrdersBySymbols { symbols } => {
                let Some(handler) = self.ctx.db.as_ref() else {
                    panic!(
                        "CRITICAL_FAILURE: Missing database handler for orphaned order cleanup on {:?}. \
                         Engine state has advanced without a durable expiry sink. Restart required.",
                        symbols
                    );
                };
                match handler.cancel_orphaned_orders_by_symbols_sync(symbols) {
                    Ok(0) => {}
                    Ok(n) => {
                        warn!(
                            "Settlement cleanup: cancelled {} orphaned order_infos rows on {} instruments",
                            n,
                            symbols.len()
                        );
                    }
                    Err(error) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed settlement orphan cleanup for {:?}: {}. \
                             Orphaned orders may remain in order_infos. Restart required.",
                            symbols, error
                        );
                    }
                }
            }
            crate::rsm::apply::ExpiryEffect::ApplySettlement(intent) => {
                let Some(handler) = self.ctx.db.as_ref() else {
                    panic!(
                        "CRITICAL_FAILURE: Missing database handler for durable settlement of {}/{} at {}. \
                         Engine state has advanced without a settlement sink. Restart required.",
                        intent.wallet, intent.symbol, intent.event_ts_ms
                    );
                };
                let outcome = match handler.try_apply_settlement_sync(
                    &intent.wallet,
                    &intent.symbol,
                    intent.position_size,
                    intent.settlement_price,
                    intent.settlement_value,
                    intent.margin_mode,
                    intent.event_ts_ms,
                    intent.settlement_entry_price,
                    intent.cost_basis,
                    intent.net_pnl,
                ) {
                    Ok(outcome) => outcome,
                    Err(error) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed to apply durable settlement for {}/{}: {}. \
                             Engine state has advanced but durable settlement is unknown. Restart required.",
                            intent.wallet, intent.symbol, error
                        );
                    }
                };
                self.reconcile_settlement_balance(intent, outcome);
            }
        }

        Ok(())
    }

    pub(crate) fn apply_standby_expiry_effect(
        &mut self,
        effect: &crate::rsm::apply::ExpiryEffect,
    ) -> Result<(), EngineError> {
        match effect {
            crate::rsm::apply::ExpiryEffect::ApplySettlement(intent) => {
                let Some(handler) = self.ctx.db.as_ref() else {
                    panic!(
                        "CRITICAL_FAILURE: Missing database handler for replayed settlement of {}/{} at {}. \
                         Standby memory state has advanced without a durable settlement source.",
                        intent.wallet, intent.symbol, intent.event_ts_ms
                    );
                };
                let outcome = match Self::observe_standby_settlement_with_retry(handler, intent) {
                    Ok(outcome) => outcome,
                    Err(error) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed to observe durable replayed settlement for {}/{}: {}. \
                             Standby memory state has advanced but durable settlement is unknown.",
                            intent.wallet, intent.symbol, error
                        );
                    }
                };
                self.reconcile_settlement_balance(intent, outcome);
            }
            crate::rsm::apply::ExpiryEffect::UpdateInstrumentStatus { symbols, status } => {
                debug!(
                    status,
                    symbols = ?symbols,
                    "Skipping primary-applied instrument status expiry effect during standby replay"
                );
            }
            crate::rsm::apply::ExpiryEffect::BatchCancelOrdersForSettlement {
                order_ids,
                now_ms,
            } => {
                debug!(
                    order_count = order_ids.len(),
                    now_ms,
                    "Skipping primary-applied settlement order cancel effect during standby replay"
                );
            }
            crate::rsm::apply::ExpiryEffect::CancelOrphanedOrdersBySymbols { symbols } => {
                debug!(
                    symbols = ?symbols,
                    "Skipping primary-applied orphan cleanup expiry effect during standby replay"
                );
            }
        }

        Ok(())
    }

    /// Apply TickExpiry runtime effects after deterministic command replay.
    ///
    /// The replayed command remains the authority for in-memory balance state.
    /// This recovery-only path only catches durable projections up to the WAL
    /// command: it writes idempotent settlement evidence, persists idempotent
    /// status/cancel projections, and then checks that replayed engine cash
    /// matches the durable balance. It must never set `balance_ledger` from
    /// durable accounting.
    pub(crate) fn apply_replayed_expiry_effects(
        &mut self,
        effects: &[crate::rsm::apply::ExpiryEffect],
    ) -> Result<(), EngineError> {
        for effect in effects {
            match effect {
                crate::rsm::apply::ExpiryEffect::ApplySettlement(intent) => {
                    let Some(handler) = self.ctx.db.as_ref() else {
                        panic!(
                            "CRITICAL_FAILURE: Missing database handler for replayed settlement of {}/{} at {}. \
                             Recovery cannot persist settlement cash replay.",
                            intent.wallet, intent.symbol, intent.event_ts_ms
                        );
                    };
                    let outcome = match handler.try_apply_settlement_sync(
                        &intent.wallet,
                        &intent.symbol,
                        intent.position_size,
                        intent.settlement_price,
                        intent.settlement_value,
                        intent.margin_mode,
                        intent.event_ts_ms,
                        intent.settlement_entry_price,
                        intent.cost_basis,
                        intent.net_pnl,
                    ) {
                        Ok(outcome) => outcome,
                        Err(error) => {
                            panic!(
                                "CRITICAL_FAILURE: Failed to apply durable replayed settlement for {}/{}: {}. \
                                 Recovery cannot persist settlement cash replay.",
                                intent.wallet, intent.symbol, error
                            );
                        }
                    };
                    debug!(
                        wallet = %intent.wallet,
                        symbol = %intent.symbol,
                        newly_persisted = outcome.newly_persisted,
                        "Applied replayed TickExpiry settlement effect"
                    );
                }
                crate::rsm::apply::ExpiryEffect::UpdateInstrumentStatus { symbols, status } => {
                    let Some(handler) = self.ctx.db.as_ref() else {
                        panic!(
                            "CRITICAL_FAILURE: Missing database handler for replayed instrument status update {} on {:?}. \
                             Recovery cannot persist expiry status projection.",
                            status, symbols
                        );
                    };
                    let update_result = if status == "EXPIRED_PENDING_PRICE" {
                        handler.transition_active_instruments_to_expired_pending_sync(symbols)
                    } else {
                        handler.update_instrument_status_sync(symbols, status)
                    };
                    if let Err(error) = update_result {
                        panic!(
                            "CRITICAL_FAILURE: Failed to persist replayed instrument status {} for {:?}: {}. \
                             Recovery cannot persist expiry status projection.",
                            status, symbols, error
                        );
                    }
                }
                crate::rsm::apply::ExpiryEffect::BatchCancelOrdersForSettlement {
                    order_ids,
                    now_ms,
                } => {
                    let Some(handler) = self.ctx.db.as_ref() else {
                        panic!(
                            "CRITICAL_FAILURE: Missing database handler for replayed settlement order cancels at {}. \
                             Recovery cannot persist expiry order projection.",
                            now_ms
                        );
                    };
                    for chunk in order_ids.chunks(5000) {
                        if let Err(error) =
                            handler.batch_cancel_orders_for_settlement_sync(chunk, *now_ms as i64)
                        {
                            panic!(
                                "CRITICAL_FAILURE: Failed to persist {} replayed settlement order cancels: {}. \
                                 Recovery cannot persist expiry order projection.",
                                chunk.len(),
                                error
                            );
                        }
                    }
                }
                crate::rsm::apply::ExpiryEffect::CancelOrphanedOrdersBySymbols { symbols } => {
                    let Some(handler) = self.ctx.db.as_ref() else {
                        panic!(
                            "CRITICAL_FAILURE: Missing database handler for replayed orphaned order cleanup on {:?}. \
                             Recovery cannot persist expiry order projection.",
                            symbols
                        );
                    };
                    if let Err(error) = handler.cancel_orphaned_orders_by_symbols_sync(symbols) {
                        panic!(
                            "CRITICAL_FAILURE: Failed replayed settlement orphan cleanup for {:?}: {}. \
                             Recovery cannot persist expiry order projection.",
                            symbols, error
                        );
                    }
                }
            }
        }

        Ok(())
    }

    fn observe_standby_settlement_with_retry(
        handler: &hypercall_db_diesel::DatabaseHandler,
        intent: &crate::rsm::apply::ExpirySettlementIntent,
    ) -> Result<hypercall_db::SettlementResult, anyhow::Error> {
        let mut backoff_ms = STANDBY_SETTLEMENT_OBSERVE_INITIAL_BACKOFF_MS;
        let mut last_error = None;

        for attempt in 1..=STANDBY_SETTLEMENT_OBSERVE_ATTEMPTS {
            match handler.observe_applied_settlement_sync(
                &intent.wallet,
                &intent.symbol,
                intent.position_size,
                intent.settlement_price,
                intent.settlement_value,
                intent.margin_mode,
                intent.settlement_entry_price,
                intent.cost_basis,
                intent.net_pnl,
            ) {
                Ok(outcome) => return Ok(outcome),
                Err(error) => {
                    if attempt == STANDBY_SETTLEMENT_OBSERVE_ATTEMPTS {
                        last_error = Some(error);
                        break;
                    }
                    warn!(
                        wallet = %intent.wallet,
                        symbol = %intent.symbol,
                        attempt,
                        max_attempts = STANDBY_SETTLEMENT_OBSERVE_ATTEMPTS,
                        error = %error,
                        "Standby settlement observation missed durable row; retrying"
                    );
                    last_error = Some(error);
                    std::thread::sleep(Duration::from_millis(backoff_ms));
                    backoff_ms = (backoff_ms * 2).min(STANDBY_SETTLEMENT_OBSERVE_MAX_BACKOFF_MS);
                }
            }
        }

        Err(last_error.expect("standby settlement observation must record a final error"))
    }

    fn reconcile_settlement_balance(
        &mut self,
        intent: &crate::rsm::apply::ExpirySettlementIntent,
        outcome: hypercall_db::SettlementResult,
    ) {
        debug!(
            wallet = %intent.wallet,
            symbol = %intent.symbol,
            balance_after = %self.ctx.balance_ledger.balance(&intent.wallet),
            newly_persisted = outcome.newly_persisted,
            "Observed durable TickExpiry settlement outcome without mutating engine cash"
        );
    }

    /// Flush the journal batcher, ensuring all pending entries are committed to DB.
    async fn flush_journal(&self) {
        if let Some(ref sender) = self.journal_batch_sender {
            let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
            if let Err(e) = sender
                .send(crate::journal::JournalMessage::Flush(ack_tx))
                .await
            {
                error!("Failed to send flush to journal batcher: {}", e);
                return;
            }
            match ack_rx.await {
                Ok(()) => {
                    debug!("Journal batcher flush completed");
                }
                Err(_) => {
                    error!("Journal batcher flush ack channel dropped");
                }
            }
        }
    }

    async fn hydrate_runtime_state_from_dependencies(
        &mut self,
        context: &'static str,
        hydrate_market_data: bool,
    ) {
        // Seed engine_positions from PortfolioService on startup.
        // After startup, positions are updated from fills inside apply().
        if let Some(ref portfolio_service) = self.ctx.deps.portfolio_service {
            let all = portfolio_service.all_portfolios().await;
            let previous_count = self.ctx.engine_positions.len();
            self.ctx.engine_positions.clear();

            let mut count = 0usize;
            for (wallet, balance) in all {
                for (symbol, pos) in balance.positions {
                    if pos.amount != rust_decimal::Decimal::ZERO {
                        self.ctx.engine_positions.insert(
                            (wallet, symbol),
                            crate::rsm::engine_deps::EnginePosition {
                                quantity: pos.amount,
                                entry_price: pos.entry_price,
                            },
                        );
                        count += 1;
                    }
                }
            }
            if count > 0 || previous_count > 0 {
                info!(
                    context,
                    previous_count, count, "Reconciled engine positions from PortfolioService"
                );
            }
        }

        if hydrate_market_data {
            self.ingest_price_updates_without_side_effects().await;
            self.ingest_iv_updates_without_side_effects().await;
        }
    }

    pub(crate) async fn hydrate_standby_base_state(&mut self) {
        self.hydrate_runtime_state_from_dependencies("standby_startup", !self.runtime_quiesced)
            .await;
        self.publish_snapshot();
        info!(
            positions = self.ctx.engine_positions.len(),
            cash_wallets = self.ctx.balance_ledger.len(),
            spot_prices = self.ctx.spot_prices.len(),
            iv_surfaces = self.ctx.iv_surfaces.len(),
            "Hydrated standby base engine state"
        );
    }

    pub(crate) async fn hydrate_primary_base_state(&mut self) {
        self.hydrate_runtime_state_from_dependencies("startup", false)
            .await;
    }

    pub(crate) async fn apply_startup_replayed_events(&mut self) {
        if self.startup_replayed_events.is_empty() {
            return;
        }

        let batches = std::mem::take(&mut self.startup_replayed_events);
        let event_count = batches
            .iter()
            .map(|(_, events)| events.len())
            .sum::<usize>();
        for (req_id, events) in batches {
            self.apply_startup_replayed_events_sync(&events, &req_id)
                .await;
        }
        info!(
            event_count,
            "Applied startup-replayed TickExpiry event projections"
        );
    }

    /// Persist engine state snapshot to disk for fast restart recovery.
    ///
    /// Reads the current WAL checkpoint and serializes the engine's in-memory
    /// state (orders, positions, balances) so the next startup can skip full
    /// replay from Postgres. Only runs when ENGINE_JOURNAL_WAL_PATH is set.
    pub(super) fn persist_engine_state_snapshot(&self) -> Option<i64> {
        let wal_path_configured = hypercall_journal::checkpoint::wal_path_is_explicitly_configured(
            self.wal_path.as_ref(),
        );
        if !wal_path_configured || self.ctx.db.is_none() {
            return None;
        }

        use crate::rsm::engine_state_snapshot::{snapshot_path_from_wal_path, write_snapshot};
        use crate::rsm::restart_components::{
            EngineRecoveryCapture, PersistentEngineStateComponent,
        };
        use hypercall_journal::checkpoint::{
            checkpoint_path_for, read_checkpoint, wal_path_from_config,
        };
        use hypercall_recovery::RestartStateComponent;

        let wal_path = wal_path_from_config(self.wal_path.as_ref());
        let checkpoint_path = checkpoint_path_for(&wal_path);
        match read_checkpoint(&checkpoint_path) {
            Ok(checkpoint) if checkpoint.last_command_id > 0 => {
                let snapshot_path = snapshot_path_from_wal_path(&wal_path);
                let snapshot = PersistentEngineStateComponent::capture(
                    &EngineRecoveryCapture::for_snapshot(&self.ctx, &checkpoint),
                );
                if let Err(e) = write_snapshot(&snapshot_path, &snapshot) {
                    warn!("Failed to write engine state snapshot: {}", e);
                    None
                } else {
                    if let Some(ref handler) = self.ctx.db {
                        if let Err(e) =
                            handler.update_snapshot_boundary_sync(checkpoint.last_command_id)
                        {
                            warn!("Failed to persist snapshot boundary to Postgres: {}", e);
                            return None;
                        }
                    }
                    info!(
                        "Persisted engine state snapshot: last_command_id={}",
                        checkpoint.last_command_id
                    );
                    Some(checkpoint.last_command_id)
                }
            }
            Ok(_) => {
                // last_command_id == 0: no commands replicated yet, skip snapshot
                None
            }
            Err(e) => {
                panic!(
                    "CRITICAL_FAILURE: failed to read checkpoint for snapshot write ({}): {}",
                    checkpoint_path.display(),
                    e
                );
            }
        }
    }

    fn current_wal_checkpoint_command_id(&self) -> i64 {
        let wal_path_configured = hypercall_journal::checkpoint::wal_path_is_explicitly_configured(
            self.wal_path.as_ref(),
        );
        if !wal_path_configured {
            return self.replay_checkpoint.last_command_id;
        }

        let wal_path = hypercall_journal::checkpoint::wal_path_from_config(self.wal_path.as_ref());
        let checkpoint_path = hypercall_journal::checkpoint::checkpoint_path_for(&wal_path);
        match hypercall_journal::checkpoint::read_checkpoint(&checkpoint_path) {
            Ok(checkpoint) => checkpoint.last_command_id,
            Err(error) => {
                warn!(
                    path = %checkpoint_path.display(),
                    error = %error,
                    "Failed to read current WAL checkpoint for quiesce report"
                );
                self.replay_checkpoint.last_command_id
            }
        }
    }

    async fn apply_runtime_balance_update(
        &mut self,
        env: crate::rsm::apply::CommandEnvelope,
        applied_tx: Option<tokio::sync::oneshot::Sender<Result<(), String>>>,
        wallet: WalletAddress,
        amount: Decimal,
        update_kind: &'static str,
        last_read_snapshot: &mut Instant,
        journal_request_id: String,
        outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
    ) {
        Self::require_external_durable_mutation_uuid(update_kind, &journal_request_id);
        match self.apply(env.clone()) {
            Ok(output) => {
                let output_is_empty = output.is_empty();
                let balance_updates = output.balance_updates;
                if balance_updates.is_empty()
                    && output_is_empty
                    && Self::is_idempotent_empty_balance_update(update_kind)
                {
                    debug!(
                        wallet = %wallet,
                        amount = %amount,
                        update_kind = update_kind,
                        "Observed duplicate balance update with no new balance effects"
                    );
                    if let Some(tx) = applied_tx {
                        let _ = tx.send(Ok(()));
                    }
                    return;
                }
                assert!(
                    !balance_updates.is_empty(),
                    "RUNTIME_INVARIANT: {} applied without canonical balance_updates",
                    update_kind
                );
                self.journal_external_balance_update_command(
                    &env,
                    &journal_request_id,
                    outbox_appends,
                    balance_updates.clone(),
                )
                .await;
                self.publish_to_nats(&env).await;
                self.publish_balance_updates_to_nats(&balance_updates).await;
                self.publish_snapshot();
                *last_read_snapshot = Instant::now();
                if let Some(tx) = applied_tx {
                    let _ = tx.send(Ok(()));
                }
            }
            Err(e) => {
                let err = e.to_string();
                if let Some(tx) = applied_tx {
                    let _ = tx.send(Err(err.clone()));
                }
                panic!(
                    "RUNTIME_INVARIANT: failed to apply {} balance_ledger update for {} amount {} before publish: {}",
                    update_kind, wallet, amount, err
                );
            }
        }
    }

    fn is_idempotent_empty_balance_update(update_kind: &str) -> bool {
        matches!(update_kind, "DepositUpdate" | "LiquidationBonusUpdate")
    }

    /// Start the engine event loop
    pub async fn start(mut self) {
        info!("Unified engine starting with SPAN calculations...");

        self.apply_startup_replayed_events().await;
        self.hydrate_primary_base_state().await;

        // Start MMP automatic eviction if cache is available
        if let Some(ref mmp_cache) = self.ctx.deps.mmp_cache {
            mmp_cache.start();
            info!("MMP cache automatic eviction started");
        }

        // Create expiry check interval (check every minute)
        let mut expiry_interval = tokio::time::interval(Duration::from_secs(60));

        // Create price update interval (feed spot prices into command stream every 2s)
        let mut price_update_interval = tokio::time::interval(Duration::from_secs(2));

        // Hydrate prices before accepting orders so margin checks don't
        // reject during the startup window before the first periodic tick.
        // If the process is starting under a persisted admin drain marker,
        // preserve the snapshot/replay state until /admin/undrain resumes
        // live external ingestion.
        if !self.runtime_quiesced {
            self.ingest_price_updates().await;
            self.ingest_iv_updates().await;
        } else {
            info!("Unified engine starting quiesced; startup market ingestion is paused");
        }

        // Settle any already-expired instruments synchronously before accepting
        // orders. Without this, expired positions linger for up to 60s (the
        // expiry tick interval), causing portfolio cache repricing errors and
        // the engine rejecting orders on expired chains.
        if !self.runtime_quiesced {
            let now_ms = get_timestamp_millis();
            let env = match self.prepare_tick_expiry_env(now_ms).await {
                Ok(env) => env,
                Err(e) => {
                    panic!(
                        "CRITICAL_FAILURE: Startup expiry preparation failed: {}. \
                         Expiry settlement cannot continue without explicit context.",
                        e
                    );
                }
            };
            let journal_env = env.clone();
            match self.apply(env) {
                Ok(output) => {
                    self.journal_replay_owned_expiry_command(&journal_env, &output.balance_updates)
                        .await;
                    if let Err(e) = self
                        .apply_expiry_effects_and_events(&output, "startup-tick-expiry")
                        .await
                    {
                        panic!(
                            "CRITICAL_FAILURE: Startup expiry effects failed: {}. \
                             Expiry runtime effects could not be applied after memory state advanced.",
                            e
                        );
                    }
                    self.publish_tick_expiry_balance_updates(&output, "startup-tick-expiry")
                        .await;
                    info!(
                        "Startup expiry check complete: {} events",
                        output.events.len()
                    );
                }
                Err(e) => {
                    panic!(
                        "CRITICAL_FAILURE: apply() failed for startup TickExpiry: {}. \
                         Expiry state could not be advanced deterministically.",
                        e
                    );
                }
            }
            self.publish_snapshot();
        } else {
            info!("Unified engine starting quiesced; startup expiry check is paused");
        }

        // Spawn periodic snapshot task
        let mut snapshot_interval = tokio::time::interval(self.snapshot_interval);
        let post_startup_reconcile = tokio::time::sleep(self.post_startup_reconcile_delay);
        tokio::pin!(post_startup_reconcile);

        let mut last_read_snapshot = Instant::now();

        // Reuse Instant variables and histogram handles to avoid repeated lookups
        let mut loop_start;
        let mut market_start;
        let mut expiry_start;
        let mut snapshot_start;

        // Get histogram handles once (they're cheap references into the global registry)
        let loop_iteration_hist = histogram!("ht_engine_loop_iteration_seconds");
        let market_request_hist = histogram!("ht_engine_market_request_seconds");
        let expiry_check_hist = histogram!("ht_engine_expiry_check_seconds");
        let snapshot_hist = histogram!("ht_engine_snapshot_seconds");

        loop {
            loop_start = Instant::now();
            select! {
                Some(request) = async {
                    match self.quiesce_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    self.handle_quiesce_request(request).await;
                }

                // Handle incoming orders
                Some(request) = self.order_receiver.recv(), if !self.runtime_quiesced => {
                    DurableJournaling::process_order_journaled(&mut self, request).await;
                    if last_read_snapshot.elapsed() >= self.read_snapshot_interval {
                        self.publish_snapshot();
                        last_read_snapshot = Instant::now();
                    }
                }

                // Handle RFQ execution requests
                Some(request) = async {
                    match self.rfq_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                }, if !self.runtime_quiesced => {
                    self.handle_rfq_execute(request).await;
                    if last_read_snapshot.elapsed() >= self.read_snapshot_interval {
                        self.publish_snapshot();
                        last_read_snapshot = Instant::now();
                    }
                }

                // Handle deposit requests (faucet/admin)
                Some(deposit) = async {
                    match self.deposit_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = deposit.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = deposit.wallet;
                    let amount = deposit.amount;
                    let source_event_hash = deposit.source_event_hash;
                    let journal_request_id = deposit.journal_request_id;
                    let outbox_appends = deposit.outbox_appends;
                    let applied_tx = deposit.applied_tx;
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        deposit.timestamp_ms,
                        crate::rsm::apply::EngineCommand::DepositUpdate {
                            wallet,
                            amount,
                            timestamp_ms: deposit.timestamp_ms,
                            sequence: deposit.sequence,
                            source_event_hash,
                        },
                    );
                    self.apply_runtime_balance_update(
                        env,
                        applied_tx,
                        wallet,
                        amount,
                        "DepositUpdate",
                        &mut last_read_snapshot,
                        journal_request_id,
                        outbox_appends,
                    )
                    .await;
                }

                // Handle option-token deposits observed on-chain.
                Some(deposit) = async {
                    match self.option_deposit_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = deposit.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = deposit.wallet;
                    let symbol = deposit.symbol;
                    let quantity = deposit.quantity;
                    let request_id = deposit.request_id;
                    let timestamp_ms = deposit.timestamp_ms;
                    let applied_tx = deposit.applied_tx;
                    if self
                        .external_option_command_already_journaled(
                            &request_id,
                            "OptionDepositUpdate",
                        )
                        .await
                    {
                        if let Some(tx) = applied_tx {
                            let _ = tx.send(Ok(()));
                        }
                        continue;
                    }
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        timestamp_ms,
                        crate::rsm::apply::EngineCommand::OptionDepositUpdate {
                            request_id: request_id.clone(),
                            wallet,
                            symbol: symbol.clone(),
                            quantity,
                            timestamp_ms,
                        },
                    );
                    match self.apply(env.clone()) {
                        Ok(_) => {
                            self
                                .journal_external_option_position_command(
                                    &env,
                                    &request_id,
                                    hypercall_db_diesel::engine_enums::CommandType::OptionDepositUpdate,
                                    Vec::new(),
                                )
                                .await;
                            if let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache {
                                portfolio_cache
                                    .handle_option_custody_delta(
                                        wallet,
                                        symbol.clone(),
                                        quantity,
                                        None,
                                    )
                                    .await;
                            }
                            self.publish_to_nats(&env).await;
                            self.publish_snapshot();
                            last_read_snapshot = Instant::now();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Ok(()));
                            }
                        }
                        Err(e) => {
                            let err = e.to_string();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Err(err.clone()));
                            }
                            panic!(
                                "RUNTIME_INVARIANT: failed to apply OptionDepositUpdate for {} {} quantity {} before publish: {}",
                                wallet, symbol, quantity, err
                            );
                        }
                    }
                }

                Some(withdrawal) = async {
                    match self.option_withdrawal_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = withdrawal.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = withdrawal.wallet;
                    let account = withdrawal.account;
                    let signer = withdrawal.signer;
                    let rsm_signer = withdrawal.rsm_signer;
                    let symbol = withdrawal.symbol;
                    let quantity = withdrawal.quantity;
                    let nonce = withdrawal.nonce;
                    let action = withdrawal.action;
                    let request_id = withdrawal.request_id;
                    let timestamp_ms = withdrawal.timestamp_ms;
                    let applied_tx = withdrawal.applied_tx;
                    if self
                        .external_option_command_already_journaled(
                            &request_id,
                            "OptionWithdrawalUpdate",
                        )
                        .await
                    {
                        let persisted_status = self
                            .persisted_withdrawal_directive_status(&request_id)
                            .unwrap_or_else(|error| {
                                panic!(
                                    "JOURNAL_FATAL: failed to load persisted OptionWithdrawalUpdate status for {}: {}",
                                    request_id, error
                                )
                            });
                        let (domain_status, delivery_status) = persisted_status.unwrap_or((
                            crate::directive_outbox::DirectiveDomainStatus::PendingChainEffect,
                            crate::directive_outbox::DirectiveDeliveryStatus::Pending,
                        ));
                        let receipt = crate::rsm::unified_engine::OptionWithdrawalApplyReceipt {
                            directive_id: request_id.clone(),
                            domain_status,
                            delivery_status,
                        };
                        if let Some(tx) = applied_tx {
                            let _ = tx.send(Ok(receipt));
                        }
                        continue;
                    }
                    if let Some(err) = self.withdrawal_directive_journal_unavailable_error(
                        "OptionWithdrawalUpdate",
                        &request_id,
                    ) {
                        warn!("{}", err);
                        if let Some(tx) = applied_tx {
                            let _ = tx.send(Err(err));
                        }
                        continue;
                    }
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        timestamp_ms,
                        crate::rsm::apply::EngineCommand::OptionWithdrawalUpdate {
                            request_id: request_id.clone(),
                            wallet,
                            account,
                            signer,
                            rsm_signer,
                            symbol: symbol.clone(),
                            quantity,
                            nonce: Some(nonce),
                            action,
                            timestamp_ms,
                        },
                    );
                    match self.apply(env.clone()) {
                        Ok(output) => {
                            let outbox_appends = output.outbox_appends;
                            let receipt = outbox_appends
                                .first()
                                .map(|append| {
                                    crate::rsm::unified_engine::OptionWithdrawalApplyReceipt {
                                        directive_id: append.directive_id.clone(),
                                        domain_status: append.domain_status.clone(),
                                        delivery_status: append.delivery_status.clone(),
                                    }
                                })
                                .unwrap_or_else(|| {
                                    crate::rsm::unified_engine::OptionWithdrawalApplyReceipt {
                                        directive_id: request_id.clone(),
                                        domain_status: crate::directive_outbox::DirectiveDomainStatus::PendingChainEffect,
                                        delivery_status: crate::directive_outbox::DirectiveDeliveryStatus::Pending,
                                    }
                                });
                            self
                                .journal_external_option_position_command(
                                    &env,
                                    &request_id,
                                    hypercall_db_diesel::engine_enums::CommandType::OptionWithdrawalUpdate,
                                    outbox_appends,
                                )
                                .await;
                            if let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache {
                                portfolio_cache
                                    .handle_option_custody_delta(
                                        wallet,
                                        symbol.clone(),
                                        -quantity,
                                        None,
                                    )
                                    .await;
                            }
                            self.publish_to_nats(&env).await;
                            self.publish_snapshot();
                            last_read_snapshot = Instant::now();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Ok(receipt));
                            }
                        }
                        Err(e) => {
                            let err = e.to_string();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Err(err.clone()));
                            }
                        }
                    }
                }

                // Handle USDC cash withdrawals through engine-owned state.
                Some(withdrawal) = async {
                    match self.cash_withdrawal_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = withdrawal.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = withdrawal.wallet;
                    let account = withdrawal.account;
                    let destination = withdrawal.destination;
                    let signer = withdrawal.signer;
                    let rsm_signer = withdrawal.rsm_signer;
                    let amount = withdrawal.amount;
                    let amount_wei = withdrawal.amount_wei;
                    let nonce = withdrawal.nonce;
                    let request_id = withdrawal.request_id;
                    let timestamp_ms = withdrawal.timestamp_ms;
                    let applied_tx = withdrawal.applied_tx;
                    if self
                        .external_option_command_already_journaled(
                            &request_id,
                            "CashWithdrawalUpdate",
                        )
                        .await
                    {
                        let persisted_status = self
                            .persisted_withdrawal_directive_status(&request_id)
                            .unwrap_or_else(|error| {
                                panic!(
                                    "JOURNAL_FATAL: failed to load persisted CashWithdrawalUpdate status for {}: {}",
                                    request_id, error
                                )
                            });
                        let (domain_status, delivery_status) = persisted_status.unwrap_or((
                            crate::directive_outbox::DirectiveDomainStatus::PendingChainEffect,
                            crate::directive_outbox::DirectiveDeliveryStatus::Pending,
                        ));
                        let receipt = crate::rsm::unified_engine::CashWithdrawalApplyReceipt {
                            directive_id: request_id.clone(),
                            domain_status,
                            delivery_status,
                            balance_after: self.ctx.balance_ledger.balance(&wallet),
                        };
                        if let Some(tx) = applied_tx {
                            let _ = tx.send(Ok(receipt));
                        }
                        continue;
                    }
                    if let Some(err) = self.withdrawal_directive_journal_unavailable_error(
                        "CashWithdrawalUpdate",
                        &request_id,
                    ) {
                        warn!("{}", err);
                        if let Some(tx) = applied_tx {
                            let _ = tx.send(Err(err));
                        }
                        continue;
                    }
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        timestamp_ms,
                        crate::rsm::apply::EngineCommand::CashWithdrawalUpdate {
                            request_id: request_id.clone(),
                            wallet,
                            account,
                            destination,
                            signer,
                            rsm_signer,
                            amount,
                            amount_wei,
                            nonce: Some(nonce),
                            timestamp_ms,
                        },
                    );
                    match self.apply(env.clone()) {
                        Ok(output) => {
                            let balance_after = self.ctx.balance_ledger.balance(&wallet);
                            let outbox_appends = output.outbox_appends;
                            let balance_updates = output.balance_updates;
                            assert!(
                                !balance_updates.is_empty(),
                                "RUNTIME_INVARIANT: CashWithdrawalUpdate applied without canonical balance_updates"
                            );
                            let receipt = outbox_appends
                                .first()
                                .map(|append| {
                                    crate::rsm::unified_engine::CashWithdrawalApplyReceipt {
                                        directive_id: append.directive_id.clone(),
                                        domain_status: append.domain_status.clone(),
                                        delivery_status: append.delivery_status.clone(),
                                        balance_after,
                                    }
                                })
                                .unwrap_or_else(|| {
                                    crate::rsm::unified_engine::CashWithdrawalApplyReceipt {
                                        directive_id: request_id.clone(),
                                        domain_status: crate::directive_outbox::DirectiveDomainStatus::PendingChainEffect,
                                        delivery_status: crate::directive_outbox::DirectiveDeliveryStatus::Pending,
                                        balance_after,
                                    }
                                });
                            self.journal_external_cash_withdrawal_command(
                                &env,
                                &request_id,
                                outbox_appends,
                                wallet,
                                amount,
                                balance_after,
                                timestamp_ms,
                                balance_updates.clone(),
                            )
                            .await;
                            self.publish_to_nats(&env).await;
                            self.publish_balance_updates_to_nats(&balance_updates).await;
                            self.publish_snapshot();
                            last_read_snapshot = Instant::now();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Ok(receipt));
                            }
                        }
                        Err(e) => {
                            let err = e.to_string();
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Err(err));
                            }
                        }
                    }
                }

                // Handle liquidation bonus credits after durable DB persistence.
                Some(bonus) = async {
                    match self.liquidation_bonus_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = bonus.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = bonus.wallet;
                    let amount = bonus.amount;
                    let balance_after = self.ctx.balance_ledger.balance(&wallet) + amount;
                    let request_id = bonus.request_id;
                    let applied_tx = bonus.applied_tx;
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        bonus.timestamp_ms,
                        crate::rsm::apply::EngineCommand::LiquidationBonusUpdate {
                            wallet,
                            amount,
                            balance_after,
                            timestamp_ms: bonus.timestamp_ms,
                            sequence: bonus.sequence,
                        },
                    );
                    self.apply_runtime_balance_update(
                        env,
                        applied_tx,
                        wallet,
                        amount,
                        "LiquidationBonusUpdate",
                        &mut last_read_snapshot,
                        request_id,
                        Vec::new(),
                    ).await;
                }

                // Handle margin-mode changes through apply() so order admission reads
                // the same engine-owned state that is snapshotted and replicated.
                Some(update) = async {
                    match self.margin_mode_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        let _ = update.applied_tx.send(Err("engine is quiesced".to_string()));
                        continue;
                    }
                    let wallet = update.wallet;
                    let margin_mode = update.margin_mode;
                    let applied_tx = update.applied_tx;
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        update.timestamp_ms,
                        crate::rsm::apply::EngineCommand::LegacyTierMarginModeUpdate {
                            wallet,
                            margin_mode,
                        },
                    );
                    let nats_env = env.clone();
                    match self.apply(env) {
                        Ok(output) => {
                            self.publish_to_nats(&nats_env).await;
                            for event in &output.events {
                                self.ctx.deps.emit_event(event);
                            }
                            let _ = applied_tx.send(Ok(()));
                        }
                        Err(e) => {
                            let err = e.to_string();
                            warn!(
                                wallet = %wallet,
                                margin_mode = ?margin_mode,
                                error = %err,
                                "Failed to apply TierUpdate"
                            );
                            let _ = applied_tx.send(Err(err));
                        }
                    }
                }

                // Handle agent authorization changes through apply()
                Some(request) = async {
                    match self.agent_auth_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        let _ = request.applied_tx.send(Err("engine is quiesced".to_string()));
                        continue;
                    }
                    let wallet = request.wallet;
                    let agent = request.agent;
                    let applied_tx = request.applied_tx;
                    let expires_at_ms = request.expires_at_ms;
                    let nonce = request.nonce;
                    let ts = get_timestamp_millis();
                    let command = if request.approve {
                        crate::rsm::apply::EngineCommand::ApproveAgent { wallet, agent, expires_at_ms, nonce, timestamp_ms: ts }
                    } else {
                        crate::rsm::apply::EngineCommand::RevokeAgent { wallet, agent, nonce, timestamp_ms: ts }
                    };
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        ts,
                        command,
                    );
                    if !self.agent_auth_journal_available() {
                        let _ = applied_tx.send(Err(
                            "agent authorization requires engine journaling to be enabled"
                                .to_string(),
                        ));
                        continue;
                    }
                    let nats_env = env.clone();
                    match self.apply(env) {
                        Ok(_) => {
                            self.journal_agent_auth_command(&nats_env).await;
                            self.publish_to_nats(&nats_env).await;
                            self.publish_snapshot();
                            last_read_snapshot = Instant::now();
                            let _ = applied_tx.send(Ok(()));
                        }
                        Err(e) => {
                            let _ = applied_tx.send(Err(e.to_string()));
                        }
                    }
                }

                // Handle nonce-only checks (QP handshakes).
                Some(request) = async {
                    match self.nonce_check_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        let _ = request.applied_tx.send(Err("engine is quiesced".to_string()));
                        continue;
                    }
                    let wallet = request.wallet;
                    let nonce = request.nonce;
                    let applied_tx = request.applied_tx;
                    let ts = get_timestamp_millis();
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        ts,
                        crate::rsm::apply::EngineCommand::NonceAdvance { wallet, nonce, timestamp_ms: ts },
                    );
                    let nats_env = env.clone();
                    match self.apply(env) {
                        Ok(_) => {
                            self.publish_to_nats(&nats_env).await;
                            let _ = applied_tx.send(Ok(()));
                        }
                        Err(e) => {
                            let _ = applied_tx.send(Err(e.to_string()));
                        }
                    }
                }

                // Handle tier updates that affect engine-owned admission state.
                Some(update) = async {
                    match self.tier_update_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        if let Some(tx) = update.applied_tx {
                            let _ = tx.send(Err("engine is quiesced".to_string()));
                        }
                        continue;
                    }
                    let wallet = update.wallet;
                    let tier = update.tier;
                    let trading_limits = update.trading_limits;
                    let margin_mode = update.margin_mode;
                    let applied_tx = update.applied_tx;
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        update.timestamp_ms,
                        crate::rsm::apply::EngineCommand::TierUpdate {
                            wallet,
                            margin_mode,
                            tier,
                            trading_limits,
                        },
                    );
                    match self.apply(env.clone()) {
                        Ok(_) => {
                            self.publish_to_nats(&env).await;
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Ok(()));
                            }
                        }
                        Err(e) => {
                            let err = e.to_string();
                            warn!(
                                wallet = %wallet,
                                error = %err,
                                "Failed to apply TierUpdate"
                            );
                            if let Some(tx) = applied_tx {
                                let _ = tx.send(Err(err));
                            }
                        }
                    }
                }

                // Handle PM settlement admin commands.
                Some(request) = async {
                    match self.pm_settlement_admin_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if self.runtime_quiesced {
                        let _ = request.applied_tx.send(Err("engine is quiesced".to_string()));
                        continue;
                    }
                    let timestamp_ms = get_timestamp_millis();
                    let command =
                        Self::stamp_pm_settlement_admin_command(request.command, timestamp_ms);
                    let request_uuid = Self::pm_settlement_request_id(&command)
                        .unwrap_or_else(|| {
                            panic!(
                                "RUNTIME_INVARIANT: PM settlement admin channel received unsupported command {}",
                                command.command_type()
                            )
                        });
                    let request_id = request_uuid.to_string();
                    let env = crate::rsm::apply::CommandEnvelope::new(timestamp_ms, command);
                    if matches!(
                        &env.command,
                        crate::rsm::apply::EngineCommand::RecordPmVaultDeposit(_)
                            | crate::rsm::apply::EngineCommand::RequestPmVaultWithdrawal(_)
                    ) && self
                        .external_option_command_already_journaled(
                            &request_id,
                            env.command.command_type(),
                        )
                        .await
                    {
                        let _ = request.applied_tx.send(Ok(()));
                        continue;
                    }
                    if let Err(error) = self.preflight_pm_settlement_admin_command(&env.command) {
                        let _ = request.applied_tx.send(Err(error));
                        continue;
                    }
                    self.journal_pm_settlement_admin_command(&env, request_uuid).await;
                    let output = self.apply(env.clone()).unwrap_or_else(|error| {
                        panic!(
                            "RUNTIME_INVARIANT: PM settlement command {} request_id {} passed preflight but apply rejected: {}",
                            env.command.command_type(),
                            request_id,
                            error
                        )
                    });
                    if let Err(error) = self.apply_pm_settlement_projection_effects_sync(
                        &output.pm_settlement_effects,
                        &request_id,
                    ) {
                        panic!(
                            "CRITICAL_FAILURE: failed to apply PM settlement projection effects for request_id {}: {}",
                            request_id, error
                        );
                    }
                    self.publish_to_nats(&env).await;
                    self.publish_snapshot();
                    last_read_snapshot = Instant::now();
                    let _ = request.applied_tx.send(Ok(()));
                }

                // Handle HyperCore equity updates from Hydromancer feed.
                Some(update) = async {
                    match self.hypercore_equity_receiver.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                }, if !self.runtime_quiesced => {
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        update.timestamp_ms,
                        crate::rsm::apply::EngineCommand::HypercoreEquityUpdate {
                            wallet: update.wallet,
                            account_value: update.account_value,
                            timestamp_ms: update.timestamp_ms,
                        },
                    );
                    match self.apply(env.clone()) {
                        Ok(_) => {
                            self.publish_to_nats(&env).await;
                        }
                        Err(e) => {
                            warn!(
                                wallet = %update.wallet,
                                error = %e,
                                "Failed to apply HypercoreEquityUpdate"
                            );
                        }
                    }
                }

                // Handle market management requests through apply()
                Some(request) = self.market_receiver.recv() => {
                    if self.runtime_quiesced {
                        let status = match request.message.action {
                            MarketAction::CreateMarket => MarketUpdateStatus::MarketCreationFailed,
                            MarketAction::DeleteMarket => MarketUpdateStatus::MarketDeletionFailed,
                            MarketAction::ExpireMarket => MarketUpdateStatus::MarketPendingSettlement,
                        };
                        let _ = request
                            .response_tx
                            .send(MarketUpdateMessage {
                                market: request.message.market,
                                status,
                                timestamp: request.message.timestamp,
                                reason: Some("engine is quiesced".to_string()),
                            })
                            .await;
                        continue;
                    }
                    market_start = Instant::now();
                    let original_response_tx = request.response_tx;
                    let message = request.message;
                    let market_cmd = if matches!(message.action, MarketAction::ExpireMarket) {
                        match self.prepare_manual_market_expiry_context(&message).await {
                            Ok(context) => crate::rsm::apply::MarketActionCommand::with_expiry_context(message, context),
                            Err(e) => {
                                error!("Failed to prepare MarketAction expiry context: {}", e);
                                let _ = original_response_tx
                                    .send(MarketUpdateMessage {
                                        market: message.market,
                                        status: MarketUpdateStatus::MarketPendingSettlement,
                                        timestamp: message.timestamp,
                                        reason: Some(e.to_string()),
                                    })
                                    .await;
                                continue;
                            }
                        }
                    } else {
                        crate::rsm::apply::MarketActionCommand::new(message)
                    };
                    if let Err(e) = self.validate_manual_market_expiry_nats_payload(&market_cmd) {
                        error!("Failed to validate MarketAction NATS payload: {}", e);
                        let _ = original_response_tx
                            .send(MarketUpdateMessage {
                                market: market_cmd.message.market,
                                status: MarketUpdateStatus::MarketPendingSettlement,
                                timestamp: market_cmd.message.timestamp,
                                reason: Some(e.to_string()),
                            })
                            .await;
                        continue;
                    }
                    let failure_market = (
                        market_cmd.message.market.clone(),
                        market_cmd.message.timestamp,
                    );
                    let env = crate::rsm::apply::CommandEnvelope::new(
                        market_cmd.message.timestamp,
                        crate::rsm::apply::EngineCommand::MarketAction(market_cmd),
                    );
                    let nats_env = env.clone();
                    match self.apply(env) {
                        Ok(output) => {
                            self.journal_replay_owned_expiry_command(
                                &nats_env,
                                &output.balance_updates,
                            )
                            .await;
                            if let Err(e) = self.apply_market_effects(&output).await {
                                panic!(
                                    "CRITICAL_FAILURE: apply_market_effects() failed for MarketAction: {}",
                                    e
                                );
                            }
                            let events_emitted_by_expiry_effects =
                                if output.expiry_effects.is_empty() {
                                    false
                                } else if let Err(e) = self
                                    .apply_expiry_effects_and_events(&output, "manual-market-expiry")
                                    .await
                                {
                                    panic!(
                                        "CRITICAL_FAILURE: apply_expiry_effects_and_events() failed for MarketAction: {}. \
                                         Expiry runtime effects could not be applied after memory state advanced.",
                                        e
                                    );
                                } else {
                                    true
                                };
                            self.publish_to_nats(&nats_env).await;
                            if !events_emitted_by_expiry_effects {
                                for event in &output.events {
                                    self.ctx.deps.emit_event(event);
                                }
                            }
                            if let Some(response) = output.market_response {
                                let _ = original_response_tx.send(response).await;
                            }
                            self.publish_balance_updates_to_nats(&output.balance_updates)
                                .await;
                        }
                        Err(e) => {
                            error!("apply() failed for MarketAction: {}", e);
                            let (market, timestamp) = failure_market;
                            let _ = original_response_tx
                                .send(MarketUpdateMessage {
                                    market,
                                    status: MarketUpdateStatus::MarketPendingSettlement,
                                    timestamp,
                                    reason: Some(e.to_string()),
                                })
                                .await;
                            market_request_hist.record(market_start.elapsed().as_secs_f64());
                            continue;
                        }
                    }
                    self.publish_snapshot();
                    last_read_snapshot = Instant::now();
                    market_request_hist.record(market_start.elapsed().as_secs_f64());
                }

                // Handle live trading_mode updates from catalog manager.
                // `watch::Receiver::changed()` returns Ok(()) when a new
                // value has been published and Err when the sender side
                // is dropped (e.g. catalog manager disabled or crashed).
                // On Err we park forever — continuing to poll would hot-
                // loop since `changed()` returns Err synchronously after
                // the sender is gone.
                Some(()) = async {
                    match self.trading_mode_receiver.as_mut() {
                        Some(rx) => match rx.changed().await {
                            Ok(()) => Some(()),
                            Err(_) => {
                                std::future::pending::<()>().await;
                                unreachable!()
                            }
                        },
                        None => std::future::pending().await,
                    }
                }, if !self.runtime_quiesced => {
                    if let Some(rx) = self.trading_mode_receiver.as_mut() {
                        let update = rx.borrow_and_update().clone();
                        if !update.is_empty() {
                            info!(
                                update_size = update.len(),
                                "Received live trading_mode update from catalog manager"
                            );
                            let now_ms = get_timestamp_millis();
                            let env = crate::rsm::apply::CommandEnvelope::new(
                                now_ms,
                                crate::rsm::apply::EngineCommand::TradingModeUpdate {
                                    modes: update,
                                    timestamp_ms: now_ms,
                                },
                            );
                            let nats_env = env.clone();
                            match self.apply(env) {
                                Ok(_) => { self.publish_to_nats(&nats_env).await; }
                                Err(e) => { error!("apply() failed for TradingModeUpdate: {}", e); }
                            }
                        }
                    }
                }

                // Periodic expiry check — routed through apply() for hash chain
                _ = expiry_interval.tick(), if !self.runtime_quiesced => {
                    expiry_start = Instant::now();
                    let now_ms = get_timestamp_millis();
                    let env = match self.prepare_tick_expiry_env(now_ms).await {
                        Ok(env) => env,
                        Err(e) => {
                            panic!(
                                "CRITICAL_FAILURE: prepare_tick_expiry_env() failed: {}. \
                                 Expiry settlement cannot continue without explicit context.",
                                e
                            );
                        }
                    };
                    let nats_env = env.clone();
                    match self.apply(env) {
                        Ok(output) => {
                            self.journal_replay_owned_expiry_command(
                                &nats_env,
                                &output.balance_updates,
                            )
                            .await;
                            if let Err(e) = self
                                .apply_expiry_effects_and_events(&output, "tick-expiry")
                                .await
                            {
                                panic!(
                                    "CRITICAL_FAILURE: apply_expiry_effects_and_events() failed for TickExpiry: {}. \
                                     Expiry runtime effects could not be applied after memory state advanced.",
                                    e
                                );
                            } else {
                                self.publish_to_nats(&nats_env).await;
                            }
                            self.publish_tick_expiry_balance_updates(&output, "tick-expiry")
                                .await;
                        }
                        Err(e) => {
                            panic!(
                                "CRITICAL_FAILURE: apply() failed for TickExpiry: {}. \
                                 Expiry state could not be advanced deterministically.",
                                e
                            );
                        }
                    }
                    self.publish_snapshot();
                    last_read_snapshot = Instant::now();
                    expiry_check_hist.record(expiry_start.elapsed().as_secs_f64());
                }

                // Feed spot prices into the engine as deterministic commands
                _ = price_update_interval.tick(), if !self.runtime_quiesced => {
                    self.ingest_price_updates().await;
                    self.ingest_iv_updates().await;
                }

                // One-time post-startup reconciliation after persistence has had time to catch up.
                _ = &mut post_startup_reconcile, if !self.post_startup_reconciled => {
                    self.post_startup_reconciled = true;
                    ReplayRecovery::run_post_startup_reconciliation(&mut self);
                    if !self.sync_status.is_ready() {
                        self.sync_status.set_ready();
                        info!("UnifiedEngine sync status: Ready (post-startup reconciliation complete)");
                    }
                }

                // Periodic journal flush + engine state snapshot
                _ = snapshot_interval.tick() => {
                    let now = Instant::now();

                    if now.duration_since(self.last_snapshot) >= self.snapshot_interval {
                        snapshot_start = Instant::now();
                        self.flush_journal().await;

                        self.persist_engine_state_snapshot();
                        snapshot_hist.record(snapshot_start.elapsed().as_secs_f64());
                        self.last_snapshot = now;
                    }
                }

                // Handle shutdown signal
                _ = self.shutdown_receiver.recv() => {
                    info!("Unified engine received shutdown signal");
                    self.flush_journal().await;
                    self.persist_engine_state_snapshot();
                    break;
                }

                // If channels are closed, exit
                else => {
                    info!("Channels closed, shutting down engine");
                    self.flush_journal().await;
                    self.persist_engine_state_snapshot();
                    break;
                }
            }

            // Record loop iteration time (metrics crate uses lock-free atomics, very fast)
            loop_iteration_hist.record(loop_start.elapsed().as_secs_f64());
        }

        info!("Unified engine stopped");
    }

    async fn handle_quiesce_request(&mut self, request: EngineQuiesceRequest) {
        match request.action {
            EngineQuiesceAction::Quiesce => {
                self.runtime_quiesced = true;
                self.flush_journal().await;
                let snapshot_command_id = self.persist_engine_state_snapshot();
                let last_command_id =
                    snapshot_command_id.unwrap_or_else(|| self.current_wal_checkpoint_command_id());
                self.publish_snapshot();
                self.last_snapshot = Instant::now();
                info!(
                    last_command_id,
                    last_l2_seq = self.ctx.l2_update_seq.load(Ordering::SeqCst),
                    "Unified engine entered quiesced mode"
                );
                let _ = request.response_tx.send(EngineQuiesceReport {
                    phase: hypercall_recovery::RecoveryPhase::Snapshotting,
                    quiesced: true,
                    last_command_id,
                    last_l2_seq: self.ctx.l2_update_seq.load(Ordering::SeqCst),
                    snapshot_persisted: snapshot_command_id.is_some(),
                    paused_sources: crate::rsm::restart_components::registered_mutation_sources(),
                });
            }
            EngineQuiesceAction::Resume => {
                self.runtime_quiesced = false;
                info!("Unified engine resumed live mutation sources");
                let _ = request.response_tx.send(EngineQuiesceReport {
                    phase: hypercall_recovery::RecoveryPhase::Recovered,
                    quiesced: false,
                    last_command_id: self.current_wal_checkpoint_command_id(),
                    last_l2_seq: self.ctx.l2_update_seq.load(Ordering::SeqCst),
                    snapshot_persisted: false,
                    paused_sources: Vec::new(),
                });
            }
        }
    }

    /// Read spot prices from the external greeks cache and apply them as
    /// deterministic PriceUpdate commands to the engine state.
    async fn ingest_price_updates(&mut self) {
        self.ingest_price_updates_with_side_effects(true).await;
    }

    async fn ingest_price_updates_without_side_effects(&mut self) {
        self.ingest_price_updates_with_side_effects(false).await;
    }

    async fn ingest_price_updates_with_side_effects(&mut self, emit_side_effects: bool) {
        use crate::journal::engine_journal_batcher::{JournalEntry, JournalMessage};
        use crate::rsm::apply::{CommandEnvelope, EngineCommand, PriceUpdatePayload};
        use rust_decimal::Decimal;

        let greeks_cache = match &self.ctx.deps.greeks_cache {
            Some(cache) => cache.clone(),
            None => return,
        };

        let mut underlyings = greeks_cache.get_configured_underlyings();
        underlyings.sort();
        let now_ms = get_timestamp_millis();

        for underlying in &underlyings {
            if let Some(price_f64) = greeks_cache.get_spot_price(underlying).await {
                let spot_price = match Decimal::from_f64_retain(price_f64) {
                    Some(d) => d,
                    None => continue,
                };

                if self.ctx.spot_prices.get(underlying) == Some(&spot_price) {
                    continue;
                }

                let env = CommandEnvelope::new(
                    now_ms,
                    EngineCommand::PriceUpdate {
                        underlying: underlying.clone(),
                        spot_price,
                        timestamp_ms: now_ms,
                    },
                );

                #[cfg(feature = "rsm-state")]
                let cmd_identity_hash = emit_side_effects.then(|| env.command.identity_hash());
                let nats_env = emit_side_effects.then(|| env.clone());
                if let Err(e) = self.apply(env) {
                    warn!(
                        underlying = %underlying,
                        error = %e,
                        "Failed to apply PriceUpdate command"
                    );
                    continue;
                }
                if let Some(nats_env) = nats_env {
                    self.publish_to_nats(&nats_env).await;
                }

                if let (Some(ref batch_sender), Some(_cmd_identity_hash)) = (
                    &self.journal_batch_sender,
                    #[cfg(feature = "rsm-state")]
                    cmd_identity_hash,
                    #[cfg(not(feature = "rsm-state"))]
                    emit_side_effects.then_some(()),
                ) {
                    let payload = PriceUpdatePayload {
                        underlying: underlying.clone(),
                        spot_price,
                        timestamp_ms: now_ms,
                    };
                    let command_data = hypercall_types::serialize_to_wire_bytes(&payload);

                    let entry = JournalEntry {
                        received_ts_ms: now_ms,
                        command_data,
                        response_data: None,
                        order_id: None,
                        pre_digest: Default::default(),
                        post_digest: Default::default(),
                        duration_ms: 0,
                        events: Vec::new(),
                        outbox_appends: Vec::new(),
                        fill_side_effects: Vec::new(),
                        cash_withdrawal_side_effect: None,
                        balance_updates: Vec::new(),
                        created_at: Instant::now(),
                        commit_ack: None,
                        request_uuid: hypercall_db_diesel::engine_enums::DbUuid(
                            uuid::Uuid::new_v4(),
                        ),
                        command_type_enum: Some(
                            hypercall_db_diesel::engine_enums::CommandType::PriceUpdate,
                        ),
                        #[cfg(feature = "rsm-state")]
                        command_identity_hash: _cmd_identity_hash,
                        #[cfg(feature = "rsm-state")]
                        rsm_state_digest: Some(
                            crate::rsm::engine_snapshot::EngineStateDigest::from_ctx(
                                &self.ctx,
                                self.ctx
                                    .l2_update_seq
                                    .load(std::sync::atomic::Ordering::SeqCst),
                            ),
                        ),
                    };
                    if let Err(e) = batch_sender.send(JournalMessage::Entry(entry)).await {
                        panic!(
                            "JOURNAL_FATAL: failed to journal PriceUpdate for {}: {}",
                            underlying, e
                        );
                    }
                }
            }
        }
    }

    /// Read IV surfaces from the external vol oracle and apply them as
    /// deterministic IvUpdate commands to the engine state.
    ///
    /// Heavy work (surface rebuild, serialization) happens here, outside the
    /// engine's critical path. apply() only does two map inserts.
    async fn ingest_iv_updates(&mut self) {
        self.ingest_iv_updates_with_side_effects(true).await;
    }

    async fn ingest_iv_updates_without_side_effects(&mut self) {
        self.ingest_iv_updates_with_side_effects(false).await;
    }

    async fn ingest_iv_updates_with_side_effects(&mut self, emit_side_effects: bool) {
        use crate::journal::engine_journal_batcher::{JournalEntry, JournalMessage};
        use crate::rsm::apply::{CommandEnvelope, EngineCommand, IvUpdatePayload};
        use crate::vol_oracle::vol_surface_cache::VolatilitySurface;

        let vol_oracle = match &self.external_vol_oracle {
            Some(oracle) => oracle.clone(),
            None => return,
        };

        let statuses = vol_oracle.statuses();
        let now_ms = get_timestamp_millis();

        let mut all_underlyings: Vec<String> =
            statuses.iter().map(|s| s.underlying.clone()).collect();
        all_underlyings.sort();
        all_underlyings.dedup();

        // Clear engine surfaces for underlyings whose providers are no longer ready.
        // This ensures the fallback oracle takes over when feeds go stale.
        for underlying in &all_underlyings {
            let is_ready = statuses
                .iter()
                .any(|s| s.underlying == *underlying && s.ready);
            if !is_ready {
                if self.ctx.iv_surfaces.remove(underlying).is_some() {
                    if let Ok(mut shared) = self.engine_iv_surfaces.write() {
                        shared.remove(underlying);
                    }
                    debug!(
                        underlying = %underlying,
                        "Cleared stale engine IV surface (provider no longer ready)"
                    );
                }
                continue;
            }
        }

        let ready_underlyings: Vec<String> = all_underlyings
            .iter()
            .filter(|u| statuses.iter().any(|s| s.underlying == **u && s.ready))
            .cloned()
            .collect();

        for underlying in &ready_underlyings {
            let snapshot = match vol_oracle.get_surface_snapshot(underlying) {
                Some(s) if !s.strike_points.is_empty() => s,
                _ => match synthesize_fixed_surface_snapshot(
                    &self.ctx.orderbooks,
                    &vol_oracle,
                    &statuses,
                    underlying,
                    now_ms as i64,
                ) {
                    Some(snapshot) => snapshot,
                    None => continue,
                },
            };

            // Always rebuild surface outside the engine loop
            let mut surface = VolatilitySurface::new();
            for point in &snapshot.strike_points {
                surface.insert(point.strike, point.expiry, point.iv);
            }

            // Serialize in parallel with content comparison (both outside critical path)
            let journal_data = if emit_side_effects && self.journal_batch_sender.is_some() {
                let payload = IvUpdatePayload {
                    underlying: underlying.clone(),
                    strike_points: snapshot.strike_points.clone(),
                    timestamp_ms: now_ms,
                };
                Some(hypercall_types::serialize_to_wire_bytes(&payload))
            } else {
                None
            };

            // Full content comparison: check every point against existing surface.
            // This is outside the engine loop so compute cost doesn't matter.
            // Only apply + journal if something actually changed.
            let changed = match self.ctx.iv_surfaces.get(underlying) {
                None => true,
                Some(existing) => {
                    // Detect point removals (expiry drops, grid shrinks)
                    snapshot.strike_points.len() != existing.len()
                        || snapshot.strike_points.iter().any(|point| {
                            existing
                                .get_interpolated(point.strike, point.expiry)
                                .map(|existing_iv| (existing_iv - point.iv).abs() > 1e-12)
                                .unwrap_or(true)
                        })
                }
            };

            if !changed {
                if let Some(ts) = snapshot.last_update_ts_ms {
                    self.ctx.iv_source_timestamps.insert(underlying.clone(), ts);
                }
                continue;
            }

            let env = CommandEnvelope::new(
                now_ms,
                EngineCommand::IvUpdate {
                    underlying: underlying.clone(),
                    surface,
                    journal_data: journal_data.clone(),
                    timestamp_ms: now_ms,
                },
            );

            if let Some(ts) = snapshot.last_update_ts_ms {
                self.ctx.iv_source_timestamps.insert(underlying.clone(), ts);
            }

            #[cfg(feature = "rsm-state")]
            let cmd_identity_hash = emit_side_effects.then(|| env.command.identity_hash());
            let nats_env = emit_side_effects.then(|| env.clone());
            if let Err(e) = self.apply(env) {
                warn!(
                    underlying = %underlying,
                    error = %e,
                    "Failed to apply IvUpdate command"
                );
                continue;
            }
            if let Some(nats_env) = nats_env {
                self.publish_to_nats(&nats_env).await;
            }

            if let (Some(ref batch_sender), Some(command_data), Some(_cmd_identity_hash)) = (
                &self.journal_batch_sender,
                journal_data,
                #[cfg(feature = "rsm-state")]
                cmd_identity_hash,
                #[cfg(not(feature = "rsm-state"))]
                emit_side_effects.then_some(()),
            ) {
                let entry = JournalEntry {
                    received_ts_ms: now_ms,
                    command_data,
                    response_data: None,
                    order_id: None,
                    pre_digest: Default::default(),
                    post_digest: Default::default(),
                    duration_ms: 0,
                    events: Vec::new(),
                    outbox_appends: Vec::new(),
                    fill_side_effects: Vec::new(),
                    cash_withdrawal_side_effect: None,
                    balance_updates: Vec::new(),
                    created_at: Instant::now(),
                    commit_ack: None,
                    request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
                    command_type_enum: Some(
                        hypercall_db_diesel::engine_enums::CommandType::IvUpdate,
                    ),
                    #[cfg(feature = "rsm-state")]
                    command_identity_hash: _cmd_identity_hash,
                    #[cfg(feature = "rsm-state")]
                    rsm_state_digest: Some(
                        crate::rsm::engine_snapshot::EngineStateDigest::from_ctx(
                            &self.ctx,
                            self.ctx
                                .l2_update_seq
                                .load(std::sync::atomic::Ordering::SeqCst),
                        ),
                    ),
                };
                if let Err(e) = batch_sender.send(JournalMessage::Entry(entry)).await {
                    panic!(
                        "JOURNAL_FATAL: failed to journal IvUpdate for {}: {}",
                        underlying, e
                    );
                }
            }
        }
    }
}

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

    #[test]
    fn external_durable_mutation_uuid_accepts_uuid() {
        let request_id = uuid::Uuid::now_v7().to_string();
        let parsed =
            UnifiedEngine::require_external_durable_mutation_uuid("DepositUpdate", &request_id);
        assert_eq!(parsed.to_string(), request_id);
    }

    #[test]
    #[should_panic(
        expected = "RUNTIME_INVARIANT: external engine command DepositUpdate request_id not-a-uuid is not a UUID"
    )]
    fn external_durable_mutation_uuid_panics_on_non_uuid() {
        UnifiedEngine::require_external_durable_mutation_uuid("DepositUpdate", "not-a-uuid");
    }

    #[test]
    fn empty_external_balance_updates_are_idempotent_for_journal_retries() {
        assert!(UnifiedEngine::is_idempotent_empty_balance_update(
            "DepositUpdate"
        ));
        assert!(UnifiedEngine::is_idempotent_empty_balance_update(
            "LiquidationBonusUpdate"
        ));
        assert!(!UnifiedEngine::is_idempotent_empty_balance_update(
            "CashWithdrawalUpdate"
        ));
    }

    #[test]
    fn pm_settlement_admin_stamp_replaces_client_timestamp() {
        let client_timestamp_ms = 1_800_000_000_000;
        let server_timestamp_ms = client_timestamp_ms + 60_000;
        let command = crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(
            hypercall_engine::command::AccruePmSettlementInterestCommand {
                request_id: uuid::Uuid::now_v7(),
                input_digest: "admin-pm-interest-test".to_string(),
                wallet: hypercall_types::wallet_address::test_wallet(1),
                underlying: "BTC".to_string(),
                to_ms: client_timestamp_ms as i64,
                timestamp_ms: client_timestamp_ms,
            },
        );

        let stamped =
            UnifiedEngine::stamp_pm_settlement_admin_command(command, server_timestamp_ms);

        match stamped {
            crate::rsm::apply::EngineCommand::AccruePmSettlementInterest(command) => {
                assert_eq!(command.timestamp_ms, server_timestamp_ms);
                assert_eq!(command.to_ms, client_timestamp_ms as i64);
            }
            other => panic!("unexpected command {}", other.command_type()),
        }
    }

    #[test]
    fn expire_market_is_replay_journal_owned() {
        let market = hypercall_types::Market {
            symbol: "BTC-20260531-70000-C".to_string(),
            underlying: "BTC".to_string(),
            expiry: 20260531,
            strike: dec!(70000),
            option_type: hypercall_types::OptionType::Call,
        };
        let command = crate::rsm::apply::MarketActionCommand::with_expiry_context(
            hypercall_types::MarketActionMessage {
                market,
                action: hypercall_types::MarketAction::ExpireMarket,
                timestamp: 1_800_000_000_000,
            },
            crate::rsm::apply::TickExpiryContext::empty(),
        );
        let env = crate::rsm::apply::CommandEnvelope::new(
            1_800_000_000_000,
            crate::rsm::apply::EngineCommand::MarketAction(command),
        );

        let (command_type, data, label) = UnifiedEngine::replay_owned_expiry_command_payload(&env)
            .expect("ExpireMarket must be journaled for replay");

        assert_eq!(
            command_type,
            hypercall_db_diesel::engine_enums::CommandType::ExpireMarket
        );
        assert_eq!(label, "ExpireMarket");
        let decoded = crate::nats::deserialize::deserialize_command(
            crate::nats::CommandType::MarketAction,
            crate::nats::COMMAND_WIRE_VERSION_V1,
            &data,
        )
        .expect("journaled ExpireMarket payload should decode for replay");
        assert!(matches!(
            decoded,
            crate::rsm::apply::EngineCommand::MarketAction(command)
                if command.message.action == hypercall_types::MarketAction::ExpireMarket
        ));
    }

    #[test]
    fn non_empty_tick_expiry_is_replay_journal_owned() {
        let context = crate::rsm::apply::TickExpiryContext {
            due_expiries: vec![crate::rsm::apply::TickExpiryDueGroup {
                expiry_ts: 1_800_000_000,
                symbols: vec!["BTC-20260531-70000-C".to_string()],
            }],
            pending_settlements: Vec::new(),
            settlement_prices: Vec::new(),
            margin_modes: Vec::new(),
            pm_settlements: Vec::new(),
        };
        let env = crate::rsm::apply::CommandEnvelope::new(
            1_800_000_000_000,
            crate::rsm::apply::EngineCommand::TickExpiry {
                now_ms: 1_800_000_000_000,
                context,
            },
        );

        let (command_type, data, label) = UnifiedEngine::replay_owned_expiry_command_payload(&env)
            .expect("non-empty TickExpiry must be journaled for replay");

        assert_eq!(
            command_type,
            hypercall_db_diesel::engine_enums::CommandType::TickExpiry
        );
        assert_eq!(label, "TickExpiry");
        let decoded = crate::nats::deserialize::deserialize_command(
            crate::nats::CommandType::TickExpiry,
            crate::nats::COMMAND_WIRE_VERSION_V1,
            &data,
        )
        .expect("journaled TickExpiry payload should decode for replay");
        assert!(matches!(
            decoded,
            crate::rsm::apply::EngineCommand::TickExpiry { .. }
        ));
    }

    #[test]
    fn replay_owned_expiry_journal_entry_carries_balance_updates() {
        let context = crate::rsm::apply::TickExpiryContext {
            due_expiries: vec![crate::rsm::apply::TickExpiryDueGroup {
                expiry_ts: 1_800_000_000,
                symbols: vec!["BTC-20260531-70000-C".to_string()],
            }],
            pending_settlements: Vec::new(),
            settlement_prices: Vec::new(),
            margin_modes: Vec::new(),
            pm_settlements: Vec::new(),
        };
        let env = crate::rsm::apply::CommandEnvelope::new(
            1_800_000_000_000,
            crate::rsm::apply::EngineCommand::TickExpiry {
                now_ms: 1_800_000_000_000,
                context,
            },
        );
        let balance_update = hypercall_types::BalanceUpdate {
            balance_update_seq: 17,
            wallet: hypercall_types::wallet_address::test_wallet(7),
            delta: dec!(12.5),
            balance_after: dec!(112.5),
            reason: hypercall_types::BalanceUpdateReason::Settlement,
            reference_id: Some("settlement:BTC-20260531-70000-C".to_string()),
            source_command_id: None,
            timestamp_ms: 1_800_000_000_000,
        };
        let request_uuid = hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::nil());

        let (entry, label) = UnifiedEngine::replay_owned_expiry_journal_entry(
            &env,
            std::slice::from_ref(&balance_update),
            request_uuid,
            None,
        )
        .expect("non-empty TickExpiry must produce a replay journal entry");

        assert_eq!(label, "TickExpiry");
        assert_eq!(entry.balance_updates, vec![balance_update]);
        assert_eq!(entry.request_uuid, request_uuid);
        assert_eq!(
            entry.command_type_enum,
            Some(hypercall_db_diesel::engine_enums::CommandType::TickExpiry)
        );
    }

    #[test]
    fn empty_tick_expiry_is_not_replay_journal_owned() {
        let env = crate::rsm::apply::CommandEnvelope::new(
            1_800_000_000_000,
            crate::rsm::apply::EngineCommand::TickExpiry {
                now_ms: 1_800_000_000_000,
                context: crate::rsm::apply::TickExpiryContext::empty(),
            },
        );

        assert!(UnifiedEngine::replay_owned_expiry_command_payload(&env).is_none());
    }

    #[test]
    fn synthesize_fixed_surface_snapshot_uses_live_orderbook_grid() {
        let mut orderbooks = std::collections::HashMap::new();
        orderbooks.insert(
            "GOLD-20261231-4700-C".to_string(),
            OrderBook::with_symbol(
                hypercall_types::expiry_date_to_timestamp("GOLD", 20261231) as u64,
                dec!(4700),
                hypercall_types::OptionType::Call,
                "GOLD-20261231-4700-C".to_string(),
            ),
        );
        orderbooks.insert(
            "GOLD-20261231-4700-P".to_string(),
            OrderBook::with_symbol(
                hypercall_types::expiry_date_to_timestamp("GOLD", 20261231) as u64,
                dec!(4700),
                hypercall_types::OptionType::Put,
                "GOLD-20261231-4700-P".to_string(),
            ),
        );

        let oracle: crate::vol_oracle::SharedVolOracle =
            std::sync::Arc::new(crate::vol_oracle::FixedTestRiskVolOracle::with_underlyings(
                0.50,
                vec!["GOLD".to_string()],
            ));
        let statuses = oracle.statuses();
        let snapshot = synthesize_fixed_surface_snapshot(
            &orderbooks,
            &oracle,
            &statuses,
            "GOLD",
            1_800_000_000_000,
        )
        .expect("fixed snapshot should be synthesized");

        assert_eq!(snapshot.strike_points.len(), 1);
        assert_eq!(snapshot.strike_points[0].strike, 4700.0);
        assert_eq!(
            snapshot.strike_points[0].expiry,
            hypercall_types::expiry_date_to_timestamp("GOLD", 20261231) as i64
        );
        assert_eq!(snapshot.strike_points[0].iv, 0.50);
    }

    #[test]
    fn max_listed_option_expiry_uses_engine_orderbooks() {
        let mut orderbooks = std::collections::HashMap::new();
        orderbooks.insert(
            "GOLD-20261231-4700-C".to_string(),
            OrderBook::with_symbol(
                hypercall_types::expiry_date_to_timestamp("GOLD", 20261231) as u64,
                dec!(4700),
                hypercall_types::OptionType::Call,
                "GOLD-20261231-4700-C".to_string(),
            ),
        );
        orderbooks.insert(
            "GOLD-20270630-4700-C".to_string(),
            OrderBook::with_symbol(
                hypercall_types::expiry_date_to_timestamp("GOLD", 20270630) as u64,
                dec!(4700),
                hypercall_types::OptionType::Call,
                "GOLD-20270630-4700-C".to_string(),
            ),
        );
        orderbooks.insert(
            "BTC-20280630-100000-C".to_string(),
            OrderBook::with_symbol(
                hypercall_types::expiry_date_to_timestamp("BTC", 20280630) as u64,
                dec!(100000),
                hypercall_types::OptionType::Call,
                "BTC-20280630-100000-C".to_string(),
            ),
        );

        assert_eq!(
            max_listed_option_expiry_ts_ms(&orderbooks, "GOLD").unwrap(),
            hypercall_types::expiry_date_to_timestamp("GOLD", 20270630) * 1_000
        );
    }

    #[test]
    fn max_listed_option_expiry_rejects_unknown_underlying() {
        let orderbooks = std::collections::HashMap::new();
        let error = max_listed_option_expiry_ts_ms(&orderbooks, "GOLD")
            .expect_err("missing listed markets should reject");
        assert!(error.contains("no listed option markets"));
    }
}