futu_opend/startup/

phase4.rs

//! v1.4.110 Layer 3 A: startup Phase 4 — surface server (WS/REST/gRPC/Telnet)
//! spawn / card_num expand / SIGHUP unified handler / TCP fail-closed gate /
//! 主 `tokio::select!` 循环 + 清理. 抽自原 `mod.rs::run_daemon` 533..1075 行段.
//!
//! Phase 4 取走 `Phase1Out` 所有权 (保证 `_audit_guard` 直到 fn 返回才 drop),
//! 同时取 `bridge: Arc<GatewayBridge>` 与 `Phase3Out` 启动各 surface server.

use anyhow::Result;
use std::sync::Arc;

use futu_gateway_core::bridge::GatewayBridge;
use futu_server::listener_status::ListenerBindEvent;
use futu_server::ws_listener::{WsServer, WsServerDeps};

use crate::config::RuntimeConfig;
use crate::startup::phase1::Phase1Out;
use crate::startup::phase2::{AuthPlan, execute_auth_plan};
use crate::startup::phase3::Phase3Out;

mod listener_readiness;
mod network_exposure;
mod shutdown;
mod summary;
#[cfg(test)]
mod tests;

use listener_readiness::{
    ActivatedListenerReadiness, EnabledListenerSurfaces, ListenerReadiness, SurfaceTasks,
    await_listeners_opened, wrap_grpc_server_result,
};
#[cfg(test)]
use network_exposure::legacy_network_exposure_warnings;
use network_exposure::warn_legacy_network_exposure;
use shutdown::{
    BackgroundJoinHandle, SHUTDOWN_GRACE_PERIOD, SurfaceJoinHandle, await_background_shutdown,
    await_surface_shutdown, log_surface_task_result, request_surface_shutdown,
    surface_task_result_or_pending,
};
#[cfg(test)]
use summary::http_host_for_bind;
use summary::{RestTransportKind, log_startup_summary, merge_push_health_snapshots_for_rest};

include!("phase4/preauth.rs");

pub(super) async fn run_phase4(
    config: &RuntimeConfig,
    phase1: Phase1Out,
    bridge: Arc<GatewayBridge>,
    mut auth_plan: Option<AuthPlan>,
    phase3: Phase3Out,
    shutdown_tx: tokio::sync::watch::Sender<bool>,
    mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<()> {
    let Phase1Out {
        _audit_guard,
        shared_counters,
        listen_addr,
        rest_keys_file,
        rest_key_store,
        rest_transport,
        ws_keys_file,
        ws_key_store,
        grpc_keys_file,
        grpc_key_store,
        allow_tcp_unauthenticated,
    } = phase1;
    let rest_transport_kind = match config.rest_port {
        None => RestTransportKind::Disabled,
        Some(_) if rest_transport.is_tls() => RestTransportKind::Tls,
        Some(_) => RestTransportKind::Plaintext,
    };
    let Phase3Out {
        server,
        server_config,
        ws_broadcaster,
        grpc_broadcaster,
    } = phase3;
    let push_sinks: Vec<std::sync::Arc<dyn futu_server::push::ExternalPushSink>> = vec![
        std::sync::Arc::clone(&ws_broadcaster) as _,
        std::sync::Arc::clone(&grpc_broadcaster) as _,
    ];
    let prepared_push_dispatcher = bridge.create_push_dispatcher(&server, push_sinks);
    let (listener_events_tx, mut listener_events_rx) =
        tokio::sync::mpsc::unbounded_channel::<ListenerBindEvent>();

    // v1.4.103 (B10): 把 3 个 server 的 key_store Option 提升到外层作用域,
    // 让后面的 expand_allowed_card_nums spawn task 能跨 server 块捕获.
    // 各 server 块仍各自加载 (允许独立 keys.json), 这里只共享 Arc clone.
    let mut ws_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let mut rest_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let mut grpc_key_store_holder: Option<std::sync::Arc<futu_auth::KeyStore>> = None;
    let any_keys_configured =
        rest_keys_file.is_some() || grpc_keys_file.is_some() || ws_keys_file.is_some();
    warn_legacy_network_exposure(config, any_keys_configured);

    // 7. 启动 WebSocket 服务（可选）
    let mut ws_handle = if let Some(ws_port) = config.websocket_port {
        let ws_addr = format!("{}:{}", config.ip, ws_port);
        // v1.0：WS 握手鉴权 —— 复用 REST 的 key store 设计（`--ws-keys-file` 独立
        // 指定，不指定时 legacy 放行）
        // v1.4.102 BUG-007 fix (P1, leaf v1.4.100 报告): keys-file load 失败
        // 必须 fail-closed (abort daemon), 不再 silent fallback to legacy mode.
        // 历史: 用户传 `--ws-keys-file` 表示明确意图启用 auth, 文件 typo / parse
        // 错时 daemon 只 log error 然后继续无 auth → 用户以为有门锁, 实际门锁
        // 没装上 (security misconfig 比纯 legacy 更危险).
        let ws_key_store = ws_key_store.clone();
        let ws_counters = std::sync::Arc::clone(&shared_counters);
        // v1.4.103 (B10): clone Arc 给外层 holder 持有, 同时让 ws_server 仍接管原 Arc.
        ws_key_store_holder = ws_key_store.as_ref().map(std::sync::Arc::clone);
        let ws_server = WsServer::with_auth(
            ws_addr.clone(),
            server_config.clone(),
            WsServerDeps::new(
                std::sync::Arc::clone(server.connections()),
                std::sync::Arc::clone(server.router()),
                Some(bridge.subscription_runtime().manager()),
            ),
            ws_key_store,
            Some(ws_counters),
        )
        .with_server_time_store(bridge.server_clock().anchor_store());
        tracing::info!(addr = %ws_addr, "starting WebSocket server");
        let ws_shutdown_rx = shutdown_rx.clone();
        let ws_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            ws_server
                .run_until_shutdown_with_listener_events(ws_shutdown_rx, Some(ws_listener_events))
                .await
        }))
    } else {
        None
    };

    // 8. 启动 REST API 服务（可选，含 WebSocket 推送）
    let mut rest_handle = if let Some(rest_port) = config.rest_port {
        let rest_addr = format!("{}:{}", config.ip, rest_port);
        let router = std::sync::Arc::clone(server.router());
        let broadcaster = std::sync::Arc::clone(&ws_broadcaster);
        // v1.4.102 BUG-007 fix: 同 WS 路径 (fail-closed when keys-file 显式指定).
        let rest_key_store = rest_key_store
            .clone()
            .unwrap_or_else(|| std::sync::Arc::new(futu_auth::KeyStore::empty()));
        let rest_transport_name = match rest_transport_kind {
            RestTransportKind::Tls => "https",
            RestTransportKind::Plaintext => "http",
            RestTransportKind::Disabled => unreachable!("REST port is configured"),
        };
        tracing::info!(
            addr = %rest_addr,
            transport = rest_transport_name,
            "starting REST API server (WebSocket: /ws)"
        );

        // v1.4.103 (B10): clone Arc 给外层 holder 持有 (跨 server 块共享给
        // expand_allowed_card_nums spawn task). 仅在 keys 实际配置时填充
        // (empty store 不需要 expand).
        if rest_key_store.is_configured() {
            rest_key_store_holder = Some(std::sync::Arc::clone(&rest_key_store));
        }

        // v1.4.103 codex F3.1 (P1) round 3: REST 单独的 SIGHUP reload listener
        // 已**移除** — 防与 card_num expand 间的 race (多 listener 并发顺序乱
        // 可能让 reload 覆盖 expand 写入的 sentinel, 受限 key 在 reload 窗口
        // 内 silent unrestricted). 现在 reload + expand 由文件末尾的 unified
        // SIGHUP handler 顺序操作 (调 card_num_reload_and_expand_fn(true)).

        let rest_counters = std::sync::Arc::clone(&shared_counters);
        // v1.4.32+ admin snapshot provider：closure 捕获 `Arc<GatewayBridge>`
        // 每次被 admin_status handler 调用时返回实时 StatusSnapshot JSON。
        // bridge 已经在上面 Arc 化，这里只做一次 clone。
        let bridge_for_status = std::sync::Arc::clone(&bridge);
        let admin_status_provider: futu_rest::adapter::AdminStatusProvider =
            std::sync::Arc::new(move || {
                serde_json::to_value(bridge_for_status.snapshot_status())
                    .unwrap_or_else(|_| serde_json::json!({"error": "snapshot serialize failed"}))
            });
        let rest_shutdown_tx = shutdown_tx.clone();
        let admin_shutdown_handler: futu_rest::adapter::AdminShutdownHandler =
            std::sync::Arc::new(move || {
                rest_shutdown_tx
                    .send(true)
                    .map_err(|e| format!("shutdown receiver dropped: {e}"))
            });
        // v1.4.32+ admin reload handler：closure 调 Bridge::reload() 清 cipher cache
        // v1.4.34: reload 升级为 async（内部刷 credentials 走网络 I/O），
        // handler 返 Future，axum admin_reload async handler await 之
        // v1.4.106 codex 0554 F3 [P2]: reload 拆两阶段后变 sync (sync clear +
        // tokio::spawn refresh). closure 仍返 Future 类型不变 (向后兼容
        // AdminReloadHandler API), 但内部不 await — sync 阶段已完成 + spawn
        // 已派发, ReloadReport 立即可用.
        let bridge_for_reload = std::sync::Arc::clone(&bridge);
        let admin_reload_handler: futu_rest::adapter::AdminReloadHandler =
            std::sync::Arc::new(move || {
                let bridge = std::sync::Arc::clone(&bridge_for_reload);
                Box::pin(async move {
                    serde_json::to_value(bridge.reload())
                        .unwrap_or_else(|_| serde_json::json!({"error": "reload serialize failed"}))
                })
            });
        // v1.4.83 §9 Phase 2 F5: push health snapshot provider —— closure
        // 捕获 bridge.push_runtime().push_health Arc, 每次
        // `/api/push-subscriber-info` 调用时返当前真实 snapshot.
        // v1.4.91 P1-D wiring: closure 同时捕获 bridge.push_runtime().qot_login_health,
        // 在返 push_health 同时多带一个 qot_login_health 字段供 ops 看
        // qot_logined self-heal counter (修 P1-D non-deterministic gap).
        let bridge_for_push_health = std::sync::Arc::clone(&bridge);
        let push_health_snapshot_provider: futu_rest::adapter::PushHealthSnapshotProvider =
            std::sync::Arc::new(move || {
                let push = serde_json::to_value(
                    bridge_for_push_health
                        .push_runtime()
                        .push_health()
                        .snapshot(),
                )
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "push_health snapshot serialize failed"}),
                );
                let qot_login = serde_json::to_value(
                    bridge_for_push_health
                        .push_runtime()
                        .qot_login_health()
                        .snapshot(),
                )
                .unwrap_or_else(
                    |_| serde_json::json!({"error": "qot_login_health snapshot serialize failed"}),
                );
                let backend_connected =
                    bridge_for_push_health.broker_runtime().platform_connected();
                let stock_list_status = bridge_for_push_health
                    .caches()
                    .static_cache
                    .stock_list_sync_status();
                let stock_list_static_query_ready = matches!(
                    futu_domain_static_data::stock_list_static_query_gate_action_for_status(
                        &stock_list_status.domain_facts(),
                    ),
                    futu_domain_static_data::StockListStaticQueryGateAction::Allow
                );
                merge_push_health_snapshots_for_rest(
                    push,
                    qot_login,
                    backend_connected,
                    stock_list_static_query_ready,
                )
            });
        // v1.4.105 D12 (Phase 2): 注入 card_num resolver 让 REST trade
        // handler (place_order / modify_order / cancel_all_order) 能解析
        // user 传 `card_num` 字段 → acc_id (覆盖 c2s.header.acc_id).
        // closure 捕获 bridge.caches().trd_cache, 调
        // `find_acc_ids_by_card_num(input) -> Vec<u64>`.
        let bridge_for_card_num = std::sync::Arc::clone(&bridge);
        let card_num_resolver: futu_rest::adapter::CardNumResolver =
            std::sync::Arc::new(move |cn: &str| {
                bridge_for_card_num
                    .caches()
                    .trd_cache
                    .find_acc_ids_by_card_num(cn)
            });
        let rest_shutdown_rx = shutdown_rx.clone();
        let rest_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            futu_rest::server::start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
                &rest_addr,
                router,
                broadcaster,
                rest_key_store,
                rest_counters,
                futu_rest::server::RestAdminHooks {
                    admin_status_provider: Some(admin_status_provider),
                    admin_shutdown_handler: Some(admin_shutdown_handler),
                    admin_reload_handler: Some(admin_reload_handler),
                    push_health_snapshot_provider: Some(push_health_snapshot_provider),
                    card_num_resolver: Some(card_num_resolver),
                },
                rest_transport,
                rest_shutdown_rx,
                Some(rest_listener_events),
            )
            .await
            .map_err(anyhow::Error::from)
        }))
    } else {
        None
    };

    // 9. 启动 gRPC 服务（可选，含流式推送）
    let mut grpc_handle = if let Some(grpc_port) = config.grpc_port {
        let grpc_addr = format!("{}:{}", config.ip, grpc_port);
        let router = std::sync::Arc::clone(server.router());
        let broadcaster = std::sync::Arc::clone(&grpc_broadcaster);
        // v1.4.102 BUG-007 fix: 同 WS / REST 路径 (fail-closed when keys-file 显式指定).
        let grpc_key_store = grpc_key_store
            .clone()
            .unwrap_or_else(|| std::sync::Arc::new(futu_auth::KeyStore::empty()));
        tracing::info!(addr = %grpc_addr, "starting gRPC server (SubscribePush: streaming)");

        // v1.4.103 (B10): clone Arc 给外层 holder 持有.
        if grpc_key_store.is_configured() {
            grpc_key_store_holder = Some(std::sync::Arc::clone(&grpc_key_store));
        }

        // v1.4.103 codex F3.1 (P1) round 3: gRPC 单独的 SIGHUP reload listener
        // 已**移除** — 同 REST 块, 由文件末尾的 unified SIGHUP handler 顺序
        // reload + expand 防 race.

        let grpc_counters = std::sync::Arc::clone(&shared_counters);
        let grpc_shutdown_rx = shutdown_rx.clone();
        let grpc_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            let result = futu_grpc::server::start_with_auth_until_shutdown_with_listener_events(
                &grpc_addr,
                router,
                broadcaster,
                grpc_key_store,
                grpc_counters,
                grpc_shutdown_rx,
                Some(grpc_listener_events),
            )
            .await;
            wrap_grpc_server_result(result)
        }))
    } else {
        None
    };

    // 10. 启动 Telnet 管理服务（可选）
    let telnet_addr = telnet_bind_addr(&config.telnet_ip, config.telnet_port);
    let mut telnet_handle = if let Some(telnet_addr) = telnet_addr {
        // v1.4.97 P1-D-F: relogin callback closes over bridge to clear
        // login_cache; next 30s P1-D health tick triggers relogin.
        // Aligns with C++ GTWCmd_ReLogin (FTGateway/FTGTW_Define_Key.h:5).
        let bridge_for_relogin = std::sync::Arc::clone(&bridge);
        let relogin_fn: futu_server::telnet::ReloginFn = std::sync::Arc::new(move || {
            tracing::warn!(
                "v1.4.97 P1-D-F: telnet relogin clearing login_cache; \
                     next P1-D tick will trigger central auth-session refresh"
            );
            bridge_for_relogin.caches().login_cache.clear();
        });
        let telnet_server = futu_server::telnet::TelnetServer::new(
            telnet_addr.clone(),
            std::sync::Arc::clone(server.connections()),
            Some(bridge.subscription_runtime().manager()),
            Some(std::sync::Arc::clone(server.metrics())),
            shutdown_tx.clone(),
        )
        .with_relogin_fn(relogin_fn);
        tracing::info!(addr = %telnet_addr, "starting Telnet server");
        let telnet_shutdown_rx = shutdown_rx.clone();
        let telnet_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            telnet_server
                .run_until_shutdown_with_listener_events(
                    telnet_shutdown_rx,
                    Some(telnet_listener_events),
                )
                .await
        }))
    } else {
        None
    };

    // v1.4.103 (B10) + codex F2 (P1): 立即跑首次 expand + background retry
    // + SIGHUP reload 钩子 (fail-closed sentinel 即时生效, 无 startup window).
    //
    // 行为:
    // 1. **立即跑首次 expand** — 即便 cache 空, fail-closed sentinel (codex F1)
    //    也写进 allowed_acc_ids, 短路 startup window 的 silent unrestricted.
    // 2. **background retry**: 每 10s 检查 cache, 加载后 re-expand 真 acc_id
    //    覆盖 sentinel. 60s 上限.
    // 3. **SIGHUP reload 钩子**: 重载 keys.json 后再 expand 一次 (codex F2 reload window).
    // v1.4.103 codex F3.1 (P1) round 3: 合并 reload + expand 为单一 ordered op,
    // 防 race. 当 SIGHUP 触发时, 此 fn 同步顺序: (1) reload 所有 store (从
    // keys.json 读 raw allowed_card_nums + ArcSwap.store) (2) expand (resolve
    // card_num → acc_ids + ArcSwap.store with sentinel/resolved). 因为是单线
    // 调用, 不存在多 SIGHUP listener 间的乱序 race.
    //
    // `do_reload` 控制是否先 reload (启动时第一次 / 60s retry 不需 reload,
    // 直接 expand; SIGHUP 路径需要先 reload).
    let card_num_reload_and_expand_fn: std::sync::Arc<dyn Fn(bool) + Send + Sync> = {
        let bridge_for_expand = std::sync::Arc::clone(&bridge);
        let ws_ks = ws_key_store_holder.clone();
        let rest_ks = rest_key_store_holder.clone();
        let grpc_ks = grpc_key_store_holder.clone();
        std::sync::Arc::new(move |do_reload: bool| {
            // (1) Reload phase — 仅 SIGHUP 路径调
            if do_reload {
                for (ks_name, ks_opt) in [("ws", &ws_ks), ("rest", &rest_ks), ("grpc", &grpc_ks)] {
                    let Some(ks) = ks_opt.as_ref() else { continue };
                    match ks.reload() {
                        Ok(()) => tracing::warn!(
                            ks = ks_name,
                            keys_loaded = ks.len(),
                            "v1.4.103 F3.1: keys reloaded on SIGHUP (before card_num expand)"
                        ),
                        Err(e) => tracing::error!(
                            ks = ks_name,
                            error = %e,
                            "v1.4.103 F3.1: keys reload failed (skipping expand for this store)"
                        ),
                    }
                }
            }
            // (2) Expand phase
            let trd_cache = std::sync::Arc::clone(&bridge_for_expand.caches().trd_cache);
            let resolver = {
                let cache_clone = std::sync::Arc::clone(&trd_cache);
                move |cn: &str| cache_clone.find_acc_ids_by_card_num(cn)
            };
            for (ks_name, ks_opt) in [("ws", &ws_ks), ("rest", &rest_ks), ("grpc", &grpc_ks)] {
                let Some(ks) = ks_opt.as_ref() else { continue };
                let (resolved, unresolved, ambiguous) = ks.expand_allowed_card_nums(
                    &resolver,
                    |key_id, cn| {
                        tracing::warn!(
                            key_id = %key_id,
                            card_num = %cn,
                            "v1.4.103 B10/F1 fail-closed: card_num not found in trd_cache; \
                             writing sentinel acc_id=0 to enforce restrictive denylist \
                             (limits.contains check 永远 false → reject 真账户)"
                        );
                    },
                    |key_id, cn, candidates| {
                        tracing::warn!(
                            key_id = %key_id,
                            card_num = %cn,
                            candidates = ?candidates,
                            "v1.4.103 B10/F1 fail-closed: ambiguous card_num suffix \
                             matched multiple accounts (skipped, write 完整 16 位 / specific 4 位)"
                        );
                    },
                );
                tracing::info!(
                    ks = ks_name,
                    resolved,
                    unresolved,
                    ambiguous,
                    "v1.4.103 B10: expanded allowed_card_nums into allowed_acc_ids"
                );
            }
        })
    };
    // 老 alias (不 reload, 仅 expand) 用于启动 + retry 路径
    let card_num_expand_fn: std::sync::Arc<dyn Fn() + Send + Sync> = {
        let inner = std::sync::Arc::clone(&card_num_reload_and_expand_fn);
        std::sync::Arc::new(move || (inner)(false))
    };

    // codex F2 (P1) 立即跑首次: fail-closed sentinel 即时生效, 不留 startup window
    (card_num_expand_fn)();

    // background retry loop: cache 加载后 re-expand 覆盖 sentinel
    let card_num_retry_handle: BackgroundJoinHandle = {
        let card_num_expand_fn_loop = std::sync::Arc::clone(&card_num_expand_fn);
        let bridge_for_check = std::sync::Arc::clone(&bridge);
        let mut card_num_retry_shutdown_rx = shutdown_rx.clone();
        tokio::spawn(async move {
            let trd_cache = std::sync::Arc::clone(&bridge_for_check.caches().trd_cache);
            let mut attempts = 0u32;
            let max_attempts = 6u32; // 6 × 10s = 60s
            loop {
                tokio::select! {
                    changed = card_num_retry_shutdown_rx.changed() => {
                        if changed.is_err() || *card_num_retry_shutdown_rx.borrow() {
                            tracing::debug!(
                                "v1.4.111: card_num retry loop received shutdown signal"
                            );
                            return;
                        }
                    }
                    _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {}
                }
                attempts += 1;
                let accounts = trd_cache.get_accounts();
                if accounts.is_empty() {
                    if attempts >= max_attempts {
                        tracing::warn!(
                            "v1.4.103 B10: trd_cache 仍空 (after {max_attempts} × 10s); \
                             受限 key 仍走 fail-closed sentinel reject 直到 SIGHUP / cache 加载."
                        );
                        return;
                    }
                    continue;
                }
                (card_num_expand_fn_loop)();
                return;
            }
        })
    };

    // codex F2 (P1) + F3.1 (P1) round 3: 单一 SIGHUP 钩子 — reload + expand
    // 顺序操作避免 race. 之前每 server (REST/gRPC) 各自 SIGHUP 监听 reload,
    // 加上 card_num expand 单独监听 — 多 SIGHUP listener 并发执行无序, 可能
    // expand 先跑 (写 sentinel) 然后 reload 后跑 (overwrite sentinel 用 raw
    // allowed_card_nums) → 受限 key 在 reload window 内 silent unrestricted.
    //
    // 现在: **唯一 SIGHUP listener**, 顺序 reload → expand. 删除 REST/gRPC 各
    // 自的 SIGHUP reload listener (上面 server 块内已注释).
    #[cfg(unix)]
    let sighup_handle: Option<BackgroundJoinHandle> = {
        let unified_sighup_fn = std::sync::Arc::clone(&card_num_reload_and_expand_fn);
        let mut sighup_shutdown_rx = shutdown_rx.clone();
        Some(tokio::spawn(async move {
            use tokio::signal::unix::{SignalKind, signal};
            let mut sig = match signal(SignalKind::hangup()) {
                Ok(s) => s,
                Err(e) => {
                    tracing::error!(error = %e, "SIGHUP install failed (unified reload+expand)");
                    return;
                }
            };
            tracing::info!(
                "v1.4.103 F3.1: unified SIGHUP handler installed (reload all keys + expand card_num)"
            );
            loop {
                tokio::select! {
                    signal = sig.recv() => {
                        if signal.is_none() {
                            return;
                        }
                        tracing::info!(
                            "v1.4.103 F3.1: SIGHUP received — running reload_all_stores + \
                             expand_allowed_card_nums (single ordered op, no race)"
                        );
                        (unified_sighup_fn)(true); // do_reload=true
                    }
                    changed = sighup_shutdown_rx.changed() => {
                        if changed.is_err() || *sighup_shutdown_rx.borrow() {
                            tracing::debug!(
                                "v1.4.111: unified SIGHUP handler received shutdown signal"
                            );
                            return;
                        }
                    }
                }
            }
        }))
    };
    #[cfg(not(unix))]
    let sighup_handle: Option<BackgroundJoinHandle> = None;

    // 11. v1.4.104 external reviewer S-001 (P0): native TCP keystore guard.
    //
    // 配任一 keys file (rest / grpc / ws) → 用户**意图**启用 scope mode.
    // 但 TCP FTAPI 协议 InitConnect 没有 Bearer 字段, 无法做 caller-specific
    // scope check (S-001). 默认 fail-closed: 不启 TCP listener, 用户需要
    // 通过 REST/gRPC/WS 与 daemon 交互 (这些 surface 都已加 pipeline auth).
    //
    // 显式 `--allow-tcp-unauthenticated` opt-in → 启 TCP, 但 daemon 启动
    // loud warn 该端口完全无 auth, agent skill 等 local process 可任意调用.
    // 用之前 captured 的 local 变量, args 已被 merge_config 消费
    let tcp_disabled = !super::phase1::ftapi_listener_enabled(config);

    if tcp_disabled {
        tracing::warn!(
            listen_addr = %listen_addr,
            "v1.4.104 external report S-001 (P0) fix: TCP listener (port {}) NOT started — \
             keys file configured but --allow-tcp-unauthenticated not set. \
             native TCP FTAPI protocol has no Bearer field, cannot enforce \
             caller-specific scope check; defaulting to fail-closed (skip TCP). \
             Use REST/gRPC/WS endpoints for authenticated access. \
             To restore TCP (legacy Python SDK clients) add --allow-tcp-unauthenticated, \
             but be aware that port {} will accept ANY local connection without \
             scope check (跨账户 leak risk).",
            config.port,
            config.port,
        );
        eprintln!(
            "⚠️  TCP listener (port {}) DISABLED (v1.4.104 external report S-001 fix): \
             keys file configured + no --allow-tcp-unauthenticated. \
             Pass --allow-tcp-unauthenticated to restore (with security warning).",
            config.port,
        );
    } else if any_keys_configured && allow_tcp_unauthenticated {
        tracing::warn!(
            listen_addr = %listen_addr,
            "⚠️  v1.4.104: TCP listener running WITHOUT scope check despite keys configured \
             (--allow-tcp-unauthenticated set). Port {} accepts ANY local connection — \
             跨账户 leak risk. Use REST/gRPC/WS for authenticated clients; reserve \
             TCP only for legacy Python SDK / C++ OpenD where Bearer not feasible.",
            config.port,
        );
        eprintln!(
            "⚠️  TCP port {} ACCEPTS UNAUTHENTICATED connections (--allow-tcp-unauthenticated). \
             受限 keys 不在该 surface 强制. 推荐改用 REST/gRPC/WS.",
            config.port,
        );
    }

    let tcp_shutdown_rx = shutdown_rx.clone();
    let tcp_handle: Option<SurfaceJoinHandle> = if tcp_disabled {
        None
    } else {
        let tcp_listener_events = listener_events_tx.clone();
        Some(tokio::spawn(async move {
            server
                .run_until_shutdown_with_listener_events(tcp_shutdown_rx, Some(tcp_listener_events))
                .await
        }))
    };

    let enabled_listeners = EnabledListenerSurfaces {
        ftapi: !tcp_disabled,
        websocket: config.websocket_port.is_some(),
        rest: config.rest_port.is_some(),
        grpc: config.grpc_port.is_some(),
        telnet: config.telnet_port.is_some(),
    };
    if let Some(plan) = auth_plan.as_mut() {
        let loopback = bind_includes_loopback(&config.ip);
        let available = external_verification_owner_available(VerificationOwnerFacts {
            ftapi_loopback_legacy: enabled_listeners.ftapi && loopback,
            rest_loopback_legacy: enabled_listeners.rest
                && loopback
                && rest_key_store
                    .as_ref()
                    .is_none_or(|store| !store.is_configured()),
            ws_loopback_legacy: enabled_listeners.websocket
                && loopback
                && ws_key_store
                    .as_ref()
                    .is_none_or(|store| !store.is_configured()),
            grpc_loopback_legacy: enabled_listeners.grpc
                && loopback
                && grpc_key_store
                    .as_ref()
                    .is_none_or(|store| !store.is_configured()),
            rest_auth_setup: enabled_listeners.rest
                && rest_key_store
                    .as_ref()
                    .is_some_and(|store| store.has_usable_scope(futu_auth::Scope::AuthSetup)),
            ws_auth_setup: enabled_listeners.websocket
                && ws_key_store
                    .as_ref()
                    .is_some_and(|store| store.has_usable_scope(futu_auth::Scope::AuthSetup)),
            grpc_auth_setup: enabled_listeners.grpc
                && grpc_key_store
                    .as_ref()
                    .is_some_and(|store| store.has_usable_scope(futu_auth::Scope::AuthSetup)),
        });
        plan.set_external_verification_available(available);
        tracing::info!(
            external_verification_owner_available = available,
            "computed pre-login verification ownership from bound listeners"
        );
    }
    let mut surface_tasks = SurfaceTasks {
        ftapi: tcp_handle,
        websocket: ws_handle.take(),
        rest: rest_handle.take(),
        grpc: grpc_handle.take(),
        telnet: telnet_handle.take(),
    };
    drop(listener_events_tx);
    let listener_readiness = await_listeners_opened(
        &mut listener_events_rx,
        enabled_listeners,
        &mut surface_tasks,
        &mut shutdown_rx,
    )
    .await;

    // 11. 等待任一 surface 退出、Ctrl+C 或 telnet shutdown。Bind 聚合失败
    // 也流入同一个 cleanup 尾部，避免已成功打开的 sibling surface 被 detach。
    let phase4_result: anyhow::Result<()> = match listener_readiness {
        Err(error) => Err(error),
        Ok(readiness) => {
            // Bind success is not enough for an external Verification owner:
            // the accept/serve loops must be active before authentication can
            // block awaiting a pre-login 1006 request. Every surface keeps its
            // shared readiness gate, so activation exposes only the explicit
            // pre-login allowlist until the identity transition reaches Ready.
            match activate_restricted_listeners_before_auth(
                readiness,
                enabled_listeners,
                &mut shutdown_rx,
            )
            .await
            {
                Err(error) => Err(error),
                Ok(listener_activation) => {
                    let auth_error = if let Some(auth_plan) = auth_plan {
                        match execute_auth_plan_until_shutdown(
                            Arc::clone(&bridge),
                            auth_plan,
                            &mut shutdown_rx,
                            &mut surface_tasks,
                        )
                        .await
                        {
                            Ok(push_rx) => {
                                bridge.start_prepared_push_dispatcher(
                                    prepared_push_dispatcher,
                                    push_rx,
                                );
                                tracing::info!(
                                    "authentication promoted atomically; push dispatcher started"
                                );
                                None
                            }
                            Err(error) => {
                                let startup = bridge.startup_readiness().snapshot();
                                if startup.state
                                    == futu_server::identity::StartupState::Authenticating
                                {
                                    let _ = bridge.startup_readiness().transition(
                                        startup.generation,
                                        futu_server::identity::StartupEvent::AuthFailed,
                                    );
                                }
                                crate::hints::print_auth_error_hint(&error, config);
                                Some(startup_auth_failure_after_restricted_bind(error))
                            }
                        }
                    } else {
                        None
                    };
                    if let Some(error) = auth_error {
                        Err(error)
                    } else {
                        match listener_activation
                            .activate_ready_only_and_confirm(&mut shutdown_rx, |marker| *marker)
                            .await
                        {
                            Err(error) => Err(error),
                            Ok(listeners_marker) => {
                                tracing::info!(marker = %listeners_marker, "authentication complete on active listeners");
                                log_startup_summary(
                                    config,
                                    &listen_addr,
                                    &bridge,
                                    rest_transport_kind,
                                );
                                tracing::info!(
                                    "gateway ready, accepting connections on {listen_addr}"
                                );
                                tracing::info!("press Ctrl+C to exit");
                                tokio::select! {
                                result = surface_task_result_or_pending(&mut surface_tasks.ftapi) => {
                                    surface_tasks.ftapi = None;
                                    log_surface_task_result("API server", result)
                                }
                                result = surface_task_result_or_pending(&mut surface_tasks.websocket) => {
                                    surface_tasks.websocket = None;
                                    log_surface_task_result("WebSocket", result)
                                }
                                result = surface_task_result_or_pending(&mut surface_tasks.rest) => {
                                    surface_tasks.rest = None;
                                    log_surface_task_result("REST API", result)
                                }
                                result = surface_task_result_or_pending(&mut surface_tasks.grpc) => {
                                    surface_tasks.grpc = None;
                                    log_surface_task_result("gRPC", result)
                                }
                                result = surface_task_result_or_pending(&mut surface_tasks.telnet) => {
                                    surface_tasks.telnet = None;
                                    log_surface_task_result("Telnet", result)
                                }
                                _ = tokio::signal::ctrl_c() => {
                                    tracing::info!("received Ctrl+C, shutting down gracefully...");
                                    Ok(())
                                }
                                _ = wait_for_sigterm() => {
                                    tracing::info!("received SIGTERM, shutting down gracefully...");
                                    Ok(())
                                }
                                _ = async {
                                    while shutdown_rx.changed().await.is_ok() {
                                        if *shutdown_rx.borrow() {
                                            break;
                                        }
                                    }
                                } => {
                                    tracing::info!("shutdown requested via telnet");
                                    Ok(())
                                }
                                        }
                            }
                        }
                    }
                }
            }
        }
    };

    // Publish the gateway-init failure/suspension outcome before stopping the
    // surface loops. Deferred InitConnect completions observe this generation
    // transition and enqueue their failure response while each connection's
    // bounded send queue is still owned by the active listener.
    transition_startup_to_shutdown(bridge.startup_readiness());
    if tokio::time::timeout(
        SHUTDOWN_GRACE_PERIOD,
        bridge
            .startup_readiness()
            .await_pending_init_connects_drained(),
    )
    .await
    .is_err()
    {
        tracing::warn!(
            pending_init_connects = bridge.startup_readiness().pending_init_connect_count(),
            "timed out flushing deferred InitConnect completions before surface shutdown"
        );
    }

    // 清理后台任务
    request_surface_shutdown(&shutdown_tx, "phase4 exit");
    if !bridge
        .qot_runtime()
        .indicator_calc_runtime()
        .shutdown_and_wait(SHUTDOWN_GRACE_PERIOD)
        .await
    {
        tracing::warn!("timed out cancelling indicator calculations during shutdown");
    }
    await_background_shutdown(
        "card_num retry",
        card_num_retry_handle,
        SHUTDOWN_GRACE_PERIOD,
    )
    .await;
    if let Some(handle) = sighup_handle {
        await_background_shutdown("SIGHUP reload", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = surface_tasks.ftapi {
        await_surface_shutdown("API server", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = surface_tasks.websocket {
        await_surface_shutdown("WebSocket", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = surface_tasks.rest {
        await_surface_shutdown("REST API", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = surface_tasks.grpc {
        await_surface_shutdown("gRPC", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = surface_tasks.telnet {
        await_surface_shutdown("Telnet", handle, SHUTDOWN_GRACE_PERIOD).await;
    }

    phase4_result?;
    tracing::info!("gateway stopped");
    Ok(())
}