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::{Context, Result};
use std::sync::Arc;

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

use crate::config::RuntimeConfig;
use crate::startup::phase1::Phase1Out;
use crate::startup::phase3::Phase3Out;

mod shutdown;
#[cfg(test)]
mod tests;

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,
};

fn merge_push_health_snapshots_for_rest(
    push: serde_json::Value,
    qot_login: serde_json::Value,
    backend_connected: bool,
) -> serde_json::Value {
    let mut combined = match push {
        serde_json::Value::Object(map) => map,
        other => {
            let mut map = serde_json::Map::new();
            map.insert(
                "error".to_string(),
                serde_json::Value::String(format!(
                    "push_health snapshot serialized to non-object {}",
                    json_value_kind(&other)
                )),
            );
            map
        }
    };
    combined.insert("qot_login_health".to_string(), qot_login);
    combined.insert(
        "backend_connected".to_string(),
        serde_json::Value::Bool(backend_connected),
    );
    serde_json::Value::Object(combined)
}

fn json_value_kind(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "bool",
        serde_json::Value::Number(_) => "number",
        serde_json::Value::String(_) => "string",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct LegacyNetworkExposureWarning {
    surface: &'static str,
    key_flag: &'static str,
}

fn is_loopback_bind(ip: &str) -> bool {
    matches!(ip, "127.0.0.1" | "localhost" | "::1") || ip.starts_with("127.")
}

fn legacy_network_exposure_warnings(
    ip: &str,
    any_keys_configured: bool,
    allow_tcp_unauthenticated: bool,
    rest_key_configured: Option<bool>,
    grpc_key_configured: Option<bool>,
    ws_key_configured: Option<bool>,
) -> Vec<LegacyNetworkExposureWarning> {
    if is_loopback_bind(ip) {
        return Vec::new();
    }

    let mut warnings = Vec::new();
    if !any_keys_configured || allow_tcp_unauthenticated {
        warnings.push(LegacyNetworkExposureWarning {
            surface: "FTAPI TCP",
            key_flag: "--rest-keys-file/--grpc-keys-file/--ws-keys-file",
        });
    }
    for (enabled_keyed, surface, key_flag) in [
        (rest_key_configured, "REST", "--rest-keys-file"),
        (grpc_key_configured, "gRPC", "--grpc-keys-file"),
        (ws_key_configured, "WS", "--ws-keys-file"),
    ] {
        if enabled_keyed == Some(false) {
            warnings.push(LegacyNetworkExposureWarning { surface, key_flag });
        }
    }
    warnings
}

fn warn_legacy_network_exposure(config: &RuntimeConfig, any_keys_configured: bool) {
    for warning in legacy_network_exposure_warnings(
        &config.ip,
        any_keys_configured,
        config.allow_tcp_unauthenticated,
        config.rest_port.map(|_| config.rest_keys_file.is_some()),
        config.grpc_port.map(|_| config.grpc_keys_file.is_some()),
        config.websocket_port.map(|_| config.ws_keys_file.is_some()),
    ) {
        tracing::warn!(
            bind_ip = %config.ip,
            surface = warning.surface,
            key_flag = warning.key_flag,
            "legacy network exposure: surface is reachable on a non-loopback bind without API-key auth"
        );
        eprintln!(
            "⚠️  {} legacy mode is reachable on {} without API-key auth. \
             For production, bind 127.0.0.1 or configure {}.",
            warning.surface, config.ip, warning.key_flag
        );
    }
}

pub(super) async fn run_phase4(
    config: &RuntimeConfig,
    phase1: Phase1Out,
    bridge: Arc<GatewayBridge>,
    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,
        ws_keys_file,
        grpc_keys_file,
        allow_tcp_unauthenticated,
    } = phase1;
    let Phase3Out {
        server,
        server_config,
        ws_broadcaster,
        grpc_broadcaster,
    } = phase3;

    // 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 = match &ws_keys_file {
            Some(path) => match futu_auth::KeyStore::load(path) {
                Ok(ks) => {
                    tracing::info!(
                        path = %path.display(),
                        keys_loaded = ks.len(),
                        "WS keys file loaded (Bearer/?token auth enabled)"
                    );
                    Some(std::sync::Arc::new(ks))
                }
                Err(e) => {
                    // v1.4.102 BUG-007: fail-closed
                    tracing::error!(
                        error = %e,
                        path = %path.display(),
                        "failed to load WS keys file (--ws-keys-file 明确指定 → fail-closed). \
                         daemon abort. fix the keys file then restart. \
                         不再 fallback to legacy unauth (v1.4.102 BUG-007 fix)."
                    );
                    return Err(anyhow::anyhow!(
                        "failed to load WS keys file {}: {e}. \
                         --ws-keys-file 明确指定 → fail-closed (BUG-007).",
                        path.display()
                    ));
                }
            },
            None => None,
        };
        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_offset_secs(bridge.server_clock().offset_secs());
        tracing::info!(addr = %ws_addr, "starting WebSocket server");
        let ws_shutdown_rx = shutdown_rx.clone();
        Some(tokio::spawn(async move {
            ws_server
                .run_until_shutdown(ws_shutdown_rx)
                .await
                .context("WebSocket server error")
        }))
    } 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 = match &rest_keys_file {
            Some(path) => match futu_auth::KeyStore::load(path) {
                Ok(ks) => {
                    tracing::info!(
                        path = %path.display(),
                        keys_loaded = ks.len(),
                        "REST keys file loaded (Bearer auth enabled)"
                    );
                    std::sync::Arc::new(ks)
                }
                Err(e) => {
                    tracing::error!(
                        error = %e,
                        path = %path.display(),
                        "failed to load REST keys file (--rest-keys-file 明确指定 → fail-closed). \
                         daemon abort. fix the keys file then restart. \
                         不再 fallback to legacy unauth (v1.4.102 BUG-007 fix)."
                    );
                    return Err(anyhow::anyhow!(
                        "failed to load REST keys file {}: {e}. \
                         --rest-keys-file 明确指定 → fail-closed (BUG-007).",
                        path.display()
                    ));
                }
            },
            None => std::sync::Arc::new(futu_auth::KeyStore::empty()),
        };
        tracing::info!(addr = %rest_addr, "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();
                merge_push_health_snapshots_for_rest(push, qot_login, backend_connected)
            });
        // 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();
        Some(tokio::spawn(async move {
            futu_rest::server::start_with_auth_full_admin_until_shutdown(
                &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_shutdown_rx,
            )
            .await
            .context("REST API server error")
        }))
    } 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 = match &grpc_keys_file {
            Some(path) => match futu_auth::KeyStore::load(path) {
                Ok(ks) => {
                    tracing::info!(
                        path = %path.display(),
                        keys_loaded = ks.len(),
                        "gRPC keys file loaded (Bearer auth enabled)"
                    );
                    std::sync::Arc::new(ks)
                }
                Err(e) => {
                    tracing::error!(
                        error = %e,
                        path = %path.display(),
                        "failed to load gRPC keys file (--grpc-keys-file 明确指定 → fail-closed). \
                         daemon abort. fix the keys file then restart. \
                         不再 fallback to legacy unauth (v1.4.102 BUG-007 fix)."
                    );
                    return Err(anyhow::anyhow!(
                        "failed to load gRPC keys file {}: {e}. \
                         --grpc-keys-file 明确指定 → fail-closed (BUG-007).",
                        path.display()
                    ));
                }
            },
            None => 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();
        Some(tokio::spawn(async move {
            futu_grpc::server::start_with_auth_until_shutdown(
                &grpc_addr,
                router,
                broadcaster,
                grpc_key_store,
                grpc_counters,
                grpc_shutdown_rx,
            )
            .await
            .map_err(|e| anyhow::anyhow!("gRPC server error: {e}"))
        }))
    } else {
        None
    };

    // 10. 启动 Telnet 管理服务（可选）
    let mut telnet_handle = if let Some(telnet_port) = config.telnet_port {
        let telnet_addr = format!("{}:{}", config.ip, telnet_port);
        // 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 AuthRefresher relogin"
            );
            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();
        Some(tokio::spawn(async move {
            telnet_server
                .run_until_shutdown(telnet_shutdown_rx)
                .await
                .context("Telnet server error")
        }))
    } else {
        None
    };

    tracing::info!("gateway ready, accepting connections on {listen_addr}");
    tracing::info!("press Ctrl+C to exit");

    // 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 = any_keys_configured && !allow_tcp_unauthenticated;

    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 mut tcp_handle: Option<SurfaceJoinHandle> = if tcp_disabled {
        None
    } else {
        Some(tokio::spawn(async move {
            server
                .run_until_shutdown(tcp_shutdown_rx)
                .await
                .context("API server error")
        }))
    };

    // 11. 等待任一 surface 退出、Ctrl+C 或 telnet shutdown
    let phase4_result: anyhow::Result<()> = tokio::select! {
        result = surface_task_result_or_pending(&mut tcp_handle) => {
            tcp_handle = None;
            log_surface_task_result("API server", result)
        }
        result = surface_task_result_or_pending(&mut ws_handle) => {
            ws_handle = None;
            log_surface_task_result("WebSocket", result)
        }
        result = surface_task_result_or_pending(&mut rest_handle) => {
            rest_handle = None;
            log_surface_task_result("REST API", result)
        }
        result = surface_task_result_or_pending(&mut grpc_handle) => {
            grpc_handle = None;
            log_surface_task_result("gRPC", result)
        }
        result = surface_task_result_or_pending(&mut telnet_handle) => {
            telnet_handle = None;
            log_surface_task_result("Telnet", result)
        }
        _ = tokio::signal::ctrl_c() => {
            tracing::info!("received Ctrl+C, shutting down gracefully...");
            Ok(())
        }
        _ = async {
            while shutdown_rx.changed().await.is_ok() {
                if *shutdown_rx.borrow() {
                    break;
                }
            }
        } => {
            tracing::info!("shutdown requested via telnet");
            Ok(())
        }
    };

    // 清理后台任务
    request_surface_shutdown(&shutdown_tx, "phase4 exit");
    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) = tcp_handle {
        await_surface_shutdown("API server", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = ws_handle {
        await_surface_shutdown("WebSocket", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = rest_handle {
        await_surface_shutdown("REST API", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = grpc_handle {
        await_surface_shutdown("gRPC", handle, SHUTDOWN_GRACE_PERIOD).await;
    }
    if let Some(handle) = telnet_handle {
        await_surface_shutdown("Telnet", handle, SHUTDOWN_GRACE_PERIOD).await;
    }

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