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.

#![allow(unused_imports)]

use anyhow::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;

#[cfg(test)]
mod tests;

fn merge_push_health_snapshots_for_rest(
    push: serde_json::Value,
    qot_login: serde_json::Value,
) -> 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);
    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",
    }
}

pub(super) async fn run_phase4(
    config: &RuntimeConfig,
    phase1: Phase1Out,
    bridge: Arc<GatewayBridge>,
    phase3: Phase3Out,
) -> 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;

    // 7. 启动 WebSocket 服务（可选）
    let 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(std::sync::Arc::clone(&bridge.subscriptions)),
            ),
            ws_key_store,
            Some(ws_counters),
        );
        tracing::info!(addr = %ws_addr, "starting WebSocket server");
        Some(tokio::spawn(async move {
            if let Err(e) = ws_server.run().await {
                tracing::error!(error = %e, "WebSocket server error");
            }
        }))
    } else {
        None
    };

    // 8. 启动 REST API 服务（可选，含 WebSocket 推送）
    let 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"}))
            });
        // 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"}),
                );
                merge_push_health_snapshots_for_rest(push, qot_login)
            });
        // 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)
            });
        Some(tokio::spawn(async move {
            if let Err(e) = futu_rest::server::start_with_auth_full_admin(
                &rest_addr,
                router,
                broadcaster,
                rest_key_store,
                rest_counters,
                futu_rest::server::RestAdminHooks {
                    admin_status_provider: Some(admin_status_provider),
                    admin_reload_handler: Some(admin_reload_handler),
                    push_health_snapshot_provider: Some(push_health_snapshot_provider),
                    card_num_resolver: Some(card_num_resolver),
                },
            )
            .await
            {
                tracing::error!(error = %e, "REST API server error");
            }
        }))
    } else {
        None
    };

    // 9. 启动 gRPC 服务（可选，含流式推送）
    let 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);
        Some(tokio::spawn(async move {
            if let Err(e) = futu_grpc::server::start_with_auth(
                &grpc_addr,
                router,
                broadcaster,
                grpc_key_store,
                grpc_counters,
            )
            .await
            {
                tracing::error!(error = %e, "gRPC server error");
            }
        }))
    } else {
        None
    };

    // 10. 启动 Telnet 管理服务（可选）
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
    let 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(std::sync::Arc::clone(&bridge.subscriptions)),
            Some(std::sync::Arc::clone(server.metrics())),
            shutdown_tx,
        )
        .with_relogin_fn(relogin_fn);
        tracing::info!(addr = %telnet_addr, "starting Telnet server");
        Some(tokio::spawn(async move {
            if let Err(e) = telnet_server.run().await {
                tracing::error!(error = %e, "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_expand_fn_loop = std::sync::Arc::clone(&card_num_expand_fn);
        let bridge_for_check = std::sync::Arc::clone(&bridge);
        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::time::sleep(std::time::Duration::from_secs(10)).await;
                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 unified_sighup_fn = std::sync::Arc::clone(&card_num_reload_and_expand_fn);
        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)"
            );
            while sig.recv().await.is_some() {
                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
            }
        });
    }

    // 11. v1.4.104 eli 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 any_keys_configured =
        rest_keys_file.is_some() || grpc_keys_file.is_some() || ws_keys_file.is_some();
    let tcp_disabled = any_keys_configured && !allow_tcp_unauthenticated;

    if tcp_disabled {
        tracing::warn!(
            listen_addr = %listen_addr,
            "v1.4.104 eli 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 eli 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,
        );
    }

    // 11. 启动 API 服务 + 信号处理
    tokio::select! {
        result = async {
            if tcp_disabled {
                // TCP 不跑, 但仍要等 ctrl_c / telnet shutdown
                std::future::pending::<anyhow::Result<()>>().await
            } else {
                server.run().await
            }
        } => {
            if let Err(e) = result {
                tracing::error!(error = %e, "API server error");
            }
        }
        _ = tokio::signal::ctrl_c() => {
            tracing::info!("received Ctrl+C, shutting down gracefully...");
        }
        _ = async {
            while shutdown_rx.changed().await.is_ok() {
                if *shutdown_rx.borrow() {
                    break;
                }
            }
        } => {
            tracing::info!("shutdown requested via telnet");
        }
    }

    // 清理后台任务
    if let Some(handle) = ws_handle {
        handle.abort();
    }
    if let Some(handle) = rest_handle {
        handle.abort();
    }
    if let Some(handle) = grpc_handle {
        handle.abort();
    }
    if let Some(handle) = telnet_handle {
        handle.abort();
    }

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