hypercall/runtime/

integrated.rs

use anyhow::Context;
use hypercall_types::WalletAddress;
use metrics::gauge;
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::signal;
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};

#[cfg(feature = "testnet")]
const TESTNET_BOOTSTRAP_DEPOSIT_SEQUENCE_BASE: u64 = 9_000_000_000;

#[cfg(feature = "testnet")]
fn testnet_bootstrap_deposit_sequence(index: usize) -> u64 {
    TESTNET_BOOTSTRAP_DEPOSIT_SEQUENCE_BASE
        .checked_add(index as u64)
        .expect("testnet bootstrap deposit sequence overflow")
}

#[cfg(feature = "testnet")]
fn testnet_bootstrap_source_event_hash(sequence: u64) -> alloy::primitives::FixedBytes<32> {
    let mut bytes = [0u8; 32];
    bytes[..16].copy_from_slice(b"testnet_bootstra");
    bytes[24..].copy_from_slice(&sequence.to_be_bytes());
    alloy::primitives::FixedBytes::from(bytes)
}

#[cfg(feature = "testnet")]
fn testnet_bootstrap_request_id(index: usize) -> String {
    uuid::Uuid::from_u128(
        0x1707_0000_0000_0000_0000_0000_0000_0000u128
            .checked_add(index as u128)
            .expect("testnet bootstrap request UUID overflow"),
    )
    .to_string()
}

fn parse_configured_wallet_set(
    config_name: &str,
    wallets: &[String],
) -> anyhow::Result<std::collections::BTreeSet<WalletAddress>> {
    wallets
        .iter()
        .map(|wallet| {
            WalletAddress::from_str(wallet)
                .map_err(|error| anyhow::anyhow!("Invalid {config_name} wallet {wallet}: {error}"))
        })
        .collect()
}

/// Record a startup phase duration as both a log line and a Prometheus gauge.
/// Uses gauges (not histograms) since startup only happens once per process.
macro_rules! startup_timing {
    ($phase:expr, $elapsed:expr) => {{
        let elapsed = $elapsed;
        info!("[startup-timing] {}: {:?}", $phase, elapsed);
        gauge!("ht_startup_phase_seconds", "phase" => $phase).set(elapsed.as_secs_f64());
    }};
    ($phase:expr, $elapsed:expr, $($arg:tt)+) => {{
        let elapsed = $elapsed;
        info!("[startup-timing] {}: {:?} ({})", $phase, elapsed, format!($($arg)+));
        gauge!("ht_startup_phase_seconds", "phase" => $phase).set(elapsed.as_secs_f64());
    }};
}

use crate::observability::metrics_collector::{MetricsCollector, MetricsCollectorConfig};
use crate::shared::option_token_address::ensure_option_token_derivation_supported;
use crate::shared::service::{Service, ServiceRegistry};
use crate::shared::shutdown::Shutdown;
use crate::shared::task_group::TaskGroup;

use crate::hypercore::HypercorePositionService;
use crate::messaging::{ChannelEventBus, EventBusTrait};
use crate::portfolio::PortfolioService;
use crate::price_oracle::hydromancer_client::{HydromancerClient, HydromancerConfig};
use crate::price_oracle::hyperliquid_oracle::{
    HyperliquidMarkPriceOracle, HyperliquidOracleConfig,
};
use crate::price_oracle::hyperliquid_ws::HyperliquidWsFeed;
use crate::read_cache::greeks::GreeksCache;
use crate::read_cache::portfolio::PortfolioCache;
use crate::readiness::{ReadinessRegistry, SyncStatusReadiness, VolOracleReadiness};
use crate::rsm::margin_service::SpanMarginService;
use crate::rsm::portfolio_margin::risk_account_builder::RiskAccountBuilder;
use crate::rsm::unified_engine::{UnifiedEngineBuilder, UnifiedEngineRequest};
use crate::runtime::tasks::bbo_snapshot::{BboSnapshotService, BboSnapshotTaskConfig};
use crate::runtime::tasks::historical_pnl::{HistoricalPnlSnapshotTask, HistoricalPnlTaskConfig};
use crate::runtime::tasks::historical_theo::{
    HistoricalTheoSnapshotTask, HistoricalTheoTaskConfig,
};
use crate::runtime::tasks::index_price_publisher::{
    IndexPricePublisher, IndexPriceUpdatePublisher,
};
use crate::snapshot::{
    DbInstrumentsSnapshotLoader, DbInstrumentsSnapshotWriter, InstrumentsSnapshotTask,
    PortfolioSnapshotTask, SnapshotTaskConfig,
};
use crate::standard_margin::{StandardAccountBuilder, StandardMarginService};
use crate::startup::{oracles, services};
use crate::types::Config;
use crate::vol_oracle::VolOracleFactory;
use axum::body::Body;
use axum::extract::{Request, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::Router;
use hypercall_api::candles::{
    candle_ws_poll_interval_ms, CandleWsPublisher, HyperliquidCandleSource, UnderlyingCandleSource,
};
use hypercall_api::trading_halt::TradingHaltState;
use hypercall_api::websocket::{
    PubSubManager, WsCompetitionFinalStanding, WsCompetitionGapUpdate, WsCompetitionRankChange,
    WsState,
};
use hypercall_api::{directives, handlers, middleware};
use hypercall_competition::CompetitionService;
use hypercall_db_diesel::DatabaseHandler;
use hypercall_types::EngineMessage;
use std::collections::{HashMap, HashSet};
use tokio::task::JoinHandle;
use tower::util::ServiceExt;

type PromoteSender = Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>>;

impl IndexPriceUpdatePublisher for PubSubManager {
    fn publish_index_prices(&self, update: hypercall_types::ws_protocol::WsIndexPriceUpdate) {
        PubSubManager::publish_index_prices(self, update);
    }
}

#[derive(Clone)]
pub struct StartupProgress {
    phase: Arc<std::sync::Mutex<String>>,
    counter: Arc<AtomicU64>,
    last_progress_unix_ms: Arc<AtomicU64>,
}

pub(crate) struct StartupProgressSnapshot {
    pub(crate) phase: String,
    pub(crate) counter: u64,
    pub(crate) last_progress_unix_ms: Option<u64>,
    pub(crate) last_progress_age_ms: Option<u64>,
}

impl StartupProgress {
    fn new(initial_phase: impl Into<String>) -> Self {
        let progress = Self {
            phase: Arc::new(std::sync::Mutex::new(String::new())),
            counter: Arc::new(AtomicU64::new(0)),
            last_progress_unix_ms: Arc::new(AtomicU64::new(0)),
        };
        progress.mark(initial_phase);
        progress
    }

    fn mark(&self, phase: impl Into<String>) {
        let phase = phase.into();
        *self
            .phase
            .lock()
            .expect("startup progress phase mutex poisoned") = phase;
        self.record_progress();
    }

    fn heartbeat(&self) {
        self.record_progress();
    }

    pub(crate) fn snapshot(&self) -> StartupProgressSnapshot {
        let phase = self
            .phase
            .lock()
            .expect("startup progress phase mutex poisoned")
            .clone();
        let counter = self.counter.load(Ordering::Relaxed);
        let last_progress_unix_ms = match self.last_progress_unix_ms.load(Ordering::Relaxed) {
            0 => None,
            ts => Some(ts),
        };
        let last_progress_age_ms =
            last_progress_unix_ms.map(|ts| current_unix_ms().saturating_sub(ts));

        StartupProgressSnapshot {
            phase,
            counter,
            last_progress_unix_ms,
            last_progress_age_ms,
        }
    }

    fn record_progress(&self) {
        self.counter.fetch_add(1, Ordering::Relaxed);
        self.last_progress_unix_ms
            .store(current_unix_ms(), Ordering::Relaxed);
    }
}

fn current_unix_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("system clock before UNIX_EPOCH")
        .as_millis() as u64
}

#[derive(Clone)]
struct BootstrapHttpState {
    shutdown: Shutdown,
    standby_progress: Option<crate::nats::replay_loop::ReplayProgress>,
    standby_promote: Option<PromoteSender>,
    startup_progress: Option<StartupProgress>,
    full_app: Arc<tokio::sync::RwLock<Option<Router>>>,
}

struct BootstrapHttpServer {
    full_app: Arc<tokio::sync::RwLock<Option<Router>>>,
    task: JoinHandle<()>,
}

fn build_bootstrap_http_app(state: BootstrapHttpState) -> Router {
    Router::new()
        .route("/health", get(bootstrap_health))
        .route("/standby-ready", get(bootstrap_standby_ready))
        .fallback(bootstrap_fallback)
        .with_state(state)
}

async fn bootstrap_health(State(state): State<BootstrapHttpState>) -> Response {
    let response = hypercall_api::models::HealthResponse {
        status: if state.shutdown.is_triggered() {
            "shutting_down".to_string()
        } else {
            "ok".to_string()
        },
    };

    if state.shutdown.is_triggered() {
        (
            StatusCode::SERVICE_UNAVAILABLE,
            hypercall_api::sonic_json::SonicJson(response),
        )
            .into_response()
    } else {
        (
            StatusCode::OK,
            hypercall_api::sonic_json::SonicJson(response),
        )
            .into_response()
    }
}

async fn bootstrap_standby_ready(State(state): State<BootstrapHttpState>) -> Response {
    if state.full_app.read().await.is_none() {
        let response = match state.standby_progress.as_ref() {
            Some(progress) => {
                let mut payload = handlers::health::standby_status_payload(progress, false);
                payload["status"] = serde_json::json!("api_starting");
                payload["promotable"] = serde_json::json!(false);
                payload["api_router_ready"] = serde_json::json!(false);
                add_startup_progress_payload(&mut payload, state.startup_progress.as_ref());
                payload
            }
            None => {
                let mut payload = serde_json::json!({
                "status": "not_standby_mode",
                "api_router_ready": false,
                });
                add_startup_progress_payload(&mut payload, state.startup_progress.as_ref());
                payload
            }
        };
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            hypercall_api::sonic_json::SonicJson(response),
        )
            .into_response();
    }

    let already_promoted =
        handlers::health::standby_already_promoted(state.standby_promote.as_ref());
    handlers::health::standby_ready_response_with_startup(
        state
            .standby_progress
            .as_ref()
            .map(|progress| progress as &dyn hypercall_api::runtime_status::StandbyReplayProgress),
        already_promoted,
        state
            .startup_progress
            .as_ref()
            .map(|progress| progress as &dyn hypercall_api::runtime_status::StartupProgressReader),
    )
}

fn add_startup_progress_payload(
    payload: &mut serde_json::Value,
    progress: Option<&StartupProgress>,
) {
    let Some(progress) = progress else {
        return;
    };
    let snapshot = progress.snapshot();
    payload["startup_phase"] = serde_json::json!(snapshot.phase);
    payload["startup_progress_counter"] = serde_json::json!(snapshot.counter);
    payload["last_startup_progress_unix_ms"] = serde_json::json!(snapshot.last_progress_unix_ms);
    payload["last_startup_progress_age_ms"] = serde_json::json!(snapshot.last_progress_age_ms);
}

async fn bootstrap_fallback(
    State(state): State<BootstrapHttpState>,
    request: Request<Body>,
) -> Response {
    let app = state.full_app.read().await.clone();
    let Some(app) = app else {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            hypercall_api::sonic_json::SonicJson(serde_json::json!({
                "status": "starting",
                "message": "API router is still starting"
            })),
        )
            .into_response();
    };

    app.oneshot(request)
        .await
        .unwrap_or_else(|error| match error {})
}

fn start_bootstrap_http_server(
    listener: tokio::net::TcpListener,
    addr: SocketAddr,
    shutdown: &Shutdown,
    standby_progress: Option<crate::nats::replay_loop::ReplayProgress>,
    standby_promote: Option<PromoteSender>,
    startup_progress: Option<StartupProgress>,
) -> BootstrapHttpServer {
    let full_app = Arc::new(tokio::sync::RwLock::new(None));
    let app = build_bootstrap_http_app(BootstrapHttpState {
        shutdown: shutdown.clone(),
        standby_progress,
        standby_promote,
        startup_progress,
        full_app: full_app.clone(),
    });
    let mut http_shutdown_rx = shutdown.subscribe();

    let task = tokio::spawn(async move {
        tracing::info!("Bootstrap standby HTTP server listening on {}", addr);
        let server = axum::serve(listener, app).with_graceful_shutdown(async move {
            let _ = http_shutdown_rx.recv().await;
            tracing::info!("Shutting down bootstrap standby HTTP server...");
        });

        if let Err(error) = server.await {
            tracing::error!("Bootstrap standby HTTP server error: {}", error);
        }
    });

    BootstrapHttpServer { full_app, task }
}

pub async fn run_integrated_server(
    config: Config,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> anyhow::Result<()> {
    // Use IPv6 dual-stack binding (::) which accepts both IPv4 and IPv6 connections.
    let port = std::env::var("PORT").unwrap_or_else(|_| "3000".to_string());
    let addr: SocketAddr = format!("[::]:{}", port)
        .parse()
        .expect("valid socket address");
    run_integrated_server_on(config, addr, runtime).await
}

/// Run the integrated server on a specific address.
///
/// This is a convenience wrapper that creates a new Shutdown instance.
/// For testing or custom shutdown handling, use `run_integrated_server_on_with_shutdown`.
pub async fn run_integrated_server_on(
    config: Config,
    addr: SocketAddr,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> anyhow::Result<()> {
    let shutdown = Shutdown::new();
    run_integrated_server_on_with_shutdown(config, addr, shutdown, runtime).await
}

pub fn spawn_integrated_server_on(
    config: Config,
    addr: SocketAddr,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> tokio::task::JoinHandle<anyhow::Result<()>> {
    tokio::spawn(Box::pin(run_integrated_server_on(config, addr, runtime)))
}

/// Run the integrated server with an externally provided shutdown signal.
///
/// This allows tests and external callers to control when the server shuts down.
pub async fn run_integrated_server_on_with_shutdown(
    config: Config,
    addr: SocketAddr,
    shutdown: Shutdown,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> anyhow::Result<()> {
    let listener = tokio::net::TcpListener::bind(addr).await?;
    boxed_integrated_server_with_listener_and_shutdown(config, listener, shutdown, runtime).await
}

/// Run the integrated server on a pre-bound listener.
///
/// Useful for tests that must reserve a port before startup to avoid bind races.
pub async fn run_integrated_server_with_listener(
    config: Config,
    listener: tokio::net::TcpListener,
) -> anyhow::Result<()> {
    let runtime = Arc::new(crate::backend_config::BackendRuntime::from_legacy_env()?);
    let shutdown = Shutdown::new();
    boxed_integrated_server_with_listener_and_shutdown(config, listener, shutdown, runtime).await
}

fn boxed_integrated_server_with_listener_and_shutdown(
    config: Config,
    listener: tokio::net::TcpListener,
    shutdown: Shutdown,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send>> {
    Box::pin(run_integrated_server_with_listener_and_shutdown(
        config, listener, shutdown, runtime,
    ))
}

/// Build the narrow [`hypercall_admin::AdminState`] from the full api
/// [`handlers::AppState`].
///
/// `qp_cache` is sourced from the RFQ handler state because `AppState` does not
/// own the quote-provider registry directly. This mapping previously lived as
/// `AdminState::from_app_state` in `hypercall-api`; it moved here when admin
/// became a sibling crate so neither crate depends on the other.
fn build_admin_state(
    app_state: &handlers::AppState,
    qp_cache: Arc<hypercall_runtime_api::QuoteProviderCache>,
) -> hypercall_admin::state::AdminState {
    hypercall_admin::state::AdminState {
        db: app_state.db.clone(),
        sync_db: app_state.sync_db.clone(),
        portfolio_cache: app_state.portfolio_cache.clone(),
        greeks_cache: app_state.greeks_cache.clone(),
        quote_provider: app_state.quote_provider.clone(),
        tier_cache: app_state.tier_cache.clone(),
        risk_vol_oracle: app_state.risk_vol_oracle.clone(),
        instruments_cache: app_state.instruments_cache.clone(),
        engine_state_digest_provider: app_state.engine_state_digest_provider.clone(),
        engine_journal_reader: app_state.engine_journal_reader.clone(),
        balance_snapshot_provider: app_state.balance_snapshot_provider.clone(),
        recovery_safety_report: app_state.recovery_safety_report.clone(),
        qp_cache,
        competition: app_state.competition_service.clone(),
        order_sender: app_state.order_sender.clone(),
        engine_quiesce_sender: app_state.engine_quiesce_sender.clone(),
        tier_update_sender: app_state.tier_update_sender.clone(),
        pm_settlement_admin_sender: app_state.pm_settlement_admin_sender.clone(),
        agent_auth_sender: app_state.agent_auth_sender.clone(),
        rsm_signer: app_state.rsm_signer.clone(),
        trading_halt: app_state.trading_halt.clone(),
        standby_promote: app_state.standby_promote.clone(),
        standby_controller: app_state.standby_controller.clone(),
        drain_signal: app_state.drain_signal.clone(),
        is_draining: app_state.is_draining.clone(),
        build_info: app_state.build_info.clone(),
        runtime_config: Arc::new(hypercall_admin::state::AdminRuntimeConfig {
            wal_path: app_state.runtime_config.wal_path.clone(),
            db_host: app_state.runtime_config.db_host.clone(),
            db_name: app_state.runtime_config.db_name.clone(),
            testnet_mode: app_state.runtime_config.testnet_mode,
            portfolio_margin_pool_enabled: app_state.runtime_config.portfolio_margin_pool_enabled,
            portfolio_margin_settlement_allowlist: app_state
                .runtime_config
                .portfolio_margin_settlement_allowlist
                .clone(),
        }),
        server_started_at: app_state.server_started_at,
        boot_id: app_state.boot_id.clone(),
        admin_api_key: app_state.admin_api_key.clone(),
        allow_unauthenticated_monitoring: app_state.allow_unauthenticated_monitoring,
    }
}

/// Run the integrated server on a pre-bound listener with an external shutdown signal.
///
/// This is primarily used by integration tests that need race-free port reservation
/// and deterministic shutdown control.
pub async fn run_integrated_server_with_listener_and_shutdown(
    config: Config,
    listener: tokio::net::TcpListener,
    shutdown: Shutdown,
    runtime: Arc<crate::backend_config::BackendRuntime>,
) -> anyhow::Result<()> {
    let addr = listener.local_addr()?;
    let mut listener = Some(listener);

    // Initialize Prometheus metrics early to avoid losing metrics emitted during startup
    crate::observability::prometheus::init_prometheus();
    // Start the upkeep task now that we're in a tokio runtime context
    crate::observability::prometheus::start_upkeep_task();
    tracing::info!(
        "Starting integrated API server with unified engine on {}...",
        addr
    );

    let startup_begin = Instant::now();
    runtime.apply_process_env();

    #[cfg(not(feature = "test-endpoints"))]
    if runtime.config.modes.enable_test_endpoints {
        tracing::warn!(
            "modes.enable_test_endpoints=true in config but binary built without test-endpoints feature. \
             Test endpoints, mock feeds, and mock oracles are disabled."
        );
    }

    #[cfg(not(feature = "faucet"))]
    if runtime.config.modes.testnet_mode {
        tracing::info!(
            "modes.testnet_mode=true in config but binary built without faucet feature. \
             The /faucet route is not available."
        );
    }

    ensure_option_token_derivation_supported()
        .map_err(|err| anyhow::anyhow!("Option token derivation startup gate failed: {}", err))?;

    // Install the per-underlying expiry time policy before any component
    // derives expiry timestamps (engine recovery, caches, settlement, token
    // address derivation all depend on it).
    let expiry_times = runtime
        .config
        .catalog
        .expiry_times()
        .map_err(|err| anyhow::anyhow!("Invalid catalog expiry time configuration: {err:#}"))?;
    hypercall_types::install_expiry_times(expiry_times)
        .map_err(|err| anyhow::anyhow!("Expiry time policy install failed: {err}"))?;

    let database_resources = crate::startup::database::build_database_resources(&runtime).await?;
    let engine_database_url = database_resources.engine_database_url;
    let db_auth = database_resources.db_auth;
    let engine_db_auth = database_resources.engine_db_auth;
    let db_pool = database_resources.db_pool;
    let diesel_db = database_resources.diesel_db;
    let shared_db = database_resources.shared_db;
    let standby_mode_active = database_resources.standby_mode_active;
    let skip_db_migrations = database_resources.skip_db_migrations;
    let standby_progress = standby_mode_active.then(crate::nats::replay_loop::ReplayProgress::new);
    let startup_progress =
        standby_mode_active.then(|| StartupProgress::new("bootstrap_http_starting"));
    let (standby_promote, mut standby_replay_promote_rx): (
        Option<PromoteSender>,
        Option<tokio::sync::oneshot::Receiver<()>>,
    ) = if standby_mode_active {
        let (promote_tx, promote_rx) = tokio::sync::oneshot::channel::<()>();
        (
            Some(Arc::new(tokio::sync::Mutex::new(Some(promote_tx)))),
            Some(promote_rx),
        )
    } else {
        (None, None)
    };
    let bootstrap_http = if standby_mode_active {
        Some(start_bootstrap_http_server(
            listener
                .take()
                .expect("standby bootstrap HTTP requires the bound listener"),
            addr,
            &shutdown,
            standby_progress.clone(),
            standby_promote.clone(),
            startup_progress.clone(),
        ))
    } else {
        None
    };
    if let Some(progress) = startup_progress.as_ref() {
        progress.mark("bootstrap_http_ready");
    }

    // Create PubSub manager
    let pubsub = Arc::new(PubSubManager::new());

    // Create TaskGroup for tracking background tasks
    let mut task_group = TaskGroup::new();
    let mut service_registry = ServiceRegistry::new();
    let post_promote_services: Arc<tokio::sync::Mutex<Vec<Arc<dyn Service>>>> =
        Arc::new(tokio::sync::Mutex::new(Vec::new()));

    // Create channel for order processing (from HTTP handlers to unified engine)
    let (order_tx, order_rx) = mpsc::channel::<UnifiedEngineRequest>(100);

    // Create in-process channel event bus.
    let t = Instant::now();
    let event_bus: Arc<dyn EventBusTrait> = Arc::new(
        ChannelEventBus::new()
            .map_err(|e| anyhow::anyhow!("Failed to create ChannelEventBus: {}", e))?,
    );

    startup_timing!("event_bus_create", t.elapsed());

    // NOTE: Do NOT start event bus processing yet – start it after all caches
    // have subscribed to avoid missing early events.

    crate::startup::transaction_submitter::build_transaction_submitter_resources(
        &runtime,
        event_bus.clone(),
        Some(diesel_db.clone()),
        &shutdown,
        &mut task_group,
    )
    .await?;

    let sync_exchange_address =
        alloy::primitives::Address::from_str(&runtime.config.contracts.exchange_contract_address)
            .ok();

    // Get sender for unified engine to send events
    let engine_event_tx = event_bus.get_sender();

    // Direct channel for WS event forwarding — low-latency path for WebSocket updates.
    let (ws_direct_tx, ws_direct_rx_for_forwarder) = mpsc::unbounded_channel();

    let rsm_signer_chain_id = if runtime.config.modes.testnet_mode {
        hypercall_types::directives::HYPERCALL_TESTNET_CHAIN_ID
    } else {
        hypercall_types::directives::HYPERCALL_MAINNET_CHAIN_ID
    };

    // Initialize PortfolioCache with the shared handler
    let portfolio_cache = Arc::new(PortfolioCache::new(shared_db.clone()));

    // Initialize from latest snapshot (returns next engine command_id to replay)
    let t = Instant::now();
    let next_portfolio_command_id = portfolio_cache.initialize().await?;
    startup_timing!("portfolio_snapshot_load", t.elapsed());

    // NOTE: Readiness registry is built later, after all components are initialized.

    // Initialize TierCache early so margin-mode dependent projections can be
    // rebuilt before catchup replay.
    let tier_cache_config = crate::read_cache::tier::TierCacheConfig {
        max_open_orders_default: runtime.config.api.max_open_orders_default,
        max_open_positions_default: runtime.config.api.max_open_positions_default,
    };
    let tier_cache = Arc::new(crate::read_cache::tier::TierCache::new_with_config(
        shared_db.clone(),
        tier_cache_config,
    )?);
    let t = Instant::now();
    tier_cache.load_from_db().await?;
    startup_timing!("tier_cache", t.elapsed());

    // Initialize RateLimitCache for API rate limiting
    let t = Instant::now();
    let rate_limit_cache = Arc::new(
        hypercall_api::caches::rate_limit::RateLimitCache::new(
            tier_cache.clone(),
            runtime.secrets.redis_url.as_deref(),
        )
        .await,
    );
    startup_timing!("rate_limit_cache", t.elapsed());

    // Initialize LiquidationCache for pre-liquidation order blocking
    let liquidation_cache = Arc::new(crate::liquidator::LiquidationCache::new());
    if runtime.config.liquidation.enabled {
        liquidation_cache
            .load_from_db(diesel_db.as_ref())
            .await
            .map_err(|error| {
                anyhow::anyhow!("Failed to load liquidation cache from DB: {}", error)
            })?;
        info!("LiquidationCache initialized successfully");
    } else {
        info!("LiquidationCache restore skipped because liquidation runtime is disabled");
    }

    let journal_resources = crate::startup::journal::build_journal_resources(
        &runtime,
        &db_auth,
        &shutdown,
        &mut task_group,
    )?;
    let engine_journal_writer = journal_resources.engine_journal_writer;
    let journal_batch_sender = journal_resources.journal_batch_sender;

    // Wire tier_cache to PortfolioService BEFORE catchup replay. Runtime balance
    // reads are snapshot-backed via EngineSnapshot.balance_ledger.
    {
        let portfolio_svc = portfolio_cache.get_service();
        if let Some(impl_svc) = portfolio_svc
            .as_any()
            .downcast_ref::<crate::portfolio::PortfolioServiceImpl>()
        {
            impl_svc.set_tier_cache(tier_cache.clone()).await;
            info!("✓ Wired TierCache to PortfolioService for catchup replay");
        } else {
            return Err(anyhow::anyhow!(
                "Could not wire TierCache to PortfolioService for catchup replay: service is not PortfolioServiceImpl"
            ));
        }
    }

    {
        let t = Instant::now();
        info!(
            "Starting portfolio catchup replay from engine command_id={}",
            next_portfolio_command_id
        );

        let mut after_command_id = next_portfolio_command_id - 1;
        let mut total_events_replayed: u64 = 0;
        let mut total_custody_commands_replayed: u64 = 0;
        loop {
            let replay_db: &dyn hypercall_db::JournalReplayReader = shared_db.as_ref();
            let command_batch = replay_db
                .get_replay_commands_after_command_id_sync(after_command_id, None, 1000)
                .map_err(|e| anyhow::anyhow!("Failed to load portfolio replay commands: {}", e))?;

            let Some(last_command) = command_batch.last() else {
                break;
            };
            let end_command_id = last_command.command_id;

            let replay_events = replay_db
                .get_portfolio_events_for_command_range_sync(after_command_id + 1, end_command_id)
                .map_err(|e| anyhow::anyhow!("Failed to load portfolio replay events: {}", e))?;

            let mut replay_events_by_command = std::collections::BTreeMap::new();
            for replay_event in replay_events {
                replay_events_by_command
                    .entry(replay_event.command_id)
                    .or_insert_with(Vec::new)
                    .push(replay_event);
            }

            for command in &command_batch {
                if let Some(events) = replay_events_by_command.remove(&command.command_id) {
                    for replay_event in events {
                        match replay_event.event_type {
                            hypercall_db::EventType::OrderFilled => {
                                if replay_event.event_data.len() <= 1 {
                                    panic!(
                                        "CRITICAL_FAILURE: empty replay fill payload for command_id={}",
                                        replay_event.command_id
                                    );
                                }
                                let fill =
                                    match hypercall_types::EngineMessage::deserialize_from_wire(
                                        crate::shared::topics::TOPIC_FILLS,
                                        &replay_event.event_data,
                                    ) {
                                        Ok(hypercall_types::EngineMessage::OrderFilled {
                                            fill,
                                            ..
                                        }) => fill,
                                        Ok(other) => {
                                            panic!(
                                                "CRITICAL_FAILURE: expected replay fill for command_id={}, got {}",
                                                replay_event.command_id,
                                                other.type_name()
                                            )
                                        }
                                        Err(e) => {
                                            panic!(
                                                "CRITICAL_FAILURE: failed to deserialize replay fill for command_id={}: {}",
                                                replay_event.command_id,
                                                e
                                            )
                                        }
                                    };
                                portfolio_cache
                                    .replay_journal_fill(&fill, replay_event.command_id)
                                    .await;
                            }
                            hypercall_db::EventType::PositionExpired => {
                                if replay_event.event_data.len() <= 1 {
                                    panic!(
                                        "CRITICAL_FAILURE: empty replay position expiry payload for command_id={}",
                                        replay_event.command_id
                                    );
                                }
                                let expiry = rmp_serde::from_slice::<
                                    hypercall_types::PositionExpiredMessage,
                                >(&replay_event.event_data[1..])
                                .unwrap_or_else(|e| {
                                    panic!(
                                        "CRITICAL_FAILURE: failed to deserialize replay position expiry for command_id={}: {}",
                                        replay_event.command_id, e
                                    )
                                });
                                // TODO: Add a dedicated replay_journal_expiry method that skips
                                // the settlement_payouts lookup, analogous to replay_journal_fill.
                                // Current behavior is safe but adds startup latency.
                                portfolio_cache
                                    .handle_engine_message(
                                        EngineMessage::PositionExpired(expiry),
                                        replay_event.command_id,
                                    )
                                    .await;
                            }
                            other => {
                                panic!(
                                    "CRITICAL_FAILURE: unexpected portfolio replay event type {:?} for command_id={}",
                                    other, replay_event.command_id
                                )
                            }
                        }
                        total_events_replayed += 1;
                    }
                }

                let command_type = match command.command_type.as_str() {
                    "OptionDepositUpdate" => Some(crate::nats::CommandType::OptionDepositUpdate),
                    "OptionWithdrawalUpdate" => {
                        Some(crate::nats::CommandType::OptionWithdrawalUpdate)
                    }
                    _ => None,
                };
                let Some(command_type) = command_type else {
                    continue;
                };
                let decoded = crate::nats::deserialize::deserialize_command(
                    command_type,
                    crate::nats::LEGACY_COMMAND_WIRE_VERSION,
                    &command.command_data,
                )
                .unwrap_or_else(|error| {
                    panic!(
                        "CRITICAL_FAILURE: failed to decode portfolio custody command {} (command_id={}): {}",
                        command.command_type, command.command_id, error
                    )
                });
                match decoded {
                    crate::rsm::apply::EngineCommand::OptionDepositUpdate {
                        wallet,
                        symbol,
                        quantity,
                        ..
                    } => {
                        portfolio_cache
                            .replay_option_custody_delta(
                                wallet,
                                symbol,
                                quantity,
                                command.command_id,
                            )
                            .await;
                        total_custody_commands_replayed += 1;
                    }
                    crate::rsm::apply::EngineCommand::OptionWithdrawalUpdate {
                        wallet,
                        symbol,
                        quantity,
                        ..
                    } => {
                        portfolio_cache
                            .replay_option_custody_delta(
                                wallet,
                                symbol,
                                -quantity,
                                command.command_id,
                            )
                            .await;
                        total_custody_commands_replayed += 1;
                    }
                    other => {
                        panic!(
                            "CRITICAL_FAILURE: decoded unexpected portfolio custody command {} for command_id={}",
                            other.command_type(),
                            command.command_id
                        );
                    }
                }
            }

            if let Some((&command_id, _)) = replay_events_by_command.iter().next() {
                panic!(
                    "CRITICAL_FAILURE: portfolio replay event found for command_id={} without matching replay command in batch",
                    command_id
                );
            }

            after_command_id = end_command_id;
        }

        startup_timing!(
            "portfolio_catchup_replay",
            t.elapsed(),
            "{} events, {} custody commands",
            total_events_replayed,
            total_custody_commands_replayed
        );
        portfolio_cache.set_ready();
    }
    // Initialize instruments cache.
    let t = Instant::now();
    let instruments_cache =
        Arc::new(crate::read_cache::instruments_registry::InstrumentsCache::new());

    // Create instruments snapshot loader for snapshot-based initialization
    let instruments_snapshot_loader = DbInstrumentsSnapshotLoader::new(shared_db.clone());
    let bootstrap_db: Arc<dyn hypercall_db::BootstrapReader> = diesel_db.clone();
    let instruments_offsets = instruments_cache
        .clone()
        .initialize_with_snapshot(
            &bootstrap_db,
            event_bus.clone(),
            &instruments_snapshot_loader,
            shutdown.subscribe(),
        )
        .await?;

    let instruments_count = instruments_cache.len().await;
    startup_timing!(
        "instruments_snapshot_init",
        t.elapsed(),
        "{} instruments",
        instruments_count
    );

    // Catchup replay for instruments cache if we restored from snapshot
    if let Some(offsets) = instruments_offsets {
        let t = Instant::now();
        info!(
            "Instruments snapshot restored with {} stream offsets - starting catchup replay",
            offsets.len()
        );

        let instruments_cache_replay = instruments_cache.clone();
        let handler: Box<dyn Fn(EngineMessage) + Send + Sync> = Box::new(move |event| {
            // Only handle MarketUpdate events during replay
            if let EngineMessage::MarketUpdate(market_update) = event {
                let cache = instruments_cache_replay.clone();
                tokio::task::block_in_place(|| {
                    tokio::runtime::Handle::current().block_on(async {
                        if let Err(e) = cache.handle_market_update(market_update).await {
                            error!("Error replaying market update to instruments cache: {}", e);
                        }
                    });
                });
            }
        });

        let replay_result = event_bus.replay_from_offsets(offsets, handler).await;

        match replay_result {
            Ok(final_offsets) => {
                // Advance seq tracking to the final replay offsets so subsequent
                // snapshots record where live consumption should resume from.
                if let Err(e) = instruments_cache
                    .apply_snapshot_offsets(final_offsets)
                    .await
                {
                    error!("Failed to apply replay offsets to instruments cache: {}", e);
                }
                let final_count = instruments_cache.len().await;
                startup_timing!(
                    "instruments_catchup_replay",
                    t.elapsed(),
                    "{} instruments",
                    final_count
                );
                instruments_cache.sync_status().set_ready();
            }
            Err(e) => {
                startup_timing!("instruments_catchup_replay", t.elapsed(), "FAILED: {}", e);
                error!(
                    "Instruments catchup replay failed: {} - marking as ready with potentially stale state",
                    e
                );
                // Still mark as ready so service can start, but log the error
                instruments_cache.sync_status().set_ready();
            }
        }
    }

    // Start instruments snapshot task for periodic snapshots
    if !standby_mode_active {
        let instruments_cache_for_snapshot = instruments_cache.clone();
        let instruments_get_offsets = move || instruments_cache_for_snapshot.snapshot_offsets();
        let instruments_snapshot_writer = Arc::new(DbInstrumentsSnapshotWriter::new(
            shared_db.clone(),
            instruments_cache.clone(),
            instruments_get_offsets,
        ));
        let instruments_snapshot_task = Arc::new(InstrumentsSnapshotTask::new(
            instruments_snapshot_writer,
            SnapshotTaskConfig::default(),
        ));
        service_registry.register(instruments_snapshot_task);
        info!("Instruments snapshot task registered (60s interval)");
    } else {
        info!("Instruments snapshot task deferred (standby mode)");
    }

    let catalog_config = runtime.config.catalog.clone();
    info!(
        "Loaded inline catalog config version {} ({} underlyings)",
        catalog_config.version,
        catalog_config.underlyings.len()
    );
    let collateral_registry = Arc::new(
        catalog_manager::CollateralRegistry::from_catalog(&catalog_config)
            .context("Failed to build collateral registry from catalog config")?,
    );

    // Resolve oracle markets from catalog (with optional env overrides).
    let oracle_markets = oracles::configured_oracle_markets(&catalog_config);
    let underlying_to_candle_coin: Arc<HashMap<String, String>> = Arc::new(
        oracle_markets
            .iter()
            .map(|market| (market.underlying.clone(), market.hyperliquid_coin.clone()))
            .collect(),
    );
    let candle_source: Arc<dyn UnderlyingCandleSource> = Arc::new(
        HyperliquidCandleSource::from_config(
            &runtime.config.pricing,
            runtime.config.modes.testnet_mode,
        )
        .map_err(|e| anyhow::anyhow!("Failed to initialize HyperliquidCandleSource: {}", e))?,
    );

    // Build WsFeed symbol list from resolved oracle markets.
    let mut ws_symbols = Vec::new();
    let mut seen_symbols = HashSet::new();
    for market in &oracle_markets {
        if seen_symbols.insert(market.hyperliquid_coin.clone()) {
            ws_symbols.push(market.hyperliquid_coin.clone());
        }
    }

    let ws_url = runtime.config.pricing.hyperliquid_ws_url.clone();
    let ws_feed = {
        #[cfg(feature = "test-endpoints")]
        {
            if runtime.config.modes.skip_external_oracle {
                let mut default_prices = HashMap::new();
                default_prices.insert("BTC".to_string(), 90_000.0);
                default_prices.insert("ETH".to_string(), 2_500.0);
                default_prices.insert("HYPE".to_string(), 42.0);
                default_prices.insert("km:US500".to_string(), 5_500.0);
                default_prices.insert("km:USOIL".to_string(), 65.0);
                oracles::apply_mock_oracle_price_overrides(&mut default_prices);
                let feed = Arc::new(HyperliquidWsFeed::new_mock(ws_symbols, &default_prices));
                let feed_clone = feed.clone();
                let mock_shutdown_rx = shutdown.subscribe();
                task_group.spawn("MockWsFeedRefresh", async move {
                    let mut interval = tokio::time::interval(Duration::from_secs(5));
                    let mut shutdown = mock_shutdown_rx;
                    loop {
                        tokio::select! {
                            _ = interval.tick() => feed_clone.refresh_mock_timestamps().await,
                            _ = shutdown.recv() => break,
                        }
                    }
                    Ok(())
                });
                info!("HyperliquidWsFeed mock with default prices (skip_external_oracle=true)");
                feed
            } else {
                let feed = Arc::new(HyperliquidWsFeed::new(ws_url.clone(), ws_symbols));
                let ws_feed_clone = feed.clone();
                let ws_shutdown_rx = shutdown.subscribe();
                task_group.spawn("HyperliquidWsFeed", async move {
                    ws_feed_clone.start(ws_shutdown_rx).await;
                    Ok(())
                });
                info!("HyperliquidWsFeed live mode enabled (test-endpoints build)");
                feed
            }
        }
        #[cfg(not(feature = "test-endpoints"))]
        {
            let feed = Arc::new(HyperliquidWsFeed::new(ws_url.clone(), ws_symbols));
            let ws_feed_clone = feed.clone();
            let ws_shutdown_rx = shutdown.subscribe();
            task_group.spawn("HyperliquidWsFeed", async move {
                ws_feed_clone.start(ws_shutdown_rx).await;
                Ok(())
            });
            info!("HyperliquidWsFeed started (url={})", ws_url);
            feed
        }
    };

    let hydromancer_client =
        if runtime.config.hydromancer.enabled && !cfg!(feature = "test-endpoints") {
            let api_key = runtime
                .secrets
                .require_hydromancer_api_key()
                .expect("Hydromancer is enabled but HYDROMANCER_API_KEY is missing")
                .to_string();
            let config = HydromancerConfig {
                api_url: runtime.config.hydromancer.api_url.clone(),
                api_key,
            };
            info!("Hydromancer fallback oracle enabled");
            Some(Arc::new(HydromancerClient::new(config)))
        } else {
            info!("Hydromancer fallback oracle disabled");
            None
        };

    // Initialize mark price oracles for each underlying.
    let t = Instant::now();
    let spot_price_notify = Arc::new(tokio::sync::Notify::new());
    let mut mark_price_oracles: HashMap<String, Arc<HyperliquidMarkPriceOracle>> = HashMap::new();
    for market in &oracle_markets {
        let t_oracle = Instant::now();
        let oracle_config = HyperliquidOracleConfig {
            symbol: market.hyperliquid_coin.clone(),
            oracle_writer: Some(shared_db.clone()),
            ws_feed: Some(ws_feed.clone()),
            price_notify: Some(spot_price_notify.clone()),
            ..Default::default()
        };
        let mut oracle = {
            #[cfg(feature = "test-endpoints")]
            {
                HyperliquidMarkPriceOracle::new(oracle_config).map_err(|e| {
                    anyhow::anyhow!(
                        "Failed to construct mark price oracle {}: {}",
                        market.underlying,
                        e
                    )
                })?
            }
            #[cfg(not(feature = "test-endpoints"))]
            {
                match HyperliquidMarkPriceOracle::new_with_init(oracle_config.clone()).await {
                    Ok(oracle) => oracle,
                    Err(err) => {
                        error!(
                            "Failed to initialize mark price oracle for {} (coin: {}): {}. Starting without initial price.",
                            market.underlying, market.hyperliquid_coin, err
                        );
                        HyperliquidMarkPriceOracle::new(oracle_config).map_err(|e| {
                            anyhow::anyhow!(
                                "Failed to construct mark price oracle {}: {}",
                                market.underlying,
                                e
                            )
                        })?
                    }
                }
            }
        };
        if let Some(ref hc) = hydromancer_client {
            oracle.set_hydromancer_client(hc.clone());
        }
        startup_timing!("oracle_init", t_oracle.elapsed(), "{}", market.underlying);
        gauge!("ht_startup_oracle_seconds", "underlying" => market.underlying.clone())
            .set(t_oracle.elapsed().as_secs_f64());
        let oracle = Arc::new(oracle);
        mark_price_oracles.insert(market.underlying.clone(), oracle);
    }
    startup_timing!("oracles_total", t.elapsed());

    // Start polling tasks for each oracle.
    for market in &oracle_markets {
        let oracle_clone = mark_price_oracles
            .get(&market.underlying)
            .unwrap_or_else(|| {
                panic!(
                    "Mark price oracle missing for {} after initialization",
                    market.underlying
                )
            })
            .clone();
        let oracle_shutdown_rx = shutdown.subscribe();
        let task_name = format!("MarkPriceOracle-{}", market.underlying);
        let task_name_static = Box::leak(task_name.into_boxed_str());
        task_group.spawn(task_name_static, async move {
            let handle = oracle_clone.start_polling_with_shutdown(oracle_shutdown_rx);
            if let Err(e) = handle.await {
                error!("MarkPriceOracle task panicked: {:?}", e);
            }
            Ok(())
        });
        info!(
            "Initialized mark price oracle for {} (coin: {})",
            market.underlying, market.hyperliquid_coin
        );
    }

    // Clone for CatalogManager before moving to engine builder
    let mark_price_oracles_for_catalog = mark_price_oracles.clone();
    info!(
        "Mark price oracles initialized for: {:?}",
        mark_price_oracles.keys().collect::<Vec<_>>()
    );

    let read_snapshot = std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(
        crate::rsm::engine_snapshot::EngineSnapshot::empty(),
    ));
    let snapshot_provider = std::sync::Arc::new(
        crate::rsm::engine_snapshot::SnapshotQuoteProvider::new(read_snapshot.clone()),
    );
    let quote_provider: std::sync::Arc<dyn hypercall_runtime_api::QuoteProvider> =
        snapshot_provider.clone();
    let engine_state_digest_provider: std::sync::Arc<
        dyn crate::rsm::engine_snapshot::EngineStateDigestProvider,
    > = snapshot_provider.clone();
    let balance_provider: std::sync::Arc<dyn crate::rsm::ledger::BalanceProvider> =
        snapshot_provider.clone();
    let order_snapshot: std::sync::Arc<dyn hypercall_runtime_api::OrderSnapshotProvider> =
        snapshot_provider.clone();
    let agent_auth: std::sync::Arc<dyn hypercall_runtime_api::AgentAuthProvider> =
        snapshot_provider.clone();
    // Initialize GreeksCache with quote provider for on-demand mid price reads
    let t = Instant::now();
    let greeks_cache = GreeksCache::new(
        diesel_db.as_ref(),
        event_bus.clone(),
        mark_price_oracles.clone(),
        shutdown.subscribe(),
        quote_provider.clone(),
    )
    .await
    .map_err(|e| anyhow::anyhow!("Failed to create GreeksCache: {}", e))?;
    startup_timing!("greeks_cache", t.elapsed());
    let competition_service = Arc::new(CompetitionService::new(
        diesel_db.clone(),
        greeks_cache.clone(),
    ));

    // Initialize MmpCache with the shared handler and GreeksCache
    let t = Instant::now();
    let mmp_cache = Arc::new(crate::read_cache::mmp::MmpCache::new(
        shared_db.clone(),
        greeks_cache.clone(),
    )?);
    mmp_cache.load_configs_from_db().await?;
    startup_timing!("mmp_cache", t.elapsed());
    service_registry.register(mmp_cache.clone());

    // TierCache is already initialized earlier (before catchup replay)

    // WsEventForwarder reads from the direct channel for low-latency WebSocket updates.
    // The sender half (ws_direct_tx) is wired to the engine builder below.
    let ws_receiver = ws_direct_rx_for_forwarder;

    // Startup catchup: reload TierCache to catch any tier changes that
    // occurred between initial load and engine start.
    // Version-based ordering ensures stale messages will be ignored.
    tier_cache.load_from_db().await?;
    debug!("TierCache startup catchup complete");

    // Initialize push notification service (optional, requires VAPID_PRIVATE_KEY_PEM env var)
    let push_service: Option<Arc<hypercall_api::push_service::PushNotificationService>> =
        match std::env::var("VAPID_PRIVATE_KEY_PEM") {
            Ok(pem_b64) => {
                let pem_b64_clean: String =
                    pem_b64.chars().filter(|c| !c.is_whitespace()).collect();
                match base64::Engine::decode(
                    &base64::engine::general_purpose::STANDARD,
                    &pem_b64_clean,
                ) {
                    Ok(pem_bytes) => {
                        match hypercall_api::push_service::PushNotificationService::new(
                            diesel_db.clone(),
                            &pem_bytes,
                        ) {
                            Ok(svc) => {
                                tracing::info!("✓ PushNotificationService initialized");
                                Some(Arc::new(svc))
                            }
                            Err(e) => {
                                tracing::warn!("Push notification service init failed: {e}");
                                None
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!("VAPID_PRIVATE_KEY_PEM base64 decode failed: {e}");
                        None
                    }
                }
            }
            Err(_) => {
                tracing::info!("VAPID_PRIVATE_KEY_PEM not set, push notifications disabled");
                None
            }
        };

    // Initialize persisted notification feed service.
    //
    // Postgres is the source of truth — feed writes/reads work without Redis.
    // Redis, when available, caches a per-wallet `has_unread` bit for the
    // edge-served bell dot. The service always initializes; the Redis client
    // is optional and populated only when we get a valid `redis://` / `rediss://`
    // URL. REST-only URLs (`https://…`) aren't usable from the `redis` crate
    // and are ignored with a warning rather than silently disabling the feed.
    let notification_service: Arc<hypercall_api::notification_service::NotificationService> = {
        let redis_client_opt: Option<redis::Client> = match std::env::var("UPSTASH_REDIS_URL")
            .ok()
            .or_else(|| std::env::var("REDIS_URL").ok())
        {
            Some(url) if !url.trim().is_empty() => {
                let trimmed = url.trim();
                if trimmed.starts_with("redis://") || trimmed.starts_with("rediss://") {
                    match redis::Client::open(trimmed.to_string()) {
                        Ok(c) => {
                            tracing::info!("✓ NotificationService Redis cache initialized");
                            Some(c)
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                "NotificationService: Redis client open failed; feed persistence stays on, bell cache disabled"
                            );
                            None
                        }
                    }
                } else {
                    tracing::warn!(
                        "NotificationService: UPSTASH_REDIS_URL / REDIS_URL must be a redis:// or rediss:// URI (got scheme mismatch); feed persistence stays on, bell cache disabled"
                    );
                    None
                }
            }
            _ => {
                tracing::info!(
                    "NotificationService: no Redis URL configured (UPSTASH_REDIS_URL / REDIS_URL), bell cache disabled; feed persistence still active"
                );
                None
            }
        };
        hypercall_api::notification_service::NotificationService::new(
            diesel_db.clone(),
            redis_client_opt,
            shutdown.subscribe(),
        )
    };

    // Forward events to WebSocket PubSubManager, PortfolioCache, and Persistence
    let ws_forwarder_deps = hypercall_api::websocket::event_forwarder::WsEventForwarderDeps {
        pubsub: pubsub.clone(),
        portfolio_cache: portfolio_cache.clone(),
        instruments_cache: instruments_cache.clone(),
        quote_provider: quote_provider.clone(),
        greeks_cache: greeks_cache.clone(),
        tier_cache: tier_cache.clone(),
        push_service: push_service.clone(),
        db: Arc::new(
            crate::db_handler::DieselEventHandler::with_pool_no_migrations(shared_db.pool()),
        ),
        engine_event_tx: engine_event_tx.clone(),
        notification_service: notification_service.clone(),
        shutdown: shutdown.clone(),
    };
    task_group.spawn("WsEventForwarder", async move {
        hypercall_api::websocket::event_forwarder::run(ws_forwarder_deps, ws_receiver).await
    });

    // Competition fill recording is an independent downstream consumer of the
    // engine fill stream, not a side effect of websocket broadcast. It subscribes
    // to the EventBus TOPIC_FILLS fan-out (the same generic mechanism the
    // HyperCore forwarder and other downstream services use); when the engine
    // runs out-of-process this receiver is fed by a NATS consumer instead.
    let competition_fill_rx = event_bus
        .subscribe(vec![hypercall_types::topics::TOPIC_FILLS.to_string()])
        .await
        .map_err(|e| anyhow::anyhow!("Failed to subscribe competition fill recorder: {}", e))?;
    let competition_fill_recorder =
        hypercall_competition::CompetitionFillRecorder::new(competition_service.clone());
    let competition_fill_shutdown = shutdown.subscribe();
    task_group.spawn("CompetitionFillRecorder", async move {
        competition_fill_recorder
            .run(competition_fill_rx, competition_fill_shutdown)
            .await;
        Ok(())
    });

    // Get the portfolio service from the cache for wiring into the engine
    // This ensures UnifiedEngine reads executed state from the canonical source
    let portfolio_service = portfolio_cache.get_service();

    // Initialize MarketStatsCache with DB backfill for last 24h
    let market_stats_cache = Arc::new(crate::read_cache::market_stats::MarketStatsCache::new(
        portfolio_service.clone(),
    ));
    market_stats_cache
        .clone()
        .initialize_with_backfill(diesel_db.as_ref(), event_bus.clone(), shutdown.subscribe())
        .await?;
    tracing::info!("✓ MarketStatsCache initialized");

    let markets_snapshot_refresh_ms = runtime.config.api.markets_snapshot_refresh_ms;
    let markets_snapshot_cache = Arc::new(
        hypercall_api::caches::MarketsSnapshotCache::new(
            instruments_cache.clone(),
            greeks_cache.clone(),
            market_stats_cache.clone(),
        )
        .with_refresh_interval(Duration::from_millis(markets_snapshot_refresh_ms)),
    );
    markets_snapshot_cache.initialize().await;
    service_registry.register(markets_snapshot_cache.clone());
    tracing::info!(
        "✓ MarketsSnapshotCache initialized (refresh_ms={})",
        markets_snapshot_refresh_ms
    );

    let instruments_snapshot_refresh_ms =
        services::parse_positive_refresh_ms_env("INSTRUMENTS_SNAPSHOT_REFRESH_MS", 10_000);
    let instruments_snapshot_cache = Arc::new(
        hypercall_api::caches::InstrumentsSnapshotCache::new(instruments_cache.clone())
            .with_refresh_interval(Duration::from_millis(instruments_snapshot_refresh_ms)),
    );
    instruments_snapshot_cache.initialize().await;
    service_registry.register(instruments_snapshot_cache.clone());
    tracing::info!(
        "✓ InstrumentsSnapshotCache initialized (refresh_ms={})",
        instruments_snapshot_refresh_ms
    );

    // --- Competitions Snapshot Cache ---
    let competitions_snapshot_refresh_ms =
        services::parse_positive_refresh_ms_env("COMPETITIONS_SNAPSHOT_REFRESH_MS", 10_000);
    let competitions_snapshot_cache = Arc::new(
        hypercall_api::caches::CompetitionsSnapshotCache::new(competition_service.clone())
            .with_refresh_interval(Duration::from_millis(competitions_snapshot_refresh_ms)),
    );
    competitions_snapshot_cache.initialize().await;
    service_registry.register(competitions_snapshot_cache.clone());
    tracing::info!(
        "✓ CompetitionsSnapshotCache initialized (refresh_ms={})",
        competitions_snapshot_refresh_ms
    );

    // --- Sparklines Snapshot Cache ---
    let sparklines_snapshot_refresh_ms =
        services::parse_positive_refresh_ms_env("SPARKLINES_SNAPSHOT_REFRESH_MS", 15_000);
    let sparklines_snapshot_cache = Arc::new(
        hypercall_api::caches::SparklinesSnapshotCache::from_config(
            &runtime.config.pricing,
            runtime.config.modes.testnet_mode,
        )
        .map_err(|e| anyhow::anyhow!("Failed to create SparklinesSnapshotCache: {}", e))?
        .with_refresh_interval(Duration::from_millis(sparklines_snapshot_refresh_ms)),
    );
    sparklines_snapshot_cache.initialize().await;
    service_registry.register(sparklines_snapshot_cache.clone());
    tracing::info!(
        "✓ SparklinesSnapshotCache initialized (refresh_ms={})",
        sparklines_snapshot_refresh_ms
    );

    let mut hypercore_event_rx = event_bus
        .subscribe(vec![
            crate::shared::topics::TOPIC_HYPERCORE_POSITION_UPDATES.to_string(),
        ])
        .await
        .map_err(|e| anyhow::anyhow!("Failed to subscribe to HyperCore updates: {}", e))?;
    let hypercore_ws_tx = ws_direct_tx.clone();
    let mut hypercore_bridge_shutdown_rx = shutdown.subscribe();
    task_group.spawn("HypercoreEventForwarder", async move {
        loop {
            let event = tokio::select! {
                _ = hypercore_bridge_shutdown_rx.recv() => {
                    info!("HypercoreEventForwarder received shutdown signal");
                    break;
                }
                maybe_event = hypercore_event_rx.recv() => {
                    match maybe_event {
                        Some(event) => event,
                        None => break,
                    }
                }
            };

            if let EngineMessage::HypercorePositionUpdate(_) = &event {
                if hypercore_ws_tx.send(event).is_err() {
                    warn!("HypercoreEventForwarder failed to forward HyperCore update");
                    break;
                }
            }
        }

        Ok(())
    });

    let (deposit_tx, deposit_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::DepositRequest>(128);

    // NOW start event bus processing – ALL subscriptions are complete
    // CRITICAL: This MUST happen after all subscribe() calls so no events are missed
    let t = Instant::now();
    event_bus.clone().start_processing().await;
    startup_timing!("event_bus_start_processing", t.elapsed());

    let vol_oracle_output = VolOracleFactory::build(&catalog_config, &mut task_group, &shutdown)
        .await
        .map_err(|err| anyhow::anyhow!("Failed to build vol oracle router: {err:#}"))?;
    let risk_vol_oracle = vol_oracle_output.oracle;

    // Wire vol oracle into GreeksCache so PM repricing can fall back to
    // surface-interpolated IV when orderbooks are empty (e.g. after fresh deploy).
    greeks_cache.set_vol_oracle(risk_vol_oracle.clone()).await;

    // Feed platform spot prices into the Polygon vol oracle so it can
    // compute dynamic strike scales (platform_spot / etf_spot).
    {
        let gc = greeks_cache.clone();
        let spots = vol_oracle_output.polygon_platform_spots;
        let underlyings: Vec<String> = catalog_config.vol_oracles.routes.keys().cloned().collect();
        let mut feeder_shutdown = shutdown.subscribe();
        task_group.spawn("polygon_spot_feeder", async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
            loop {
                tokio::select! {
                    _ = interval.tick() => {}
                    _ = feeder_shutdown.recv() => break,
                }
                let mut updates: Vec<(String, Option<f64>)> = Vec::new();
                for underlying in &underlyings {
                    updates.push((underlying.clone(), gc.get_spot_price(underlying).await));
                }
                let mut map = spots.write().expect("platform_spots poisoned");
                for (underlying, price) in updates {
                    if let Some(p) = price {
                        map.insert(underlying, p);
                    } else {
                        map.remove(&underlying);
                    }
                }
            }
            Ok(())
        });
    }

    // Readiness checks are collected here but the registry is built after
    // the engine, so we can include the engine's sync_status. This ensures
    // the API rejects requests until post-startup reconciliation completes.
    let mut readiness_checks: Vec<Arc<dyn crate::readiness::Readiness>> = vec![
        Arc::new(SyncStatusReadiness::new(
            "portfolio",
            portfolio_cache.sync_status(),
        )),
        Arc::new(SyncStatusReadiness::new(
            "instruments",
            instruments_cache.sync_status(),
        )),
    ];
    if runtime.config.api.require_risk_vol_readiness {
        readiness_checks.push(Arc::new(VolOracleReadiness::new(
            "risk_vol_oracle",
            risk_vol_oracle.clone(),
        )));
    }

    let _hypercore_position_service =
        if std::env::var("ENABLE_INTEGRATED_HYPERCORE_POSITION_SERVICE").is_ok() {
            let update_interval_secs = std::env::var("HYPERCORE_UPDATE_INTERVAL_SECS")
                .ok()
                .and_then(|raw| raw.parse::<u64>().ok())
                .unwrap_or(30);
            let service = Arc::new(HypercorePositionService::with_default_url(
                Duration::from_secs(update_interval_secs),
                event_bus.clone(),
                shared_db.clone(),
            ));
            let service_for_task = service.clone();
            let hypercore_shutdown = shutdown.subscribe();
            task_group.spawn("HypercorePositionService", async move {
                service_for_task
                    .run_with_shutdown(hypercore_shutdown)
                    .await
                    .map_err(|e| anyhow::anyhow!("HyperCore position service failed: {}", e))
            });
            tracing::info!(
                "✓ HyperCore position service started inside integrated server (interval={}s)",
                update_interval_secs
            );
            Some(service)
        } else {
            None
        };

    // Create SpanMarginService for API use (stateless computation, separate from engine's instance)
    // This allows the API layer to compute SPAN margin without going through the engine
    let span_margin_service = Arc::new(SpanMarginService::new_with_vol_oracle(
        config.clone(),
        risk_vol_oracle.clone(),
    ));
    tracing::info!("✓ SpanMarginService initialized for API layer");

    // Create RiskAccountBuilder for API risk calculations
    // Uses the engine snapshot balance ledger published by UnifiedEngine.
    let snapshot_open_orders = Arc::new(
        crate::rsm::engine_snapshot::SnapshotOpenOrdersSource::new(order_snapshot.clone()),
    );
    let risk_account_builder = Arc::new(RiskAccountBuilder::new_with_balance_provider(
        balance_provider.clone(),
        portfolio_service.clone(),
        snapshot_open_orders,
        greeks_cache.clone(),
    ));
    tracing::info!("✓ RiskAccountBuilder initialized for API layer");

    // Create StandardMarginService for Deribit-style linear margin calculations
    let standard_margin_service = Arc::new(StandardMarginService::new());
    tracing::info!("✓ StandardMarginService initialized for API layer");

    // Create StandardAccountBuilder for Standard margin mode accounts
    let standard_account_builder = Arc::new(StandardAccountBuilder::new_with_balance_provider(
        balance_provider.clone(),
        portfolio_service.clone(),
        greeks_cache.clone(),
    ));
    tracing::info!("✓ StandardAccountBuilder initialized for API layer");

    // Wire margin service + risk account builder into PortfolioCache for real-time WS margin updates
    portfolio_cache
        .set_margin_dependencies(
            span_margin_service.clone(),
            greeks_cache.clone(),
            risk_account_builder.clone(),
            tier_cache.clone(),
            order_snapshot.clone(),
            standard_margin_service.clone(),
            standard_account_builder.clone(),
        )
        .await;

    // After all dependencies are wired, reprice all positions loaded from snapshot/replay.
    // Positions replayed from journal have unrealized_pnl=0 (stale). Repricing here ensures
    // the first /portfolio GET returns correct equity instead of inflated values.
    {
        let t = Instant::now();
        let repriced = portfolio_cache.reprice_all_wallets().await;
        startup_timing!("portfolio_reprice_all", t.elapsed(), "{} wallets", repriced);
    }

    // Start DB-writing snapshot tasks (deferred in standby — readonly pool)
    if !standby_mode_active {
        let historical_pnl_config =
            HistoricalPnlTaskConfig::from_config(&runtime.config.background_tasks.historical_pnl);
        let historical_pnl_task = HistoricalPnlSnapshotTask::new(
            portfolio_cache.clone(),
            diesel_db.clone(),
            historical_pnl_config.clone(),
        );
        service_registry.register(Arc::new(historical_pnl_task));
        info!(
            "Historical pnl snapshot task registered (max_periods={}, capture_every_5m_ms={}, capture_every_1h_ms={}, capture_every_1d_ms={})",
            historical_pnl_config.max_periods,
            historical_pnl_config.capture_every_5m_ms,
            historical_pnl_config.capture_every_1h_ms,
            historical_pnl_config.capture_every_1d_ms
        );

        let historical_theo_config = HistoricalTheoTaskConfig::from_env();
        let historical_theo_task = HistoricalTheoSnapshotTask::new(
            instruments_cache.clone(),
            greeks_cache.clone(),
            diesel_db.clone(),
            historical_theo_config.clone(),
        );
        service_registry.register(Arc::new(historical_theo_task));
        info!(
            "Historical theo snapshot task registered (max_periods={}, capture_every_5m_ms={}, capture_every_1h_ms={}, capture_every_1d_ms={})",
            historical_theo_config.max_periods,
            historical_theo_config.capture_every_5m_ms,
            historical_theo_config.capture_every_1h_ms,
            historical_theo_config.capture_every_1d_ms
        );

        let vol_surface_config =
            crate::runtime::tasks::vol_surface_snapshot::VolSurfaceSnapshotConfig::from_env();
        let vol_surface_task =
            crate::runtime::tasks::vol_surface_snapshot::VolSurfaceSnapshotTask::new(
                risk_vol_oracle.clone(),
                diesel_db.clone(),
                vol_surface_config.clone(),
            );
        service_registry.register(Arc::new(vol_surface_task));
        info!(
            "Vol surface snapshot task registered (max_periods={}, capture_every_5m_ms={}, capture_every_1h_ms={}, capture_every_1d_ms={})",
            vol_surface_config.max_periods,
            vol_surface_config.capture_every_5m_ms,
            vol_surface_config.capture_every_1h_ms,
            vol_surface_config.capture_every_1d_ms
        );
    } else {
        info!("Historical pnl/theo/vol-surface snapshot tasks deferred (standby mode)");
    }

    // Start persisted BBO snapshot task for /options-summary 24h price_change lookup.
    let bbo_snapshot_config =
        BboSnapshotTaskConfig::from_config(&runtime.config.background_tasks.bbo_snapshot);
    let bbo_snapshot_service = Arc::new(BboSnapshotService::new(
        instruments_cache.clone(),
        quote_provider.clone(),
        diesel_db.clone(),
        bbo_snapshot_config.clone(),
    ));
    if !standby_mode_active {
        service_registry.register(bbo_snapshot_service.clone());
        info!(
            "BBO snapshot task registered (interval_secs={}, retention_days={})",
            bbo_snapshot_config.interval_secs, bbo_snapshot_config.retention_days
        );
    } else {
        info!("BBO snapshot task deferred (standby mode)");
    }

    let options_summary_snapshot_refresh_ms = services::parse_positive_refresh_ms_env(
        "OPTIONS_SUMMARY_SNAPSHOT_REFRESH_MS",
        markets_snapshot_refresh_ms,
    );
    let indicative_cache =
        Arc::new(hypercall_api::rfq::indicative_quote_cache::IndicativeQuoteCache::new(60_000));
    let options_summary_snapshot_cache = Arc::new(
        hypercall_api::caches::OptionsSummarySnapshotCache::new(
            instruments_cache.clone(),
            greeks_cache.clone(),
            quote_provider.clone(),
            bbo_snapshot_service.clone(),
            Some(indicative_cache.clone()),
            diesel_db.clone(),
        )
        .with_refresh_interval(Duration::from_millis(options_summary_snapshot_refresh_ms)),
    );
    options_summary_snapshot_cache.initialize().await;
    service_registry.register(options_summary_snapshot_cache.clone());
    tracing::info!(
        "✓ OptionsSummarySnapshotCache initialized (refresh_ms={})",
        options_summary_snapshot_refresh_ms
    );

    // --- Upstash Batch Publisher ---
    let upstash_init = hypercall_api::upstash::init(
        &markets_snapshot_cache,
        &competitions_snapshot_cache,
        &sparklines_snapshot_cache,
        &instruments_snapshot_cache,
        &options_summary_snapshot_cache,
        diesel_db.clone(),
    )
    .await?;
    if let Some(publisher) = upstash_init.publisher {
        service_registry.register(publisher);
    }
    let username_redis_client = upstash_init.redis_client;

    // Periodically refresh WS portfolio margin for subscribed wallets so
    // idle accounts still receive market-driven updates.
    // Periodically refresh WS portfolio margin and greeks for subscribed wallets
    // so idle accounts still receive market-driven updates.
    let periodic_margin_cache = portfolio_cache.clone();
    let mut periodic_margin_shutdown_rx = shutdown.subscribe();
    task_group.spawn("PortfolioMarginWsRefresh", async move {
        let mut interval = tokio::time::interval(Duration::from_secs(5));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                _ = periodic_margin_shutdown_rx.recv() => {
                    info!("PortfolioMarginWsRefresh received shutdown signal");
                    break;
                }
                _ = interval.tick() => {
                    let position_wallets = periodic_margin_cache
                        .publish_position_updates_for_subscribers()
                        .await;
                    let wallets = periodic_margin_cache
                        .publish_margin_updates_for_subscribers()
                        .await;
                    let greek_wallets = periodic_margin_cache
                        .publish_greeks_updates_for_subscribers()
                        .await;
                    tracing::debug!(
                        "PortfolioMarginWsRefresh published position updates for {} wallets, margin updates for {} wallets, and greeks updates for {} wallets",
                        position_wallets,
                        wallets,
                        greek_wallets
                    );
                }
            }
        }
        Ok(())
    });

    // Periodic competition updates + finalization notifications for subscribed wallets.
    let competition_service_for_ws = competition_service.clone();
    let competition_pubsub = pubsub.clone();
    let mut competition_ws_shutdown_rx = shutdown.subscribe();
    task_group.spawn("CompetitionWsRefresh", async move {
        let mut interval = tokio::time::interval(Duration::from_secs(5));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut last_rank_by_wallet: HashMap<WalletAddress, (i64, i64)> = HashMap::new();
        let mut last_gap_by_wallet: HashMap<WalletAddress, (i64, Option<i64>, Option<rust_decimal::Decimal>)> =
            HashMap::new();

        loop {
            tokio::select! {
                _ = competition_ws_shutdown_rx.recv() => {
                    info!("CompetitionWsRefresh received shutdown signal");
                    break;
                }
                _ = interval.tick() => {
                    let now_ts_ms = chrono::Utc::now().timestamp_millis();

                    match competition_service_for_ws.finalize_ended_competitions(now_ts_ms).await {
                        Ok(final_stats) => {
                            for stats in final_stats {
                                competition_pubsub
                                    .publish_competition_final_stats(stats.clone())
                                    .await;
                                competition_pubsub
                                    .publish_competition_final_standing(WsCompetitionFinalStanding {
                                        wallet_address: stats.wallet_address,
                                        competition_id: stats.competition_id,
                                        rank: stats.rank,
                                        pnl: stats.pnl,
                                        volume: stats.volume,
                                        efficiency: stats.efficiency,
                                        medal: stats.medal,
                                        timestamp: stats.timestamp,
                                    })
                                    .await;
                            }
                        }
                        Err(err) => {
                            error!("Competition finalization failed: {}", err);
                        }
                    }

                    let wallets = competition_pubsub.subscribed_wallets("competition").await;
                    for wallet in wallets {
                        match competition_service_for_ws
                            .build_competition_update_for_wallet(wallet, now_ts_ms)
                            .await
                        {
                            Ok(Some(update)) => {
                                competition_pubsub.publish_competition_update(update).await;
                            }
                            Ok(None) => {}
                            Err(err) => {
                                error!(
                                    "Failed to build competition update for wallet {}: {}",
                                    wallet, err
                                );
                            }
                        }
                    }

                    let engagement_wallets = competition_pubsub
                        .subscribed_wallets("competition_engagement")
                        .await;
                    for wallet in engagement_wallets.iter().copied() {
                        match competition_service_for_ws
                            .build_competition_engagement_snapshot_for_wallet(wallet, now_ts_ms)
                            .await
                        {
                            Ok(Some(snapshot)) => {
                                if let Some((prev_competition_id, prev_rank)) =
                                    last_rank_by_wallet.get(&wallet).copied()
                                {
                                    if prev_competition_id == snapshot.competition_id
                                        && prev_rank != snapshot.rank
                                    {
                                        competition_pubsub
                                            .publish_competition_rank_change(WsCompetitionRankChange {
                                                wallet_address: wallet,
                                                competition_id: snapshot.competition_id,
                                                from_rank: prev_rank,
                                                to_rank: snapshot.rank,
                                                delta_places: prev_rank - snapshot.rank,
                                                pnl: snapshot.pnl,
                                                timestamp: now_ts_ms,
                                            })
                                            .await;
                                    }
                                }

                                let gap_state = (
                                    snapshot.competition_id,
                                    snapshot.next_rank,
                                    snapshot.gap_metric_value,
                                );
                                let should_publish_gap = last_gap_by_wallet
                                    .get(&wallet)
                                    .map(|prev| prev != &gap_state)
                                    .unwrap_or(true);
                                if should_publish_gap {
                                    competition_pubsub
                                        .publish_competition_gap_update(WsCompetitionGapUpdate {
                                            wallet_address: wallet,
                                            competition_id: snapshot.competition_id,
                                            rank: snapshot.rank,
                                            next_rank: snapshot.next_rank,
                                            gap_metric_value: snapshot.gap_metric_value,
                                            timestamp: now_ts_ms,
                                        })
                                        .await;
                                }

                                last_rank_by_wallet
                                    .insert(wallet, (snapshot.competition_id, snapshot.rank));
                                last_gap_by_wallet.insert(wallet, gap_state);
                            }
                            Ok(None) => {
                                last_rank_by_wallet.remove(&wallet);
                                last_gap_by_wallet.remove(&wallet);
                            }
                            Err(err) => {
                                error!(
                                    "Failed to build competition engagement snapshot for wallet {}: {}",
                                    wallet, err
                                );
                            }
                        }
                    }

                    let engagement_wallets_set =
                        engagement_wallets.iter().copied().collect::<std::collections::HashSet<_>>();
                    last_rank_by_wallet.retain(|wallet, _| engagement_wallets_set.contains(wallet));
                    last_gap_by_wallet.retain(|wallet, _| engagement_wallets_set.contains(wallet));
                }
            }
        }
        Ok(())
    });

    // Ledger and tier_cache already wired to PortfolioService earlier (before catchup replay)
    tracing::info!(
        "✓ Wired SpanMarginService + RiskAccountBuilder into PortfolioCache for WS margin updates"
    );

    // --- Standby mode setup ---
    // If STANDBY_MODE is set, the engine, journal batcher, and background write tasks
    // are deferred until POST /admin/promote is called. The HTTP server starts normally
    // so /standby-ready and /admin/promote are accessible.
    // Note: standby_mode_active is resolved near the top of this function (pool creation).
    let standby_mode = standby_mode_active;

    let (standby_controller, _standby_promote_rx) = if standby_mode {
        let (controller, promote_rx) =
            crate::runtime::standby::StandbyController::new_standby(1000);
        info!("Standby mode enabled — warmup complete, engine deferred until POST /admin/promote");
        (Some(controller), Some(promote_rx))
    } else {
        (None, None)
    };

    let rsm_signer_configured = runtime.config.rsm_signer.is_configured();
    let local_rsm_signer_configured = runtime.config.transaction_submitter.mode
        == hypercall_transaction_submitter::TransactionSubmitterMode::Direct;
    // Deposit crediting is not a separately toggled feature. If this process is
    // connected to chain, on-chain Exchange.Deposit events must become engine
    // credits. Deposits do not need the RSM signer because the user already
    // burned custody tokens in the Exchange contract.
    let onchain_deposits_enabled = true;

    let rsm_signer_service: Option<Arc<dyn hypercall_signer::RsmSigner>> = if runtime
        .config
        .liquidation
        .enabled
        || rsm_signer_configured
        || local_rsm_signer_configured
    {
        if let Some(controller) = standby_controller.clone() {
            runtime
                .secrets
                .database_url
                .as_ref()
                .context("standby RSM signer requires DATABASE_URL for post-promote init")?;
            Some(
                Arc::new(crate::startup::rsm_signer::StandbyRsmSignerService::new(
                    controller,
                    runtime.config.clone(),
                    runtime.secrets.clone(),
                    db_auth.clone(),
                    rsm_signer_chain_id,
                )) as Arc<dyn hypercall_signer::RsmSigner>,
            )
        } else {
            Some(
                crate::startup::rsm_signer::build_rsm_signer_service(
                    &runtime.config,
                    &runtime.secrets,
                    diesel_db.clone(),
                    rsm_signer_chain_id,
                )
                .await
                .map_err(|error| anyhow::anyhow!("Failed to initialize RSM signer: {}", error))?,
            )
        }
    } else {
        None
    };

    // (Journal + batcher construction moved up to before the runtime balance
    // ledger and portfolio-catchup initializations. `engine_journal_writer` and
    // `journal_batch_sender` are already in scope here.)
    //
    // Start portfolio snapshot task for periodic state persistence.
    // Capture uses the journal batcher flush barrier and the portfolio projection
    // barrier so snapshot state and replay boundary are aligned.
    let portfolio_service_for_snapshot = portfolio_cache.get_service();
    let snapshot_capture_portfolio_cache = portfolio_cache.clone();
    let snapshot_capture_diesel = shared_db.clone();
    let snapshot_capture_sender = journal_batch_sender.clone();
    let snapshot_task = Arc::new(
        PortfolioSnapshotTask::new(
            shared_db.clone(),
            shared_db.clone(),
            portfolio_service_for_snapshot,
            SnapshotTaskConfig::default(),
        )
        .with_capture_snapshot(Arc::new(move || {
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(async {
                    let _guard = snapshot_capture_portfolio_cache
                        .lock_projection_barrier()
                        .await;
                    if let Some(sender) = &snapshot_capture_sender {
                        let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
                        sender
                            .send(crate::journal::JournalMessage::Flush(ack_tx))
                            .await
                            .map_err(|e| {
                                anyhow::anyhow!("Failed to flush journal batcher: {}", e)
                            })?;
                        ack_rx.await.map_err(|_| {
                            anyhow::anyhow!("Journal batcher dropped snapshot flush ack")
                        })?;
                    }

                    let replay: &dyn hypercall_db::JournalReplayReader =
                        snapshot_capture_diesel.as_ref();
                    let next_command_id = replay.get_next_engine_command_id_sync()?;
                    let states = snapshot_capture_portfolio_cache
                        .capture_portfolios_under_barrier(&_guard)
                        .await;

                    let mut offsets = HashMap::new();
                    offsets.insert(
                        crate::read_cache::portfolio::ENGINE_COMMAND_SNAPSHOT_STREAM.to_string(),
                        HashMap::from([(0, next_command_id)]),
                    );
                    Ok((states, offsets))
                })
            })
        })),
    );
    if !standby_mode_active {
        service_registry.register(snapshot_task);
        info!("Portfolio snapshot task started (60s interval)");
    } else {
        info!("Portfolio snapshot task deferred (standby mode)");
    }

    // Save a clone of oracles for Hydromancer startup recovery (before builder consumes them)
    let mark_price_oracles_for_recovery = mark_price_oracles.clone();

    let engine_wal_path = hypercall_journal::checkpoint::wal_path_from_config(
        runtime.config.engine.persistence.journal.wal_path.as_ref(),
    );
    let engine_drain_marker_path = handlers::health::drain_marker_path(&engine_wal_path);
    let engine_wal_path_configured =
        hypercall_journal::checkpoint::wal_path_is_explicitly_configured(
            runtime.config.engine.persistence.journal.wal_path.as_ref(),
        );
    let engine_start_quiesced =
        engine_wal_path_configured && tokio::fs::metadata(&engine_drain_marker_path).await.is_ok();
    if engine_start_quiesced {
        info!(
            path = %engine_drain_marker_path.display(),
            "Persistent drain marker found; engine will start quiesced"
        );
    }
    let recovery_safety_report = hypercall_api::recovery_safety::shared_report();
    let portfolio_margin_mode_allowlist = parse_configured_wallet_set(
        "PORTFOLIO_MARGIN_MODE_ALLOWLIST",
        &runtime.config.engine.portfolio_margin_mode_allowlist,
    )?;
    let portfolio_margin_settlement_allowlist = parse_configured_wallet_set(
        "PORTFOLIO_MARGIN_SETTLEMENT_ALLOWLIST",
        &runtime.config.engine.portfolio_margin_settlement_allowlist,
    )?;

    let mut builder = UnifiedEngineBuilder::new(config)
        .with_order_buffer_size(1000)
        .with_database(&engine_database_url)
        .with_db_auth(engine_db_auth)
        .with_runtime_settings(crate::rsm::unified_engine::EngineRuntimeSettings {
            snapshot_interval: Duration::from_secs(runtime.config.engine.snapshot_interval_secs),
            read_snapshot_interval: Duration::from_millis(
                runtime.config.engine.read_snapshot_interval_ms,
            ),
            post_startup_reconcile_delay: Duration::from_secs(
                runtime.config.engine.post_startup_reconcile_delay_secs,
            ),
            wal_path: runtime.config.engine.persistence.journal.wal_path.clone(),
            start_quiesced: engine_start_quiesced,
            recovery_safety_report: Some(recovery_safety_report.clone()),
            portfolio_margin_pool_enabled: runtime.config.engine.portfolio_margin_pool_enabled,
            portfolio_margin_settlement_allowlist: portfolio_margin_settlement_allowlist.clone(),
        })
        .with_mmp_cache(mmp_cache.clone())
        .with_tier_cache(tier_cache.clone())
        .with_portfolio_service(portfolio_service)
        .with_portfolio_cache(portfolio_cache.clone())
        .with_greeks_cache(greeks_cache.clone())
        .with_mark_price_oracles(mark_price_oracles)
        .with_risk_account_builder(risk_account_builder.clone())
        .with_vol_oracle(risk_vol_oracle.clone())
        .with_standard_account_builder(standard_account_builder.clone())
        .with_liquidation_cache(liquidation_cache.clone());

    // In standby mode, skip DDL in the engine's internal DatabaseHandler.
    // The readonly replica can serve reads (max order_id, trade_id, l2_seq)
    // but cannot run migrations.
    if standby_mode_active || skip_db_migrations {
        builder = builder.with_skip_db_migrations();
    }

    // Add journal writer to builder
    if let Some(ref writer) = engine_journal_writer {
        builder = builder.with_journal_writer(writer.clone());
    }

    // Add journal batch sender to builder
    if let Some(ref sender) = journal_batch_sender {
        builder = builder.with_journal_batch_sender(sender.clone());
    }

    // If journal is disabled, use mock journal (engine requires one)
    if engine_journal_writer.is_none() && journal_batch_sender.is_none() {
        builder = builder.with_mock_journal();
    }

    let default_standby_replay_start_seq = if standby_mode_active {
        let nats_config =
            crate::nats::NatsConfig::from_env().expect("NATS_URL required for standby mode");
        let latest_seq = nats_config
            .latest_stream_sequence()
            .await
            .with_context(|| "Failed to read NATS stream tail for standby replay start sequence")?;
        info!(
            latest_seq,
            "Captured NATS stream tail before standby startup recovery"
        );
        Some(latest_seq as i64)
    } else {
        None
    };

    let mut balance_update_publisher_for_api = None;
    let mut balance_update_stream_required = false;

    // Connect NATS publisher for real-time command replication (optional)
    if let Some(nats_config) = crate::nats::NatsConfig::from_env() {
        balance_update_stream_required = true;
        match crate::nats::NatsPublisher::connect(&nats_config).await {
            Ok(publisher) => {
                builder = builder.with_nats_publisher(publisher);
                info!("NATS publisher connected for real-time command replication");
            }
            Err(e) => {
                warn!(
                    "Failed to connect NATS publisher (replication disabled): {}",
                    e
                );
            }
        }
        match crate::nats::NatsBalanceUpdatePublisher::connect(&nats_config).await {
            Ok(publisher) => {
                balance_update_publisher_for_api = Some(publisher.clone());
                builder = builder.with_nats_balance_update_publisher(publisher);
                info!("NATS balance update publisher connected");
            }
            Err(e) => {
                warn!("Failed to connect NATS balance update publisher: {}", e);
            }
        }
    }

    #[cfg(feature = "testnet")]
    let testnet_bootstrap_accounts = {
        tracing::warn!("testnet feature enabled, funding hardhat test accounts");
        let test_accounts = vec![
            (
                WalletAddress::from_str("0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266").unwrap(),
                100_000.0,
            ), // Hardhat account 0 - wallet1 in tests
            (
                WalletAddress::from_str("0x70997970C51812dc3A010C7d01b50e0d17dc79C8").unwrap(),
                100_000.0,
            ), // Hardhat account 1 - wallet2 in tests
            // Note: Hardhat account 2 (0x3C44...) is intentionally NOT funded - used as wallet3 (observer-only) in tests
            // Hardhat account 3 - used for margin rejection tests with limited $10k funding
            (
                WalletAddress::from_str("0x90F79bf6EB2c4f870365E785982E1f101E93b906").unwrap(),
                10_000.0,
            ), // Hardhat account 3 - $10k for margin rejection tests
        ];

        test_accounts
    };

    builder = builder.with_read_snapshot(read_snapshot.clone());

    builder = builder.with_ws_event_sender(ws_direct_tx.clone());

    let engine_result = builder.build_with_info(engine_event_tx.clone(), shutdown.tx());
    let mut unified_engine = engine_result.engine;
    let engine_order_tx = engine_result.order_tx;
    let engine_market_tx = engine_result.market_tx;
    let engine_margin_mode_tx = engine_result.margin_mode_tx;
    let engine_agent_auth_tx = engine_result.agent_auth_tx;
    let engine_nonce_check_tx = engine_result.nonce_check_tx;
    let engine_pm_settlement_admin_tx = engine_result.pm_settlement_admin_tx;
    let engine_quiesce_tx = engine_result.quiesce_tx;

    // Add engine sync_status to readiness gate so the API rejects orders
    // until post-startup ghost-order reconciliation completes (~5s after start).
    readiness_checks.push(Arc::new(SyncStatusReadiness::new(
        "engine",
        engine_result.sync_status,
    )));
    let readiness = Arc::new(ReadinessRegistry::new(readiness_checks));
    info!(
        "Readiness registry initialized with {} checks (including engine)",
        readiness.reports().len()
    );

    // Startup recovery: attempt Hydromancer fallback for EXPIRED_PENDING_PRICE instruments
    if hydromancer_client.is_some() {
        let recovery_pool = db_pool.clone();
        let recovery_oracles = mark_price_oracles_for_recovery.clone();
        task_group.spawn("HydromancerStartupRecovery", async move {
            match crate::startup::hydromancer::recover_stuck_settlements(
                &recovery_pool,
                &recovery_oracles,
            )
            .await
            {
                Ok(count) => {
                    if count > 0 {
                        info!(
                            "Hydromancer startup recovery: settled {} stuck instruments",
                            count
                        );
                    }
                }
                Err(e) => {
                    error!("Hydromancer startup recovery failed: {}", e);
                }
            }
            Ok(())
        });
    }

    // Wire RFQ execute channel to engine before starting it
    let (rfq_execute_tx, rfq_execute_rx) =
        tokio::sync::mpsc::channel::<hypercall_runtime_api::RfqExecuteRequest>(128);
    unified_engine.set_rfq_receiver(rfq_execute_rx);

    // Wire deposit channel for faucet/admin deposits -> balance_ledger.
    // Failed withdrawals require manual reconciliation and are not auto-refunded.
    unified_engine.set_deposit_receiver(deposit_rx);

    #[cfg(feature = "testnet")]
    let testnet_bootstrap_deposits = {
        let mut deposits = Vec::new();
        let now_ms = crate::shared::order_types::get_timestamp_millis();
        for (idx, (wallet, cash)) in testnet_bootstrap_accounts.iter().enumerate() {
            let sequence = testnet_bootstrap_deposit_sequence(idx);
            let source_event_hash = testnet_bootstrap_source_event_hash(sequence);
            if unified_engine
                .ctx
                .applied_deposit_source_event_hashes
                .contains(&source_event_hash)
            {
                continue;
            }

            let current_cash = unified_engine.ctx.balance_ledger.balance(wallet);
            if current_cash != rust_decimal::Decimal::ZERO {
                warn!(
                    wallet = %wallet,
                    current_cash = %current_cash,
                    "Skipping testnet bootstrap funding for wallet with restored engine cash but no bootstrap source hash"
                );
                continue;
            }

            let amount = rust_decimal::Decimal::from_f64_retain(*cash)
                .expect("testnet bootstrap cash f64 -> Decimal conversion");
            let (applied_tx, applied_rx) = tokio::sync::oneshot::channel();
            deposits.push((
                crate::rsm::unified_engine::DepositRequest {
                    wallet: *wallet,
                    amount,
                    timestamp_ms: now_ms,
                    sequence: Some(sequence),
                    source_event_hash,
                    journal_request_id: testnet_bootstrap_request_id(idx),
                    outbox_appends: Vec::new(),
                    applied_tx: Some(applied_tx),
                },
                applied_rx,
            ));
        }
        deposits
    };

    let (option_deposit_tx, option_deposit_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::OptionDepositRequest>(128);
    unified_engine.set_option_deposit_receiver(option_deposit_rx);

    let (option_withdrawal_tx, option_withdrawal_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::OptionWithdrawalRequest>(128);
    unified_engine.set_option_withdrawal_receiver(option_withdrawal_rx);

    let (cash_withdrawal_tx, cash_withdrawal_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::CashWithdrawalRequest>(128);
    unified_engine.set_cash_withdrawal_receiver(cash_withdrawal_rx);

    // Failed withdrawals, cash or option, are not auto-refunded in this phase.
    // Operators reconcile them manually until user-driven withdrawal proofs exist.

    if !standby_mode {
        if let Some(rsm_signer) = rsm_signer_service.as_ref() {
            let signer_address = rsm_signer.signer_address();
            let durable_next_nonce = shared_db
                .get_rsm_signer_nonce(&signer_address)
                .await
                .with_context(|| {
                    format!("failed to load durable RSM signer nonce for {signer_address}")
                })?
                .map(|record| {
                    u64::try_from(record.next_nonce).with_context(|| {
                        format!(
                            "persisted next_nonce {} for signer {} is negative",
                            record.next_nonce, signer_address
                        )
                    })
                })
                .transpose()?
                .unwrap_or(0);
            unified_engine.seed_rsm_signer_nonce_floor(signer_address, durable_next_nonce);
            info!(
                signer = %signer_address,
                durable_next_nonce,
                "Seeded engine RSM signer nonce floor from durable signer tracker"
            );
        }

        if let Some(rsm_signer) = rsm_signer_service.clone() {
            let sequencer = Arc::new(crate::directive_outbox::DirectiveDeliverySequencer::new(
                shared_db.clone(),
                rsm_signer,
                engine_event_tx.clone(),
                runtime.config.engine.persistence.outbox.poll_ms,
            ));
            let handle = sequencer.start_with_shutdown(shutdown.subscribe());
            task_group.spawn("DirectiveDeliverySequencer", async move {
                if let Err(error) = handle.await {
                    error!("DirectiveDeliverySequencer task panicked: {:?}", error);
                }
                Ok(())
            });
            tracing::info!("✓ Directive delivery sequencer started");
        }
    }

    let (tier_update_tx, tier_update_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::TierUpdateRequest>(128);
    unified_engine.set_tier_update_receiver(tier_update_rx);

    // Wire HyperCore equity channel for PM margin (Hydromancer feed -> engine)
    let (hypercore_equity_tx, hypercore_equity_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::HypercoreEquityRequest>(128);
    unified_engine.set_hypercore_equity_receiver(hypercore_equity_rx);

    // Wire liquidation bonus channel separately from deposits for semantic clarity.
    let (liquidation_bonus_tx, liquidation_bonus_rx) =
        tokio::sync::mpsc::channel::<crate::rsm::unified_engine::LiquidationBonusRequest>(128);
    unified_engine.set_liquidation_bonus_receiver(liquidation_bonus_rx);

    // Wire trading_mode notify channel. Initial value is an empty
    // map — the catalog manager publishes the real map on its first
    // rewrite tick. The engine main loop `changed()`s on the receiver
    // and applies updates via `apply_underlying_trading_mode_update`.
    let (trading_mode_notify_tx, trading_mode_notify_rx) = tokio::sync::watch::channel::<
        std::collections::HashMap<String, hypercall_types::TradingModes>,
    >(std::collections::HashMap::new());
    unified_engine.set_trading_mode_receiver(trading_mode_notify_rx);

    // --- NATS standby replication ---
    let standby_startup_heartbeat = if standby_mode {
        startup_progress.as_ref().map(|progress| {
            let progress = progress.clone();
            tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_secs(15));
                loop {
                    interval.tick().await;
                    progress.heartbeat();
                }
            })
        })
    } else {
        None
    };

    if standby_mode {
        let nats_config =
            crate::nats::NatsConfig::from_env().expect("NATS_URL required for standby mode");

        if let Some(progress) = startup_progress.as_ref() {
            progress.mark("standby_hydrate_base_state");
        }
        unified_engine.hydrate_standby_base_state().await;
        if let Some(progress) = startup_progress.as_ref() {
            progress.mark("standby_hydrate_base_state_complete");
        }

        let progress = standby_progress
            .clone()
            .context("standby replay progress missing while STANDBY_MODE is active")?;
        let promote_rx = standby_replay_promote_rx
            .take()
            .context("standby promote receiver missing while STANDBY_MODE is active")?;
        let (engine_tx, engine_rx) =
            tokio::sync::oneshot::channel::<crate::nats::standby_handler::StandbyEngineHandler>();

        let handler = crate::nats::standby_handler::StandbyEngineHandler::new(unified_engine);
        let default_replay_start_seq = default_standby_replay_start_seq.ok_or_else(|| {
            anyhow::anyhow!("NATS stream tail was not captured for standby replay")
        })?;
        let replay_start_seq = match std::env::var("NATS_REPLAY_START_SEQ") {
            Ok(value) => value
                .parse::<i64>()
                .with_context(|| format!("NATS_REPLAY_START_SEQ must be an i64, got {value:?}"))?,
            Err(std::env::VarError::NotPresent) => default_replay_start_seq,
            Err(error) => {
                return Err(anyhow::anyhow!(
                    "Failed to read NATS_REPLAY_START_SEQ: {error}"
                ));
            }
        };
        if replay_start_seq < 0 {
            return Err(anyhow::anyhow!(
                "NATS_REPLAY_START_SEQ must be >= 0, got {replay_start_seq}"
            ));
        }
        info!(
            replay_start_seq,
            default_replay_start_seq, "Starting standby NATS replay"
        );
        info!("Starting in STANDBY mode — replaying from NATS, engine deferred");
        if let Some(progress) = startup_progress.as_ref() {
            progress.mark("standby_nats_replay_starting");
        }

        task_group.spawn("StandbyReplayLoop", async move {
            let (bcast_tx, bcast_rx) = tokio::sync::broadcast::channel::<()>(1);
            tokio::spawn(async move {
                promote_rx.await.ok();
                bcast_tx.send(()).ok();
            });
            match crate::nats::replay_loop::run_replay_loop(
                &nats_config,
                replay_start_seq,
                handler,
                progress,
                bcast_rx,
            )
            .await
            {
                Ok(h) => {
                    engine_tx.send(h).ok();
                }
                Err(e) => {
                    error!("Standby replay loop error: {}", e);
                }
            }
            Ok(())
        });

        let db_auth_for_promote = db_auth.clone();
        let post_promote_services_for_promote = post_promote_services.clone();
        let shutdown_for_promote = shutdown.clone();
        let shared_db_for_promote = shared_db.clone();
        let rsm_signer_for_promote = rsm_signer_service.clone();
        let engine_event_tx_for_promote = engine_event_tx.clone();
        let directive_outbox_poll_ms = runtime.config.engine.persistence.outbox.poll_ms;
        task_group.spawn("StandbyPromote", async move {
            match engine_rx.await {
                Ok(handler) => {
                    info!("Standby promoted — creating read-write database pools");
                    let mut engine = handler.into_engine();

                    // Create a read-write DieselEventHandler from DATABASE_URL and
                    // swap it into the engine, replacing the readonly handler used
                    // during NATS replay.  This runs migrations on the primary.
                    // Failure here is fatal: starting the engine without write-capable
                    // persistence would silently drop fills/settlements.
                    match DatabaseHandler::new_with_auth(db_auth_for_promote.clone(), 3) {
                        Ok(rw_handler) => {
                            engine.set_db(rw_handler);
                            info!("Pool swap complete — engine now has read-write DB access");
                        }
                        Err(e) => {
                            panic!(
                                "CRITICAL_FAILURE: Failed to create read-write DatabaseHandler on promote: {}. \
                                 Cannot start engine without write-capable persistence.",
                                e
                            );
                        }
                    }

                    if let Some(rsm_signer) = rsm_signer_for_promote.clone() {
                        if let Err(error) = rsm_signer.status().await {
                            panic!(
                                "CRITICAL_FAILURE: Failed to initialize RSM signer after standby promotion: {}",
                                error
                            );
                        }

                        let signer_address = rsm_signer.signer_address();
                        let durable_next_nonce = shared_db_for_promote
                            .get_rsm_signer_nonce(&signer_address)
                            .await
                            .unwrap_or_else(|error| {
                                panic!(
                                    "CRITICAL_FAILURE: failed to load durable RSM signer nonce for {} after standby promotion: {}",
                                    signer_address, error
                                )
                            })
                            .map(|record| {
                                u64::try_from(record.next_nonce).unwrap_or_else(|_| {
                                    panic!(
                                        "CRITICAL_FAILURE: persisted next_nonce {} for signer {} is negative after standby promotion",
                                        record.next_nonce, signer_address
                                    )
                                })
                            })
                            .unwrap_or(0);
                        engine.seed_rsm_signer_nonce_floor(signer_address, durable_next_nonce);
                        info!(
                            signer = %signer_address,
                            durable_next_nonce,
                            "Seeded engine RSM signer nonce floor after standby promotion"
                        );

                        let sequencer =
                            Arc::new(crate::directive_outbox::DirectiveDeliverySequencer::new(
                                shared_db_for_promote.clone(),
                                rsm_signer,
                                engine_event_tx_for_promote.clone(),
                                directive_outbox_poll_ms,
                            ));
                        let handle =
                            sequencer.start_with_shutdown(shutdown_for_promote.subscribe());
                        tokio::spawn(async move {
                            if let Err(error) = handle.await {
                                error!("DirectiveDeliverySequencer task panicked: {:?}", error);
                            }
                        });
                        info!("Directive delivery sequencer started after standby promotion");
                    }


                    let services = {
                        let mut services = post_promote_services_for_promote.lock().await;
                        std::mem::take(&mut *services)
                    };
                    for service in services {
                        let service_name = service.name();
                        let owner = service.owner();
                        let shutdown_rx = shutdown_for_promote.subscribe();
                        info!(
                            service = service_name,
                            owner = %owner,
                            "Starting deferred service after standby promotion"
                        );
                        tokio::spawn(async move {
                            if let Err(error) = service.run(shutdown_rx).await {
                                error!(
                                    service = service_name,
                                    owner = %owner,
                                    error = %error,
                                    "Deferred service exited with error"
                                );
                            }
                        });
                    }

                    info!("Starting engine event loop after promote");
                    engine.start().await;
                }
                Err(_) => {
                    error!("Replay loop ended without returning engine");
                }
            }
            Ok(())
        });

        if let Some(progress) = startup_progress.as_ref() {
            progress.mark("standby_replay_task_spawned");
        }
    } else {
        task_group.spawn("UnifiedEngine", async move {
            unified_engine.start().await;
            Ok(())
        });

        #[cfg(feature = "testnet")]
        for (deposit, applied_rx) in testnet_bootstrap_deposits {
            let wallet = deposit.wallet;
            let amount = deposit.amount;
            deposit_tx.send(deposit).await.map_err(|error| {
                anyhow::anyhow!("failed to queue testnet bootstrap deposit: {error}")
            })?;
            match applied_rx.await {
                Ok(Ok(())) => {
                    info!(
                        wallet = %wallet,
                        amount = %amount,
                        "Applied journaled testnet bootstrap deposit"
                    );
                }
                Ok(Err(error)) => {
                    return Err(anyhow::anyhow!(
                        "testnet bootstrap deposit for {wallet} failed: {error}"
                    ));
                }
                Err(error) => {
                    return Err(anyhow::anyhow!(
                        "testnet bootstrap deposit ack for {wallet} dropped: {error}"
                    ));
                }
            }
        }
    }

    // Create drain signal + flag for blue-green switchover
    let drain_signal = std::sync::Arc::new(tokio::sync::Notify::new());
    let is_draining =
        std::sync::Arc::new(std::sync::atomic::AtomicBool::new(engine_start_quiesced));

    // Forward orders to unified engine (shutdown-aware, standby-aware, drain-aware)
    let mut order_rx = order_rx;
    let engine_order_tx_clone = engine_order_tx.clone();
    let mut order_forwarder_shutdown_rx = shutdown.subscribe();
    let standby_for_forwarder = standby_controller.clone();
    let is_draining_for_forwarder = is_draining.clone();
    task_group.spawn("OrderForwarder", async move {
        loop {
            tokio::select! {
                _ = order_forwarder_shutdown_rx.recv() => {
                    info!("OrderForwarder received shutdown signal");
                    break;
                }
                maybe_req = order_rx.recv() => {
                    match maybe_req {
                        Some(request) => {
                            // In standby/draining mode, reject orders immediately.
                            // The blue/green deploy script promotes before switching
                            // the Service, so write traffic shouldn't arrive in standby.
                            // Rejecting (instead of queuing) avoids hanging HTTP handlers.
                            if let Some(ref controller) = standby_for_forwarder {
                                if controller.is_standby().await {
                                    tracing::debug!("Rejecting order — server in standby mode");
                                    let _ = request.response_tx.send(hypercall_types::OrderUpdateMessage {
                                        order_id: None,
                                        info: request.message.info,
                                        status: hypercall_types::OrderUpdateStatus::Rejected,
                                        timestamp: std::time::SystemTime::now()
                                            .duration_since(std::time::UNIX_EPOCH)
                                            .unwrap_or_default()
                                            .as_millis() as u64,
                                        reason: Some("Server in standby mode, not accepting orders".to_string()),
                                        filled_size: rust_decimal::Decimal::ZERO,
                                        wallet_address: request.message.wallet,
                                        mmp_triggered: false,
                                        request_id: request.message.request_id,
                                    }).await;
                                    continue;
                                }
                            }
                            // Check drain flag before forwarding
                            if is_draining_for_forwarder.load(std::sync::atomic::Ordering::Relaxed) {
                                tracing::debug!("Rejecting order — server is draining");
                                let _ = request.response_tx.send(hypercall_types::OrderUpdateMessage {
                                    order_id: None,
                                    info: request.message.info,
                                    status: hypercall_types::OrderUpdateStatus::Rejected,
                                    timestamp: std::time::SystemTime::now()
                                        .duration_since(std::time::UNIX_EPOCH)
                                        .unwrap_or_default()
                                        .as_millis() as u64,
                                    reason: Some("Server is draining for blue/green deploy".to_string()),
                                    filled_size: rust_decimal::Decimal::ZERO,
                                    wallet_address: request.message.wallet,
                                    mmp_triggered: false,
                                    request_id: request.message.request_id,
                                }).await;
                                continue;
                            }
                            // Active mode (or no standby controller) — forward to engine
                            if engine_order_tx_clone.send(request).await.is_err() {
                                tracing::error!("Failed to send order to unified engine");
                                break;
                            }
                        }
                        None => break,
                    }
                }
            }
        }
        Ok(())
    });

    // Transition standby controller to active after promote
    if let Some(drain_controller) = standby_controller.clone() {
        task_group.spawn("StandbyFinalize", async move {
            loop {
                if drain_controller.is_promoted().await {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            }
            // No orders to drain (OrderForwarder rejects in standby instead of queuing).
            // Just transition to active so is_standby() returns false.
            let _ = drain_controller.take_drain_batch().await;
            info!("Standby controller transitioned to active");
            Ok(())
        });
    }

    // Start CatalogManager if enabled
    // CatalogManager continuously reconciles markets/instruments from policy file
    // Note: catalog_config was already loaded earlier for oracle setup
    if runtime.config.catalog_manager.enabled {
        let catalog_manager = crate::catalog_manager::CatalogManager::new(
            diesel_db.clone(),
            catalog_config,
            engine_market_tx.clone(),
            mark_price_oracles_for_catalog,
            runtime.config.catalog_manager.interval_secs,
        )
        .with_trading_mode_notify(trading_mode_notify_tx);

        let catalog_manager = Arc::new(catalog_manager);
        if standby_mode {
            let deferred_service: Arc<dyn Service> = catalog_manager;
            post_promote_services.lock().await.push(deferred_service);
            info!("CatalogManager deferred until standby promotion");
        } else {
            service_registry.register(catalog_manager);
            info!(
                "✓ CatalogManager started with interval={}s (trading_mode notify channel wired)",
                runtime.config.catalog_manager.interval_secs
            );
        }
    } else {
        debug!("CatalogManager disabled");
    }

    let trading_halt = Arc::new(tokio::sync::RwLock::new(TradingHaltState::from_config(
        runtime.config.api.global_trading_halted,
        &runtime.config.api.halted_markets,
    )));
    {
        let halt_state = trading_halt.read().await;
        if halt_state.global_halt.is_some() || !halt_state.halted_markets.is_empty() {
            tracing::warn!(
                global_halted = halt_state.global_halt.is_some(),
                halted_markets = halt_state.halted_markets.len(),
                "Trading halt loaded from startup configuration"
            );
        }
    }

    let chain_auth: Arc<dyn directives::onchain::ChainAuthReader> = Arc::new(
        directives::onchain::AlloyChainAuthReader::new(
            &runtime.config.transaction_submitter.rpc_url,
            &runtime.config.contracts.exchange_contract_address,
        )
        .map_err(|e| anyhow::anyhow!("Failed to initialize chain auth reader: {}", e))?,
    );
    let gas_provider_service = Arc::new(if cfg!(feature = "test-endpoints") {
        hypercall_api::gas_provider::GasProviderService::new_mock()
    } else {
        match hypercall_api::gas_provider::GasProviderService::new(
            &runtime.config.transaction_submitter.rpc_url,
        )
        .await
        {
            Ok(service) => service,
            Err(error)
                if runtime.config.transaction_submitter.mode
                    == hypercall_transaction_submitter::TransactionSubmitterMode::Mock =>
            {
                let fallback_chain_id = if runtime.config.modes.testnet_mode {
                    hypercall_types::directives::HYPERCALL_TESTNET_CHAIN_ID
                } else {
                    hypercall_types::directives::HYPERCALL_MAINNET_CHAIN_ID
                };
                tracing::warn!(
                    chain_id = fallback_chain_id,
                    rpc_url = %runtime.config.transaction_submitter.rpc_url,
                    "Gas provider chain ID lookup failed in mock submitter mode, falling back to configured chain id: {}",
                    error
                );
                hypercall_api::gas_provider::GasProviderService::new_with_chain_id(
                    &runtime.config.transaction_submitter.rpc_url,
                    fallback_chain_id,
                )
                .map_err(|fallback_error| {
                    anyhow::anyhow!(
                        "Failed to initialize gas provider service fallback: {}",
                        fallback_error
                    )
                })?
            }
            Err(error) => {
                return Err(anyhow::anyhow!(
                    "Failed to initialize gas provider service: {}",
                    error
                ));
            }
        }
    });

    let signing_chain_id = if runtime.config.modes.testnet_mode {
        hypercall_types::directives::HYPERCALL_TESTNET_CHAIN_ID
    } else {
        hypercall_types::directives::HYPERCALL_MAINNET_CHAIN_ID
    };
    let api_portfolio_margin_mode_allowlist = portfolio_margin_mode_allowlist
        .iter()
        .copied()
        .collect::<Vec<_>>();
    let api_portfolio_margin_settlement_allowlist = portfolio_margin_settlement_allowlist
        .iter()
        .copied()
        .collect::<Vec<_>>();
    let app_runtime_config = Arc::new(handlers::AppRuntimeConfig {
        testnet_mode: runtime.config.modes.testnet_mode,
        trade_explorer_url_template: runtime.config.api.trade_explorer_url_template.clone(),
        wal_path: hypercall_journal::checkpoint::wal_path_from_config(
            runtime.config.engine.persistence.journal.wal_path.as_ref(),
        ),
        db_host: crate::startup::database::extract_db_host(&runtime.secrets.database_url),
        db_name: crate::startup::database::extract_db_name(&runtime.secrets.database_url),
        directive_chain_id: signing_chain_id,
        signing_chain_id,
        exchange_contract_address: runtime.config.contracts.exchange_contract_address.clone(),
        portfolio_margin_pool_enabled: runtime.config.engine.portfolio_margin_pool_enabled,
        portfolio_margin_mode_allowlist: api_portfolio_margin_mode_allowlist,
        portfolio_margin_settlement_allowlist: api_portfolio_margin_settlement_allowlist,
        #[cfg(feature = "rsm-state")]
        rsm_environment: runtime
            .config
            .engine
            .persistence
            .journal
            .rsm_environment
            .into(),
    });

    // Initialize username service (Postgres + optional Redis cache)
    let username_service = Arc::new(
        hypercall_api::username_service::UsernameService::new(
            diesel_db.clone(),
            username_redis_client,
        )
        .await,
    );
    tracing::info!("✓ UsernameService initialized");

    let rfq_websocket_publisher: Arc<dyn hypercall_api::rfq::qp_ws_state::RfqWebsocketPublisher> =
        pubsub.clone();
    let rfq_resources = crate::startup::rfq::build_rfq_resources(
        shared_db.clone(),
        rfq_execute_tx,
        &shutdown,
        indicative_cache.clone(),
        agent_auth.clone(),
        signing_chain_id,
        instruments_cache.clone(),
        engine_nonce_check_tx.clone(),
        Some(rfq_websocket_publisher),
    )
    .await?;
    let rfq_manager = rfq_resources.rfq_manager.clone();
    let rfq_handler_state = rfq_resources.rfq_handler_state.clone();
    let qp_ws_state = rfq_resources.qp_ws_state;

    // Create app state
    let app_state = handlers::AppState {
        db: diesel_db.clone(),
        order_sender: order_tx.clone(),
        market_sender: engine_market_tx.clone(),
        engine_quiesce_sender: engine_quiesce_tx.clone(),
        margin_mode_sender: engine_margin_mode_tx.clone(),
        agent_auth_sender: engine_agent_auth_tx.clone(),
        agent_auth: agent_auth.clone(),
        auth_failure_recorder: Arc::new(
            crate::observability::api_boundary::MetricsAuthFailureRecorder,
        ),
        metrics_renderer: Arc::new(crate::observability::api_boundary::PrometheusMetricsRenderer),
        greeks_cache: greeks_cache.clone(),
        portfolio_cache: portfolio_cache.clone(),
        instruments_cache: instruments_cache.clone(),
        market_stats_cache: market_stats_cache.clone(),
        markets_snapshot_cache: markets_snapshot_cache.clone(),
        instruments_snapshot_cache: instruments_snapshot_cache.clone(),
        options_summary_snapshot_cache: options_summary_snapshot_cache.clone(),
        event_bus_sender: engine_event_tx.clone(),
        chain_auth: chain_auth.clone(),
        exchange_address: alloy::primitives::Address::from_str(
            &runtime.config.contracts.exchange_contract_address,
        )
        .ok(),
        gas_provider_service: gas_provider_service.clone(),
        transaction_request_journal: Some(Arc::new(
            crate::api_boundary_impls::RuntimeTransactionRequestJournal::new(
                journal_batch_sender.clone(),
                engine_journal_writer.clone(),
            ),
        )),
        mmp_cache: mmp_cache.clone(),
        tier_cache: tier_cache.clone(),
        risk_vol_oracle: risk_vol_oracle.clone(),
        readiness: readiness.clone() as Arc<dyn hypercall_api::runtime_status::ReadinessGate>,
        balance_provider: Arc::new(crate::api_boundary_impls::RuntimeBalanceProvider::new(
            balance_provider.clone(),
        )),
        balance_snapshot_provider: Arc::new(
            crate::api_boundary_impls::RuntimeBalanceSnapshotProvider::new(
                snapshot_provider.clone(),
                balance_update_publisher_for_api.clone(),
                balance_update_stream_required,
            ),
        ),
        deposit_sender: Some(deposit_tx.clone()),
        option_deposit_sender: Some(option_deposit_tx.clone()),
        option_withdrawal_sender: Some(option_withdrawal_tx.clone()),
        cash_withdrawal_sender: Some(cash_withdrawal_tx.clone()),
        tier_update_sender: Some(tier_update_tx.clone()),
        pm_settlement_admin_sender: Some(engine_pm_settlement_admin_tx.clone()),
        sync_db: Some(shared_db.clone()),
        rate_limit_cache: rate_limit_cache.clone(),
        engine_journal_reader: engine_journal_writer.clone().map(|writer| {
            Arc::new(crate::api_boundary_impls::RuntimeEngineJournalReader::new(
                writer,
            )) as Arc<dyn hypercall_api::boundary::engine::EngineJournalReader>
        }),
        runtime_config: app_runtime_config,
        build_info: hypercall_api::observability_boundary::BuildInfo {
            version: crate::observability::CARGO_VERSION.to_string(),
            commit: crate::observability::GIT_COMMIT.to_string(),
            git_ref: crate::observability::GIT_REF.to_string(),
            build_time: crate::observability::BUILD_TIME.to_string(),
        },
        collateral_registry: collateral_registry.clone(),
        admin_api_key: runtime.secrets.admin_api_key.clone(),
        allow_unauthenticated_monitoring: runtime.config.environment.name == "development",
        boot_id: uuid::Uuid::now_v7().to_string(),
        server_started_at: chrono::Utc::now(),
        trading_halt: trading_halt.clone(),
        rsm_signer: rsm_signer_service.clone(),
        rsm_signer_address: if standby_mode {
            None
        } else {
            rsm_signer_service
                .as_ref()
                .map(|signer| signer.signer_address())
        },
        // Fill checks go to HyperCore (testnet), not mainnet pricing
        hl_info_url: Some(runtime.config.pricing.hypercore_info_url.clone()),
        exchange_pool_liquidity_reader: Arc::new(
            hypercall_api::state::HypercoreExchangePoolLiquidityReader::new(
                Some(runtime.config.pricing.hypercore_info_url.clone()),
                sync_exchange_address,
            ),
        ),
        hydromancer_feed: {
            if runtime.config.hydromancer.enabled {
                match runtime
                    .config
                    .hydromancer
                    .ws_api_key
                    .as_deref()
                    .or(runtime.secrets.hydromancer_ws_api_key.as_deref())
                    .or(runtime.secrets.hydromancer_api_key.as_deref())
                {
                    Some(api_key) => {
                        let api_key = api_key.trim();
                        let ws_base = runtime
                            .config
                            .hydromancer
                            .ws_url
                            .as_deref()
                            .unwrap_or("wss://api.hydromancer.xyz/ws");
                        let ws_url = format!("{}?token={}", ws_base, api_key);
                        // Position queries go to HyperCore (testnet), not mainnet pricing
                        let hl_info = runtime.config.pricing.hypercore_info_url.clone();
                        let managed_accounts = crate::startup::hydromancer::load_exchange_accounts(
                            &runtime.config.transaction_submitter.rpc_url,
                            &runtime.config.contracts.exchange_contract_address,
                        )
                        .await;
                        info!(
                            count = managed_accounts.len(),
                            "Hydromancer feed: loaded accounts from Exchange"
                        );

                        let account_addresses: Vec<WalletAddress> =
                            managed_accounts.iter().map(|p| p.account).collect();
                        let account_to_manager: std::collections::HashMap<
                            WalletAddress,
                            WalletAddress,
                        > = managed_accounts
                            .iter()
                            .map(|p| (p.account, p.manager))
                            .collect();

                        let hydromancer_api_url = runtime.config.hydromancer.api_url.clone();
                        let hydromancer_api_key = Some(api_key.to_string());
                        let feed = Arc::new(crate::hypercore::HydromancerFeedService::new(
                            ws_url,
                            hl_info,
                            hydromancer_api_url,
                            hydromancer_api_key,
                            account_addresses,
                        ));
                        let feed_clone = feed.clone();
                        let shutdown_rx = shutdown.subscribe();
                        tokio::spawn(async move {
                            feed_clone.start(shutdown_rx).await;
                        });

                        // Bridge: perp positions from Hydromancer feed → PortfolioService for PM.
                        // Only uses PositionSnapshot (full reconciled state from REST on each
                        // connect/reconnect). Fills are handled by the feed's internal position
                        // tracking; the next snapshot will reflect them.
                        {
                            let mut position_rx = feed.subscribe();
                            let pc = portfolio_cache.clone();
                            let feed_ref = feed.clone();
                            let pubsub_for_bridge = pubsub.clone();
                            let initial_accounts: Vec<WalletAddress> =
                                managed_accounts.iter().map(|p| p.account).collect();
                            let a2m =
                                std::sync::Arc::new(tokio::sync::RwLock::new(account_to_manager));
                            let chain_auth_bridge: Arc<dyn directives::onchain::ChainAuthReader> =
                                chain_auth.clone();
                            let equity_tx = hypercore_equity_tx.clone();
                            tokio::spawn(async move {
                                info!(
                                    accounts = initial_accounts.len(),
                                    "PM bridge: task spawned, waiting 10s for feed reconciliation"
                                );
                                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                                info!("PM bridge: starting initial sync");
                                for acct in &initial_accounts {
                                    if let Some(state) = feed_ref.get_positions(acct).await {
                                        let manager = {
                                            let map = a2m.read().await;
                                            map.get(acct).copied()
                                        };
                                        if let Some(manager) = manager {
                                            for pos in &state.positions {
                                                let Some(entry_price) = pos.entry_price else {
                                                    warn!(
                                                        account = %acct,
                                                        coin = %pos.coin,
                                                        "PM bridge: skipping Hydromancer position with missing entry price"
                                                    );
                                                    continue;
                                                };
                                                let update =
                                                    crate::portfolio::HypercorePositionUpdate {
                                                        account: manager.to_string(),
                                                        coin: pos.coin.clone(),
                                                        size: pos.size,
                                                        entry_price,
                                                        unrealized_pnl: pos.unrealized_pnl,
                                                        timestamp: state.last_updated_ms,
                                                        snapshot: true,
                                                    };
                                                pc.handle_hypercore_position_update(update).await;
                                            }
                                            // Send HyperCore equity to engine for PM margin
                                            {
                                                let account_value =
                                                    rust_decimal::Decimal::from_f64_retain(
                                                        state.account_value,
                                                    )
                                                    .unwrap_or_default();
                                                let _ = equity_tx.send(
                                                    crate::rsm::unified_engine::HypercoreEquityRequest {
                                                        wallet: manager,
                                                        account_value,
                                                        timestamp_ms: state.last_updated_ms,
                                                    },
                                                ).await;
                                            }
                                            info!(
                                                account = %acct,
                                                positions = state.positions.len(),
                                                account_value = state.account_value,
                                                "PM bridge: initial sync"
                                            );
                                        }
                                    }
                                }

                                info!("PM bridge: initial sync complete, entering event loop");

                                // Only forward fills to WS that happened after bridge start
                                // to avoid replaying historical fills on reconnect
                                let bridge_start_ms =
                                    crate::shared::order_types::get_timestamp_millis();

                                let mut prev_coins: std::collections::HashMap<
                                    WalletAddress,
                                    std::collections::HashSet<String>,
                                > = std::collections::HashMap::new();

                                loop {
                                    match position_rx.recv().await {
                                        Ok(
                                            crate::hypercore::HydromancerEvent::PositionSnapshot {
                                                account,
                                                state,
                                            },
                                        ) => {
                                            let manager = {
                                                let map = a2m.read().await;
                                                map.get(&account).copied()
                                            };
                                            let manager = match manager {
                                                Some(m) => m,
                                                None => {
                                                    match chain_auth_bridge
                                                        .get_manager(account.inner())
                                                        .await
                                                    {
                                                        Ok(addr)
                                                            if addr
                                                                != alloy::primitives::Address::ZERO =>
                                                        {
                                                            let mgr = WalletAddress::from(addr);
                                                            a2m.write().await.insert(account, mgr);
                                                            mgr
                                                        }
                                                        _ => continue,
                                                    }
                                                }
                                            };

                                            let mut current_coins =
                                                std::collections::HashSet::new();
                                            for pos in &state.positions {
                                                current_coins.insert(pos.coin.clone());
                                                let Some(entry_price) = pos.entry_price else {
                                                    warn!(
                                                        account = %account,
                                                        coin = %pos.coin,
                                                        "PM bridge: skipping Hydromancer position with missing entry price"
                                                    );
                                                    continue;
                                                };
                                                let update =
                                                    crate::portfolio::HypercorePositionUpdate {
                                                        account: manager.to_string(),
                                                        coin: pos.coin.clone(),
                                                        size: pos.size,
                                                        entry_price,
                                                        unrealized_pnl: pos.unrealized_pnl,
                                                        timestamp: state.last_updated_ms,
                                                        snapshot: true,
                                                    };
                                                pc.handle_hypercore_position_update(update).await;
                                            }

                                            // Send HyperCore equity to engine for PM margin
                                            {
                                                let account_value =
                                                    rust_decimal::Decimal::from_f64_retain(
                                                        state.account_value,
                                                    )
                                                    .unwrap_or_default();
                                                let _ = equity_tx.send(
                                                    crate::rsm::unified_engine::HypercoreEquityRequest {
                                                        wallet: manager,
                                                        account_value,
                                                        timestamp_ms: state.last_updated_ms,
                                                    },
                                                ).await;
                                            }

                                            // Clear positions that disappeared since last snapshot
                                            if let Some(old) = prev_coins.get(&account) {
                                                for coin in old.difference(&current_coins) {
                                                    let update =
                                                        crate::portfolio::HypercorePositionUpdate {
                                                            account: manager.to_string(),
                                                            coin: coin.clone(),
                                                            size: 0.0,
                                                            entry_price: 0.0,
                                                            unrealized_pnl: 0.0,
                                                            timestamp: state.last_updated_ms,
                                                            snapshot: true,
                                                        };
                                                    pc.handle_hypercore_position_update(update)
                                                        .await;
                                                }
                                            }
                                            prev_coins.insert(account, current_coins);
                                        }
                                        Ok(crate::hypercore::HydromancerEvent::Fill {
                                            account,
                                            fill,
                                        }) => {
                                            if fill.time < bridge_start_ms {
                                                continue;
                                            }
                                            let manager = {
                                                let map = a2m.read().await;
                                                map.get(&account).copied()
                                            };
                                            let manager = match manager {
                                                Some(m) => m,
                                                None => match chain_auth_bridge
                                                    .get_manager(account.inner())
                                                    .await
                                                {
                                                    Ok(addr)
                                                        if addr
                                                            != alloy::primitives::Address::ZERO =>
                                                    {
                                                        let mgr = WalletAddress::from(addr);
                                                        a2m.write().await.insert(account, mgr);
                                                        mgr
                                                    }
                                                    _ => continue,
                                                },
                                            };
                                            if let (Ok(size), Ok(price)) =
                                                (fill.sz.parse::<f64>(), fill.px.parse::<f64>())
                                            {
                                                if size == 0.0 || price == 0.0 {
                                                    continue;
                                                }
                                                let ws_fill =
                                                    hypercall_api::websocket::WsFillUpdate {
                                                        order_id: fill.oid.unwrap_or(0) as i64,
                                                        fill_id: fill.time as i64,
                                                        symbol: format!("{}-PERP", fill.coin),
                                                        side: if fill.side == "B" {
                                                            "Buy"
                                                        } else {
                                                            "Sell"
                                                        }
                                                        .to_string(),
                                                        price:
                                                            rust_decimal::Decimal::from_f64_retain(
                                                                price,
                                                            )
                                                            .unwrap_or_default(),
                                                        size:
                                                            rust_decimal::Decimal::from_f64_retain(
                                                                size,
                                                            )
                                                            .unwrap_or_default(),
                                                        timestamp: fill.time as i64,
                                                        wallet_address: manager,
                                                        fee: fill
                                                            .fee
                                                            .as_deref()
                                                            .and_then(|f| f.parse().ok())
                                                            .unwrap_or_default(),
                                                        trade_id: fill.time as i64,
                                                        is_taker: true,
                                                        builder_code_address: None,
                                                        builder_code_fee: None,
                                                        instrument_type: "perp".to_string(),
                                                    };
                                                pubsub_for_bridge.publish_fill(ws_fill);

                                                if let Some(state) =
                                                    feed_ref.get_positions(&account).await
                                                {
                                                    if let Some(pos) = state
                                                        .positions
                                                        .iter()
                                                        .find(|p| p.coin == fill.coin)
                                                    {
                                                        let Some(entry_price) = pos.entry_price
                                                        else {
                                                            warn!(
                                                                account = %account,
                                                                coin = %pos.coin,
                                                                "PM bridge: skipping Hydromancer fill position update with missing entry price"
                                                            );
                                                            continue;
                                                        };
                                                        pc.handle_hypercore_position_update(
                                                            crate::portfolio::HypercorePositionUpdate {
                                                                account: manager.to_string(),
                                                                coin: pos.coin.clone(),
                                                                size: pos.size,
                                                                entry_price,
                                                                unrealized_pnl: pos.unrealized_pnl,
                                                                timestamp: state.last_updated_ms,
                                                                snapshot: false,
                                                            },
                                                        )
                                                        .await;
                                                    }
                                                    // Update HyperCore equity after fill
                                                    {
                                                        let account_value =
                                                            rust_decimal::Decimal::from_f64_retain(
                                                                state.account_value,
                                                            )
                                                            .unwrap_or_default();
                                                        let _ = equity_tx.send(
                                                            crate::rsm::unified_engine::HypercoreEquityRequest {
                                                                wallet: manager,
                                                                account_value,
                                                                timestamp_ms: state.last_updated_ms,
                                                            },
                                                        ).await;
                                                    }
                                                }
                                            }
                                        }
                                        Ok(crate::hypercore::HydromancerEvent::OrderUpdate {
                                            ..
                                        }) => {}
                                        Err(tokio::sync::broadcast::error::RecvError::Lagged(
                                            n,
                                        )) => {
                                            warn!(skipped = n, "Hydromancer→PM bridge lagged");
                                        }
                                        Err(_) => break,
                                    }
                                }
                            });
                            info!("✓ Hydromancer→PM position bridge started");
                        }

                        info!("✓ Hydromancer feed service started");
                        Some(feed)
                    }
                    None => {
                        warn!("Hydromancer enabled but no API key configured");
                        None
                    }
                }
            } else {
                None
            }
        },
        quote_provider: quote_provider.clone(),
        engine_state_digest_provider: Arc::new(
            crate::api_boundary_impls::RuntimeEngineStateDigestProvider::new(
                engine_state_digest_provider.clone(),
            ),
        ),
        order_snapshot: order_snapshot.clone(),
        competition_service: competition_service.clone(),
        candle_source: candle_source.clone(),
        underlying_to_candle_coin: underlying_to_candle_coin.clone(),
        bbo_snapshot_reader: bbo_snapshot_service.clone(),
        indicative_cache: Some(indicative_cache.clone()),
        rfq_manager: Some(rfq_manager.clone()),
        username_service: username_service.clone(),
        push_service: push_service.clone(),
        standby_controller: standby_controller.clone().map(|controller| {
            Arc::new(controller) as Arc<dyn hypercall_api::runtime_status::StandbyPromoter>
        }),
        notification_service: Some(notification_service.clone()),
        standby_promote: standby_promote.clone(),
        standby_progress: standby_progress.clone().map(|progress| {
            Arc::new(progress) as Arc<dyn hypercall_api::runtime_status::StandbyReplayProgress>
        }),
        startup_progress: startup_progress.clone().map(|progress| {
            Arc::new(progress) as Arc<dyn hypercall_api::runtime_status::StartupProgressReader>
        }),
        recovery_safety_report: recovery_safety_report.clone(),
        drain_signal: drain_signal.clone(),
        is_draining: is_draining.clone(),
        shutdown: shutdown.clone(),
    };

    if runtime.config.liquidation.enabled {
        let rsm_directive_publisher = crate::rsm_directive_publisher::RsmDirectivePublisher::new(
            engine_event_tx.clone(),
            journal_batch_sender.clone(),
        );
        let mut liquidation_executor_builder = crate::liquidator::LiquidationExecutor::new(
            liquidation_cache.clone(),
            portfolio_cache.clone(),
        )
        .with_event_sender(ws_direct_tx.clone())
        .with_rsm_directive_publisher(rsm_directive_publisher)
        .with_full_target_buffer_bps(runtime.config.liquidation.full_target_buffer_bps);
        if let Some(signer) = rsm_signer_service.clone() {
            liquidation_executor_builder = liquidation_executor_builder.with_rsm_signer(signer);
        }
        let liquidation_executor = Arc::new(liquidation_executor_builder);

        let liquidation_watcher = Arc::new(
            crate::liquidator::LiquidationWatcher::new(
                liquidation_cache.clone(),
                portfolio_cache.get_service(),
                tier_cache.clone(),
                span_margin_service.clone(),
                standard_margin_service.clone(),
                crate::liquidator::LiquidationWatcherConfig::from_runtime_config(
                    &runtime.config.liquidation,
                ),
            )
            .with_risk_account_builder(risk_account_builder.clone())
            .with_standard_account_builder(standard_account_builder.clone())
            .with_quote_provider(quote_provider.clone())
            .with_greeks_cache(greeks_cache.clone())
            .with_order_snapshot(order_snapshot.clone())
            .with_order_sender(order_tx.clone())
            .with_event_sender(ws_direct_tx.clone())
            .with_executor(liquidation_executor),
        );
        let liquidation_observer = Arc::new(
            crate::liquidator::LiquidationChainObserver::new(
                liquidation_cache.clone(),
                diesel_db.clone(),
                liquidation_bonus_tx.clone(),
                Some(ws_direct_tx.clone()),
                event_bus.clone(),
                &runtime.config.transaction_submitter.rpc_url,
                &runtime.config.contracts.exchange_contract_address,
                crate::liquidator::LiquidationObserverConfig::from_runtime_config(
                    &runtime.config.liquidation,
                ),
            )
            .map_err(|error| {
                anyhow::anyhow!("Failed to initialize liquidation chain observer: {}", error)
            })?,
        );

        if standby_mode {
            let watcher_service: Arc<dyn Service> = liquidation_watcher;
            let observer_service: Arc<dyn Service> = liquidation_observer;
            let mut services = post_promote_services.lock().await;
            services.push(watcher_service);
            services.push(observer_service);
            info!("Liquidation watcher and chain observer deferred until standby promotion");
        } else {
            service_registry.register(liquidation_watcher);
            service_registry.register(liquidation_observer);
            info!("✓ Liquidation watcher registered");
            info!("✓ Liquidation chain observer registered");
        }
    } else {
        info!("Liquidation runtime disabled by config");
    }

    if onchain_deposits_enabled {
        gauge!("ht_hypercore_cash_ledger_observer_enabled").set(1.0);
        let cash_ledger_observer = Arc::new(
            crate::hypercore_cash_ledger_observer::HypercoreCashLedgerObserver::new(
                shared_db.clone(),
                deposit_tx.clone(),
                crate::hypercore_cash_ledger_observer::HypercoreCashLedgerObserverConfig::from_runtime_config(
                    &runtime.config.onchain_deposits,
                    runtime.config.pricing.hypercore_info_url.clone(),
                    hypercall_types::WalletAddress::from(
                        alloy::primitives::Address::from_str(
                            &runtime.config.contracts.exchange_contract_address,
                        )
                        .expect("exchange_contract_address already validated at startup"),
                    ),
                    hypercall_types::WalletAddress::from(
                        alloy::primitives::Address::from_str(
                            &runtime.config.contracts.core_deposit_wallet_address,
                        )
                        .expect("core_deposit_wallet_address already validated at startup"),
                    ),
                ),
                tier_cache.clone(),
            )
            .map_err(|error| {
                anyhow::anyhow!("Failed to initialize HyperCore cash ledger observer: {}", error)
            })?,
        );

        if standby_mode {
            let observer_service: Arc<dyn Service> = cash_ledger_observer;
            let mut services = post_promote_services.lock().await;
            services.push(observer_service);
            info!("HyperCore cash ledger observer deferred until standby promotion");
        } else {
            service_registry.register(cash_ledger_observer);
            info!("✓ HyperCore cash ledger observer registered");
        }

        gauge!("ht_rsm_deposit_credit_observer_enabled").set(1.0);
        let deposit_credit_observer = Arc::new(
            crate::rsm_deposit_credit_observer::RsmDepositCreditObserver::new(
                shared_db.clone(),
                option_deposit_tx.clone(),
                engine_pm_settlement_admin_tx.clone(),
                &runtime.config.transaction_submitter.rpc_url,
                &runtime.config.contracts.exchange_contract_address,
                &runtime.config.contracts.usdc_address,
                crate::rsm_deposit_credit_observer::RsmDepositCreditObserverConfig::from_runtime_config(
                    &runtime.config.onchain_deposits,
                ),
            )
            .map_err(|error| {
                anyhow::anyhow!("Failed to initialize RSM deposit credit observer: {}", error)
            })?,
        );

        if standby_mode {
            let observer_service: Arc<dyn Service> = deposit_credit_observer;
            let mut services = post_promote_services.lock().await;
            services.push(observer_service);
            info!("RSM deposit credit observer deferred until standby promotion");
        } else {
            service_registry.register(deposit_credit_observer);
            info!("✓ RSM deposit credit observer registered");
        }
    } else {
        gauge!("ht_hypercore_cash_ledger_observer_enabled").set(0.0);
        gauge!("ht_rsm_deposit_credit_observer_enabled").set(0.0);
        info!(
            "HyperCore cash ledger and RSM deposit credit observers disabled because transaction submission is mocked"
        );
    }

    // Create rate limit state for middleware
    let rate_limit_state = middleware::RateLimitState {
        rate_limiter: rate_limit_cache.clone(),
    };

    // Create WebSocket state (after RFQ system so rfq_handler_state is available)
    let ws_state = WsState {
        pubsub: (*pubsub).clone(),
        indicative_cache: Some(indicative_cache.clone()),
        portfolio_cache: Some(portfolio_cache.clone()),
        competition_service: Some(competition_service.clone()),
        heartbeat_config: hypercall_api::websocket::ws_heartbeat_config(&runtime.config.api),
        rfq_handler_state: Some(rfq_handler_state.clone()),
        order_sender: Some(order_tx.clone()),
        trading_halt: Some(trading_halt.clone()),
        agent_auth: agent_auth.clone(),
        signing_chain_id,
    };

    let candle_ws_poll_interval_ms = candle_ws_poll_interval_ms(&runtime.config.pricing);
    let candle_publisher = CandleWsPublisher::new(
        pubsub.clone(),
        candle_source.clone(),
        underlying_to_candle_coin.clone(),
        Duration::from_millis(candle_ws_poll_interval_ms),
    );
    service_registry.register(Arc::new(candle_publisher));
    tracing::info!(
        "✓ CandleWsPublisher registered (poll_interval_ms={})",
        candle_ws_poll_interval_ms
    );

    let index_price_publisher =
        IndexPricePublisher::new(greeks_cache.clone(), pubsub.clone(), spot_price_notify);
    service_registry.register(Arc::new(index_price_publisher));
    tracing::info!("✓ IndexPricePublisher registered");

    // Start MetricsCollector for business/trading metrics
    let metrics_collector = Arc::new(MetricsCollector::new(
        MetricsCollectorConfig::from_config(&runtime.config.observability.metrics),
        portfolio_cache.clone(),
        quote_provider.clone(),
        Some(order_snapshot.clone()),
        Some(tier_cache.clone()),
        portfolio_margin_settlement_allowlist,
        Some(greeks_cache.clone()),
        Some(market_stats_cache.clone()),
        Some(shared_db.clone()),
        Some(diesel_db.clone()),
        Some(diesel_db.clone()),
        Some(diesel_db.clone()),
        Some(risk_vol_oracle.clone()),
    ));
    service_registry.register(metrics_collector);
    tracing::info!("✓ MetricsCollector registered");

    // Start all registered services
    let service_count = service_registry.len();
    service_registry.start_all(&shutdown, &mut task_group);
    info!("✓ ServiceRegistry started {} services", service_count);

    // Setup graceful shutdown (Ctrl-C and SIGTERM listener)
    // This task listens for Ctrl-C or SIGTERM and also exits if shutdown is triggered elsewhere
    let shutdown_for_signal = shutdown.clone();
    let mut shutdown_rx_signal = shutdown.subscribe();
    task_group.spawn("SignalListener", async move {
        // Create a future that resolves on either SIGINT (Ctrl-C) or SIGTERM
        let ctrl_c = async {
            signal::ctrl_c()
                .await
                .expect("Failed to setup SIGINT handler");
            "SIGINT"
        };

        #[cfg(unix)]
        let terminate = async {
            use tokio::signal::unix::{signal, SignalKind};
            signal(SignalKind::terminate())
                .expect("Failed to setup SIGTERM handler")
                .recv()
                .await;
            "SIGTERM"
        };

        #[cfg(not(unix))]
        let terminate = std::future::pending::<&str>();

        tokio::select! {
            sig = ctrl_c => {
                tracing::info!("Received {} (Ctrl-C), initiating graceful shutdown...", sig);
                shutdown_for_signal.trigger();
            }
            sig = terminate => {
                tracing::info!("Received {}, initiating graceful shutdown...", sig);
                shutdown_for_signal.trigger();
            }
            _ = shutdown_rx_signal.recv() => {
                tracing::info!("SignalListener received shutdown signal, exiting...");
            }
        }
        Ok(())
    });

    // The complete admin surface lives in `hypercall_admin::admin_router` and is
    // wrapped here in the single admin auth layer.
    let admin_state = build_admin_state(&app_state, rfq_handler_state.qp_cache.clone());
    let admin_router = hypercall_admin::admin_router(admin_state.clone()).layer(
        axum::middleware::from_fn_with_state(
            admin_state,
            hypercall_admin::auth::monitoring_auth_middleware,
        ),
    );

    // The live Swagger/Scalar spec must include the admin crate's paths (e.g. the
    // Health-tagged /recovery-safety endpoints), matching what api-spec-exporter
    // produces. api cannot merge admin_openapi() itself (api -> admin edge), so the
    // root composes the public spec here, the same way it composes the routers.
    let openapi_doc = hypercall_api::openapi::filter_hidden_tags({
        let mut doc = hypercall_api::openapi::internal_openapi();
        doc.merge(hypercall_admin::admin_openapi());
        doc
    });

    let app = hypercall_api::routes::build_app(hypercall_api::routes::BuildAppParams {
        app_state,
        rate_limit_state,
        ws_state,
        qp_ws_state,
        rfq_handler_state,
        readiness: readiness.clone() as Arc<dyn hypercall_api::runtime_status::ReadinessGate>,
        standby_progress: standby_progress.clone().map(|progress| {
            Arc::new(progress) as Arc<dyn hypercall_api::runtime_status::StandbyReplayProgress>
        }),
        standby_promote: standby_promote.clone(),
        admin_router,
        openapi_doc,
    });

    startup_timing!("total", startup_begin.elapsed());
    tracing::info!("Integrated API server listening on {}", addr);

    if let Some(bootstrap_http) = bootstrap_http {
        if let Some(progress) = startup_progress.as_ref() {
            progress.mark("full_api_router_installed");
        }
        if let Some(task) = standby_startup_heartbeat {
            task.abort();
        }
        *bootstrap_http.full_app.write().await = Some(app);
        tracing::info!("Full API router installed behind bootstrap standby HTTP server");
        if let Err(error) = bootstrap_http.task.await {
            tracing::error!("Bootstrap standby HTTP server task failed: {}", error);
        }
    } else {
        let listener = listener.expect("non-standby HTTP server requires the bound listener");
        let mut http_shutdown_rx = shutdown.subscribe();
        let server = axum::serve(listener, app);

        tokio::select! {
            result = server => {
                if let Err(e) = result {
                    tracing::error!("Server error: {}", e);
                }
            }
            _ = http_shutdown_rx.recv() => {
                tracing::info!("Shutting down HTTP server...");
            }
        }
    }

    // Shutdown all background tasks and wait for them to complete
    // Note: shutdown may already be triggered (by Ctrl-C), but trigger is idempotent
    task_group
        .shutdown_and_join(&shutdown, Duration::from_secs(15))
        .await?;

    tracing::info!("Server shutdown complete");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{build_bootstrap_http_app, BootstrapHttpState, StartupProgress};
    use crate::observability::command_trace::EngineStateDigest;
    use crate::shared::shutdown::Shutdown;
    use alloy::primitives::Address;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use axum::routing::get;
    use axum::Router;
    use hypercall_api::websocket::event_forwarder::ws_orderbook_update_from_engine_update;
    use hypercall_types::wallet_address::test_wallet;
    use hypercall_types::OrderbookUpdate;
    use hypercall_types::WalletAddress;
    use hypercall_types::{EngineMessage, SignedDirectiveTx, TransactionRequest, TransactionType};
    use rust_decimal_macros::dec;
    use std::str::FromStr;
    use std::sync::Arc;
    use tower::util::ServiceExt;

    #[tokio::test]
    async fn bootstrap_http_serves_probe_routes_before_full_app() {
        let full_app = Arc::new(tokio::sync::RwLock::new(None));
        let standby_progress = crate::nats::replay_loop::ReplayProgress::new();
        let startup_progress = StartupProgress::new("test_starting");
        startup_progress.mark("test_hydrating");
        let (promote_tx, _promote_rx) = tokio::sync::oneshot::channel::<()>();
        let app = build_bootstrap_http_app(BootstrapHttpState {
            shutdown: Shutdown::new(),
            standby_progress: Some(standby_progress),
            standby_promote: Some(Arc::new(tokio::sync::Mutex::new(Some(promote_tx)))),
            startup_progress: Some(startup_progress.clone()),
            full_app: full_app.clone(),
        });

        let health = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/health")
                    .body(Body::empty())
                    .expect("valid health request"),
            )
            .await
            .expect("health response");
        assert_eq!(health.status(), StatusCode::OK);

        let standby_ready = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/standby-ready")
                    .body(Body::empty())
                    .expect("valid standby-ready request"),
            )
            .await
            .expect("standby-ready response");
        assert_eq!(standby_ready.status(), StatusCode::SERVICE_UNAVAILABLE);
        let standby_body = axum::body::to_bytes(standby_ready.into_body(), usize::MAX)
            .await
            .expect("standby-ready body");
        let standby_payload: serde_json::Value =
            serde_json::from_slice(&standby_body).expect("standby-ready json");
        assert_eq!(standby_payload["status"], "api_starting");
        assert_eq!(standby_payload["api_router_ready"], false);
        assert_eq!(standby_payload["promotable"], false);
        assert_eq!(standby_payload["startup_phase"], "test_hydrating");
        assert_eq!(standby_payload["startup_progress_counter"], 2);
        assert!(standby_payload["last_startup_progress_unix_ms"]
            .as_u64()
            .is_some());
        assert!(standby_payload["last_startup_progress_age_ms"]
            .as_u64()
            .is_some());

        let fallback_while_starting = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/delegated")
                    .body(Body::empty())
                    .expect("valid fallback request"),
            )
            .await
            .expect("fallback response");
        assert_eq!(
            fallback_while_starting.status(),
            StatusCode::SERVICE_UNAVAILABLE
        );

        let caught_up_progress = crate::nats::replay_loop::ReplayProgress::new();
        caught_up_progress.test_update_catch_up_state(10, 10);
        let (caught_up_promote_tx, _caught_up_promote_rx) = tokio::sync::oneshot::channel::<()>();
        let caught_up_app = build_bootstrap_http_app(BootstrapHttpState {
            shutdown: Shutdown::new(),
            standby_progress: Some(caught_up_progress),
            standby_promote: Some(Arc::new(tokio::sync::Mutex::new(Some(
                caught_up_promote_tx,
            )))),
            startup_progress: Some(startup_progress),
            full_app: full_app.clone(),
        });
        let caught_up_before_router = caught_up_app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/standby-ready")
                    .body(Body::empty())
                    .expect("valid caught-up standby-ready request"),
            )
            .await
            .expect("caught-up standby-ready response");
        assert_eq!(
            caught_up_before_router.status(),
            StatusCode::SERVICE_UNAVAILABLE
        );

        *full_app.write().await = Some(Router::new().route("/delegated", get(|| async { "ok" })));
        let caught_up_after_router = caught_up_app
            .oneshot(
                Request::builder()
                    .uri("/standby-ready")
                    .body(Body::empty())
                    .expect("valid caught-up standby-ready request after router install"),
            )
            .await
            .expect("caught-up standby-ready response after router install");
        assert_eq!(caught_up_after_router.status(), StatusCode::OK);
        let caught_up_body = axum::body::to_bytes(caught_up_after_router.into_body(), usize::MAX)
            .await
            .expect("caught-up standby-ready body");
        let caught_up_payload: serde_json::Value =
            serde_json::from_slice(&caught_up_body).expect("caught-up standby-ready json");
        assert_eq!(caught_up_payload["status"], "standby_ready");
        assert_eq!(caught_up_payload["startup_phase"], "test_hydrating");
        assert_eq!(caught_up_payload["startup_progress_counter"], 2);

        let delegated = app
            .oneshot(
                Request::builder()
                    .uri("/delegated")
                    .body(Body::empty())
                    .expect("valid delegated request"),
            )
            .await
            .expect("delegated response");
        assert_eq!(delegated.status(), StatusCode::OK);
    }

    #[test]
    fn test_ws_orderbook_update_converts_contract_units_to_human_contracts() {
        let orderbook_update_contract_units_raw = OrderbookUpdate {
            symbol: "BTC-20260331-100000-C".to_string(),
            bids: vec![(dec!(100), dec!(1_000_000)), (dec!(99), dec!(2_500_000))],
            asks: vec![(dec!(101), dec!(500_000)), (dec!(102), dec!(3_000_000))],
            timestamp: 1_750_000_000_000_u64,
        };

        let ws_update = ws_orderbook_update_from_engine_update(
            &orderbook_update_contract_units_raw,
            Some(test_wallet(7)),
        );

        assert_eq!(ws_update.bids[0].1, dec!(1));
        assert_eq!(ws_update.bids[1].1, dec!(2.5));
        assert_eq!(ws_update.asks[0].1, dec!(0.5));
        assert_eq!(ws_update.asks[1].1, dec!(3));
    }

    #[test]
    fn test_disabled_wal_startup_provides_mock_journal_writer_for_directives() {
        let (engine_journal_writer, journal_batch_sender) = crate::journal::mock_journal_backends();
        assert!(
            journal_batch_sender.is_none(),
            "disabled WAL mode should not configure a batch sender"
        );

        let journal_writer = engine_journal_writer
            .expect("disabled WAL mode should inject a mock writer for directive submits");
        let wallet = WalletAddress::from(
            Address::from_str("0x1111111111111111111111111111111111111111").expect("valid wallet"),
        );
        let request_id = uuid::Uuid::now_v7().to_string();
        let tx_request = TransactionRequest {
            request_id: request_id.clone(),
            wallet_address: wallet,
            account_contract: wallet,
            transaction_type: TransactionType::UserDirective(SignedDirectiveTx {
                directive: vec![1, 2, 3],
                signature: "0xabcdef".to_string(),
            }),
            timestamp: 123,
            expires_at: 456,
        };
        let result = journal_writer.append_transition(
            tx_request.timestamp,
            &hypercall_types::serialize_to_wire_bytes(&serde_json::json!({
                "account": wallet,
                "actionKey": "hc_update_api_wallet",
                "nonce": 42,
                "recoveredSigner": wallet,
            })),
            None,
            None,
            &EngineStateDigest::default(),
            &EngineStateDigest::default(),
            0,
            &[EngineMessage::TransactionRequest(tx_request)],
            hypercall_db_diesel::engine_enums::DbUuid(
                uuid::Uuid::parse_str(&request_id).expect("request id should be a UUID"),
            ),
            None,
        );

        assert!(
            result.is_ok(),
            "disabled WAL startup writer should be usable for directive journaling"
        );
    }
}