futu_mcp/

state.rs

//! 共享状态：网关连接 + 订阅状态 + 授权

use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use anyhow::{Context, Result, anyhow};
use futu_auth::{KeyRecord, KeyStore, RuntimeCounters};
use futu_core::qot_symbol;
use futu_net::client::{ClientConfig, FutuClient, ReconnectingClient};
use futu_net::reconnect::ReconnectPolicy;
use futu_qot::types::Security;
use rmcp::{RoleServer, service::Peer};
use tokio::sync::{Mutex, Semaphore};

mod push_filter;
mod push_subscribers;
pub(crate) use push_subscribers::{LegacyPushServiceLease, PushDeliveryTarget};
#[cfg(test)]
pub(crate) use push_subscribers::{MAX_MODERN_PUSH_HANDLES, MODERN_PUSH_QUEUE_CAPACITY};
pub(crate) use push_subscribers::{parse_push_resource_uri, push_resource_uri};
#[cfg(test)]
mod tests;

use push_filter::{TradePushDecode, classify_trade_push, trd_market_int_to_str};
#[cfg(test)]
use push_filter::{
    extract_acc_id_and_market_from_push, is_trade_push_proto_id, subscriber_should_receive,
    subscriber_should_receive_with_market, subscriber_visible_to_caller,
};
use push_subscribers::{PushSubscriber, SubscriberDelivery};

use crate::qot_sdk_adapter;

const MCP_CONNECT_TOTAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
const MCP_CONNECT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(200);
pub(crate) const LEGACY_PUSH_INFO_LEVEL_RANK: u8 = 1;

/// Debug-only view for account collections at the logging boundary.
///
/// Account identifiers are financial PII. Expose only cardinality so adding a
/// tracing field cannot accidentally serialize the underlying slice.
struct RedactedAccountIdSet<'a> {
    account_ids: &'a [u64],
}

impl<'a> RedactedAccountIdSet<'a> {
    const fn new(account_ids: &'a [u64]) -> Self {
        Self { account_ids }
    }
}

impl fmt::Debug for RedactedAccountIdSet<'_> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "accounts(count={})", self.account_ids.len())
    }
}

/// Fail-closed logging view for errors whose backend text may echo an account
/// identifier or other request material.
struct RedactedMcpError<'a> {
    _error: &'a anyhow::Error,
}

impl<'a> RedactedMcpError<'a> {
    const fn new(error: &'a anyhow::Error) -> Self {
        Self { _error: error }
    }
}

impl fmt::Debug for RedactedMcpError<'_> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("request_failed")
    }
}

/// v1.4.38 Phase 5 helper: bytes → base64 (用于 push body 安全包进 JSON)
fn base64_encode_bytes(bytes: &[u8]) -> String {
    use base64::Engine as _;
    base64::engine::general_purpose::STANDARD.encode(bytes)
}

struct PendingPushDelivery {
    target: SubscriberDelivery,
    data: serde_json::Value,
    session_id: String,
    owner_key_id: Option<String>,
    proto_id: u32,
}

/// Single authorization admission shared by legacy logging and modern resource
/// delivery. A caller must pass this function before either target is created,
/// so decode-failed restricted trade bodies and current scope revocations can
/// never enter a modern queue.
fn push_delivery_is_authorized(
    subscriber: &PushSubscriber,
    key_store: &KeyStore,
    filter_registry: &futu_auth_pipeline::FilterRegistry,
    decode_result: &TradePushDecode,
    event_type: &'static str,
    push_acc_id: Option<u64>,
    push_trd_market_str: Option<&'static str>,
    proto_id: u32,
) -> bool {
    if event_type == "private_user"
        && !private_user_push_owner_is_authorized(subscriber.owner_key_id.as_deref(), key_store)
    {
        return false;
    }
    let (allowed_acc_ids, allowed_markets) = match &subscriber.delivery {
        SubscriberDelivery::LegacyPeer(_, _) => (
            subscriber.allowed_acc_ids_snapshot.clone(),
            subscriber.allowed_markets_snapshot.clone(),
        ),
        SubscriberDelivery::ModernResource(_) => {
            let Some(owner) = subscriber.owner_key_id.as_deref() else {
                return false;
            };
            let Some(current) = key_store
                .get_by_id_for_current_machine(owner)
                .filter(|record| {
                    !record.is_expired(chrono::Utc::now())
                        && record.scopes.contains(&futu_auth::Scope::AccRead)
                })
            else {
                return false;
            };
            (
                current.allowed_acc_ids.clone(),
                current.allowed_markets.clone(),
            )
        }
    };
    if matches!(decode_result, TradePushDecode::DecodeFailed) {
        let restricted = allowed_acc_ids
            .as_ref()
            .is_some_and(|allowed| !allowed.is_empty());
        if restricted {
            let key_id = subscriber.owner_key_id.as_deref().unwrap_or("<none>");
            futu_auth::metrics::bump_ws_filtered("trade_decode_failed", key_id);
            tracing::warn!(
                proto_id,
                key_id,
                "MCP trade push body decode failed; dropped before legacy/modern delivery for restricted key"
            );
            return false;
        }
    }
    let sub_state = (!subscriber.acc_ids.is_empty()).then_some(&subscriber.acc_ids);
    let ctx = futu_auth_pipeline::PushEventCtx {
        event_type,
        event_acc: push_acc_id,
        allowed_acc_ids: allowed_acc_ids.as_ref(),
        sub_state,
        event_trd_market: push_trd_market_str,
        allowed_markets: allowed_markets.as_ref(),
    };
    if filter_registry.should_drop_event(&ctx) {
        let key_id = subscriber.owner_key_id.as_deref().unwrap_or("<none>");
        futu_auth::metrics::bump_ws_filtered("trade_market", key_id);
        return false;
    }
    true
}

fn private_user_push_owner_is_authorized(owner_key_id: Option<&str>, key_store: &KeyStore) -> bool {
    let Some(owner) = owner_key_id else {
        return false;
    };
    key_store
        .get_by_id_for_current_machine(owner)
        .is_some_and(|record| {
            !record.is_expired(chrono::Utc::now())
                && record.scopes.contains(&futu_auth::Scope::AccRead)
        })
}

#[allow(deprecated)]
async fn notify_legacy_push(
    peer: &Peer<RoleServer>,
    minimum_level: &std::sync::atomic::AtomicU8,
    data: serde_json::Value,
) -> Result<(), rmcp::service::ServiceError> {
    if minimum_level.load(Ordering::Acquire) > LEGACY_PUSH_INFO_LEVEL_RANK {
        return Ok(());
    }
    let params =
        rmcp::model::LoggingMessageNotificationParam::new(rmcp::model::LoggingLevel::Info, data)
            .with_logger("futu_push");
    peer.notify_logging_message(params).await
}

/// MCP server 运行时状态
#[derive(Clone)]
pub struct ServerState {
    /// [`Inner`] 共享可变状态（gateway 地址 + 懒加载的 [`FutuClient`]）
    inner: Arc<Mutex<Inner>>,
    /// 是否启用交易写工具（place/modify/cancel）。默认 false。旧开关，仅当
    /// `key_store.is_configured() == false` 时生效。
    enable_trading: bool,
    /// 是否允许对 real 环境下单。默认 false。旧开关，同上。
    allow_real_trading: bool,
    /// keys.json 加载的 KeyStore。`is_configured()` 为 true 时走 scope 授权模式。
    key_store: Arc<KeyStore>,
    /// 调用方传入的 API Key 对应的记录；None 表示未提供 key。
    authed_key: Option<Arc<KeyRecord>>,
    opend_rest_url: Option<String>,
    opend_api_key: Option<String>,
    /// 交易密码所属登录账号。用于 `futu_unlock_trade` 从账号级 keychain
    /// `trade-password.<login-account>` 读取密码；None 时走 legacy/global/env 兼容路径。
    trade_pwd_account: Option<String>,
    /// 限额运行时（日累计计数器）
    counters: Arc<RuntimeCounters>,
    /// v1.4.38 Phase 5: MCP push 订阅者注册表（session_uuid → subscriber）。
    /// `futu_sub_acc_push` 工具在 HTTP 模式下调用时注册当前 session，daemon
    /// push 到 MCP 后按 acc_id filter 向注册的 peer 发
    /// `notify_logging_message`（server-initiated notification）。
    push_subscribers: Arc<Mutex<HashMap<String, PushSubscriber>>>,
    /// Modern resources retain bounded queues for up to 4h. A semaphore makes
    /// handle cardinality admission atomic and lets the tool fail before any
    /// daemon side effect. The fixed local limit is a memory-safety boundary,
    /// not backend/config data; raise it only after measuring worst-case push
    /// payload memory and concurrent client demand.
    modern_push_slots: Arc<Semaphore>,
    /// issue #54：gateway 重连单飞门。建连期间持有（不持 `inner` 锁），并发
    /// caller 在此排队，拿到门后复检缓存直接复用赢家结果 → 一次断连只开一条新 TCP。
    connect_gate: Arc<Mutex<()>>,
    /// 陈旧订阅者 purge task 已 spawn 次数；`compare_exchange(0, 1)` 保证整个
    /// [`ServerState`] 生命周期只 spawn 一次（重连不重复 spawn）。
    purge_task_spawns: Arc<std::sync::atomic::AtomicU64>,
    /// issue #54 Reviewer M-5: gateway 重连后 `TRD_SUB_ACC_PUSH` 重注册失败次数。
    /// 失败只记日志 + 计数，不阻塞工具调用；调用方需重新 `futu_sub_acc_push`。
    push_reregister_failures: Arc<std::sync::atomic::AtomicU64>,
    /// issue #54 Reviewer I-2: set when the post-reconnect push re-registration
    /// failed; the next `client()` that finds a live connection retries once.
    push_registration_dirty: Arc<std::sync::atomic::AtomicBool>,
}

/// ServerState 内部可变部分，加锁存放 gateway 地址 + 懒加载的 [`FutuClient`]。
struct Inner {
    /// 网关 TCP 地址（如 `127.0.0.1:11111`）
    gateway: String,
    /// 懒加载的底层连接；首次调用 [`ServerState::client`] 时建立，后续复用
    client: Option<Arc<FutuClient>>,
    /// 连接代数：每次成功安装新 client +1（首连 = 1）。
    generation: u64,
}

impl ServerState {
    /// 创建默认 state：`enable_trading=false` / `allow_real_trading=false` /
    /// 空 [`KeyStore`] / 无 authed_key。使用 `with_*` 链式方法注入额外能力。
    pub fn new(gateway: String) -> Self {
        Self {
            inner: Arc::new(Mutex::new(Inner {
                gateway,
                client: None,
                generation: 0,
            })),
            enable_trading: false,
            allow_real_trading: false,
            key_store: Arc::new(KeyStore::empty()),
            authed_key: None,
            opend_rest_url: None,
            opend_api_key: None,
            trade_pwd_account: None,
            counters: Arc::new(RuntimeCounters::new()),
            push_subscribers: Arc::new(Mutex::new(HashMap::new())),
            modern_push_slots: Arc::new(Semaphore::new(push_subscribers::MAX_MODERN_PUSH_HANDLES)),
            connect_gate: Arc::new(Mutex::new(())),
            purge_task_spawns: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            push_reregister_failures: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            push_registration_dirty: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        }
    }

    /// 当前 gateway client 代数（0 = 尚未连接）。
    #[cfg(test)]
    pub(crate) async fn client_generation_for_test(&self) -> u64 {
        self.inner.lock().await.generation
    }

    /// 陈旧订阅者 purge task 已 spawn 次数。
    #[cfg(test)]
    pub(crate) fn purge_task_spawn_count_for_test(&self) -> u64 {
        self.purge_task_spawns.load(Ordering::Acquire)
    }

    /// 启用交易写工具（构造器式链式设置）
    pub fn with_trading(mut self, enable_trading: bool, allow_real_trading: bool) -> Self {
        self.enable_trading = enable_trading;
        self.allow_real_trading = allow_real_trading;
        self
    }

    /// 设置 KeyStore（新授权模式）
    pub fn with_key_store(mut self, store: Arc<KeyStore>) -> Self {
        self.key_store = store;
        self
    }

    /// 设置已通过验证的 API Key 记录
    pub fn with_authed_key(mut self, key: Option<Arc<KeyRecord>>) -> Self {
        self.authed_key = key;
        self
    }

    pub fn with_opend_rest(mut self, base_url: Option<String>, api_key: Option<String>) -> Self {
        self.opend_rest_url = base_url.map(|value| value.trim_end_matches('/').to_string());
        self.opend_api_key = api_key.filter(|value| !value.is_empty());
        self
    }

    pub fn opend_rest_url(&self) -> Option<&str> {
        self.opend_rest_url.as_deref()
    }

    pub fn opend_api_key(&self) -> Option<&str> {
        self.opend_api_key.as_deref()
    }

    /// 设置交易密码所属登录账号（MCP 只连 gateway，本身无法可靠推断 daemon
    /// 的 login account；由 CLI/env/config 显式注入）。
    pub fn with_trade_pwd_account(mut self, account: Option<String>) -> Self {
        self.trade_pwd_account = account;
        self
    }

    /// 是否启用了 scope 授权模式
    pub fn is_scope_mode(&self) -> bool {
        self.key_store.is_configured()
    }

    /// 交易写工具开关（legacy mode）。
    pub fn enable_trading(&self) -> bool {
        self.enable_trading
    }

    /// real 环境交易写工具开关（legacy mode）。
    pub fn allow_real_trading(&self) -> bool {
        self.allow_real_trading
    }

    /// 当前 MCP API key store。返回共享引用，避免调用方替换 runtime storage。
    pub fn key_store(&self) -> &Arc<KeyStore> {
        &self.key_store
    }

    /// startup 阶段验证过的 key 快照；调用方需要 fresh record 时仍应按 id 回查 key store。
    pub fn authed_key(&self) -> Option<Arc<KeyRecord>> {
        self.authed_key.clone()
    }

    /// 交易密码所属登录账号。
    pub fn trade_pwd_account(&self) -> Option<&str> {
        self.trade_pwd_account.as_deref()
    }

    /// 限额运行时计数器。返回共享引用，避免调用方替换 runtime storage。
    pub fn counters(&self) -> &Arc<RuntimeCounters> {
        &self.counters
    }

    /// 当前配置的 gateway 地址。
    pub async fn gateway(&self) -> String {
        self.inner.lock().await.gateway.clone()
    }

    /// 获取（或懒加载）网关客户端。
    ///
    /// issue #54：gateway 重启后旧 client 的 event loop 已退出
    /// （[`FutuClient::is_closed`]），缓存视为失效并重建；并发调用经 `connect_gate`
    /// 单飞，只开一条新连接；重连成功后重发本地 push 订阅
    /// （[`Self::restore_push_registration`]）。建连期间不持 `inner` 锁。
    pub async fn client(&self) -> Result<Arc<FutuClient>> {
        let (gateway, stale) = {
            let guard = self.inner.lock().await;
            match &guard.client {
                Some(c) if !c.is_closed() => {
                    let alive = c.clone();
                    drop(guard);
                    // Reviewer I-2: a failed post-reconnect push re-registration must
                    // not stay silently broken; retry once on the next live call.
                    if self
                        .push_registration_dirty
                        .swap(false, std::sync::atomic::Ordering::AcqRel)
                    {
                        self.restore_push_registration(&alive).await;
                    }
                    return Ok(alive);
                }
                cached => (guard.gateway.clone(), cached.clone()),
            }
        };

        // 单飞：先到者建连，后到者在门上等待；拿到门后复检，赢家已装好则直接复用。
        // Reviewer M-5：门在建连（最长 MCP_CONNECT_TOTAL_TIMEOUT=3s）期间持有，gateway
        // 宕机时 N 个并发调用会串行各等一次超时；这是单飞的代价，换来只开一条 TCP。
        // push 重注册（最长一次请求超时）已移到门外，见下方 `drop(_connect_gate)`。
        let _connect_gate = self.connect_gate.lock().await;
        {
            let guard = self.inner.lock().await;
            if let Some(c) = &guard.client
                && !c.is_closed()
            {
                return Ok(c.clone());
            }
        }

        let config = ClientConfig {
            addr: gateway.clone(),
            client_ver: env!("CARGO_PKG_VERSION").to_string(),
            client_id: "futu-mcp".to_string(),
            recv_notify: false,
            rsa_key: None,
        };
        let policy =
            ReconnectPolicy::new(MCP_CONNECT_RETRY_DELAY, MCP_CONNECT_RETRY_DELAY, Some(1));
        let mut reconnector = ReconnectingClient::new(config).with_policy(policy);
        let connect_result =
            tokio::time::timeout(MCP_CONNECT_TOTAL_TIMEOUT, reconnector.connect()).await;
        let (client, mut push_rx, info) = match connect_result {
            Ok(result) => {
                result.with_context(|| format!("connect to futu gateway at {gateway}"))?
            }
            Err(_) => {
                return Err(anyhow!(
                    "connect to futu gateway at {gateway} timed out after {}s",
                    MCP_CONNECT_TOTAL_TIMEOUT.as_secs()
                ));
            }
        };

        let arc = Arc::new(client);
        let (generation, previous_conn_id) = {
            let mut guard = self.inner.lock().await;
            // CAS 语义：只替换"空 / 已 closed / 与本次观察到的同一个 stale Arc"的缓存。
            // 若缓存里已是另一个存活 client（理论上被 gate 串行化排除），沿用它，
            // 本次新建的 client 随 `arc` drop 自行关闭 event loop。
            if let Some(existing) = &guard.client
                && !existing.is_closed()
                && !stale.as_ref().is_some_and(|s| Arc::ptr_eq(s, existing))
            {
                return Ok(existing.clone());
            }
            let previous_conn_id = guard.client.as_ref().and_then(|c| c.conn_id());
            guard.client = Some(arc.clone());
            guard.generation += 1;
            (guard.generation, previous_conn_id)
        };
        if generation > 1 {
            tracing::info!(
                generation,
                old_conn_id = previous_conn_id.unwrap_or(0),
                new_conn_id = info.conn_id,
                "mcp gateway client reconnected"
            );
        }

        // v1.4.38 Phase 5 (100%): 按 acc_ids 过滤的 push broadcast
        //
        // 流程：
        // 1. push_rx 收 daemon 转发的 push
        // 2. 对 TRD_UPDATE_ORDER (2208) / TRD_UPDATE_ORDER_FILL (2218) 解包
        //    提取 acc_id
        // 3. 遍历订阅者，**只推给 acc_ids 匹配的**（或订阅者 acc_ids 空 = 不
        //    过滤，所有 acc 都收）
        // 4. 行情 push（QOT_UPDATE_*）无 acc_id 语义，广播给所有订阅者
        //
        // Per-session 独立 spawn notify，避免一个慢 session 阻塞其他
        let subs_for_push = Arc::downgrade(&self.push_subscribers);
        let key_store_for_push = Arc::downgrade(&self.key_store);
        // v1.4.105 F5 fix (codex review C4 USER_ACK B): MCP push filter 改用
        // FilterRegistry::should_drop_event 单一注册中心 (跟 4 surface 一致),
        // 替代之前 inline subscriber_should_receive_with_market. 防 sibling-route
        // bypass — 加新 push event filter 维度只在 install_defaults 注册一次,
        // MCP 自动覆盖.
        let filter_registry =
            std::sync::Arc::new(futu_auth_pipeline::FilterRegistry::with_defaults());
        tokio::spawn(async move {
            while let Some(push) = push_rx.recv().await {
                let Some(subs_for_push) = subs_for_push.upgrade() else {
                    break;
                };
                let Some(key_store_for_push) = key_store_for_push.upgrade() else {
                    break;
                };
                let subscribers = {
                    let subs = subs_for_push.lock().await;
                    if subs.is_empty() {
                        Vec::new()
                    } else {
                        subs.iter()
                            .map(|(session_id, sub)| (session_id.clone(), sub.clone()))
                            .collect::<Vec<_>>()
                    }
                };
                if subscribers.is_empty() {
                    continue; // fast path: no listeners, drop
                }
                // v1.4.105 T-C2 + v1.4.106 codex 0932 F6/F7: classify push by proto_id
                // (set membership), 不再靠 body decode 成功推断. trade body decode
                // 失败现在归 TradePushDecode::DecodeFailed (event_type="trade",
                // 无 acc/market gate 信息) — restricted key 应 drop, unrestricted
                // 透传带 decode_status="failed".
                let decode_result = classify_trade_push(push.proto_id, &push.body);
                let (push_acc_id, push_trd_market, decode_status, event_type) = match &decode_result
                {
                    TradePushDecode::NotTrade
                        if push.proto_id == futu_core::proto_id::QOT_UPDATE_STOCK_NOTE =>
                    {
                        (None, None, "ok", "private_user")
                    }
                    TradePushDecode::NotTrade => (None, None, "ok", "quote"),
                    TradePushDecode::Decoded { acc_id, trd_market } => {
                        (Some(*acc_id), Some(*trd_market), "ok", "trade")
                    }
                    TradePushDecode::DecodeFailed => (None, None, "failed", "trade"),
                };
                let push_trd_market_str = push_trd_market.map(trd_market_int_to_str);
                // v1.4.106 codex 0932 F7 [P3]: payload 加 event_type / trd_market /
                // decode_status — 让客户端不需要按 proto_id 自己 derive (4 surface 一致).
                // body_base64 后向兼容保留.
                let payload = serde_json::json!({
                    "kind": "futu_push",
                    "proto_id": push.proto_id,
                    "acc_id": push_acc_id,
                    "event_type": event_type,
                    "trd_market": push_trd_market_str,
                    "decode_status": decode_status,
                    "body_base64": base64_encode_bytes(&push.body),
                });
                let deliveries = {
                    let mut deliveries = Vec::with_capacity(subscribers.len());
                    for (session_id, sub) in subscribers.iter() {
                        if !push_delivery_is_authorized(
                            sub,
                            &key_store_for_push,
                            &filter_registry,
                            &decode_result,
                            event_type,
                            push_acc_id,
                            push_trd_market_str,
                            push.proto_id,
                        ) {
                            continue;
                        }
                        deliveries.push(PendingPushDelivery {
                            target: sub.delivery.clone(),
                            data: payload.clone(),
                            session_id: session_id.clone(),
                            owner_key_id: sub.owner_key_id.clone(),
                            proto_id: push.proto_id,
                        });
                    }
                    deliveries
                };
                for delivery in deliveries {
                    match delivery.target {
                        SubscriberDelivery::LegacyPeer(peer, minimum_level) => {
                            tokio::spawn(async move {
                                let result =
                                    notify_legacy_push(&peer, &minimum_level, delivery.data)
                                        .await
                                        .map_err(|error| error.to_string());
                                if let Err(err) = result {
                                    tracing::warn!(
                                        proto_id = delivery.proto_id,
                                        session_id = delivery.session_id,
                                        owner_key_id = delivery.owner_key_id.as_deref().unwrap_or("<none>"),
                                        error = %err,
                                        "mcp push notification send failed"
                                    );
                                }
                            });
                        }
                        SubscriberDelivery::ModernResource(resource) => {
                            if let Some(work) = resource.enqueue(delivery.data) {
                                tokio::spawn(async move {
                                    resource.send_pending_notification(work).await;
                                });
                            }
                        }
                    }
                }
            }
        });

        // v1.4.39 Phase 5 stale cleanup: 5 分钟跑一次，移除 registered_at > 4h
        // 的订阅者。避免长跑 daemon 累积陈旧 subscriber（客户端断开 /  rmcp
        // session gone 但没显式 unregister 的情况）。
        // issue #54：purge task 与连接代数无关，整个 ServerState 只 spawn 一次。
        let subs_for_purge = Arc::downgrade(&self.push_subscribers);
        if self
            .purge_task_spawns
            .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
        {
            tokio::spawn(async move {
                use std::time::Duration;
                const PURGE_INTERVAL: Duration = Duration::from_secs(5 * 60);
                let mut ticker = tokio::time::interval(PURGE_INTERVAL);
                ticker.tick().await; // skip the immediate first tick
                loop {
                    ticker.tick().await;
                    let Some(subs_for_purge) = subs_for_purge.upgrade() else {
                        break;
                    };
                    let now = std::time::Instant::now();
                    let removed = {
                        let mut subs = subs_for_purge.lock().await;
                        let expired = subs
                            .iter()
                            .filter_map(|(handle, sub)| {
                                now.checked_duration_since(sub.registered_at)
                                    .is_some_and(|age| {
                                        age >= push_subscribers::PUSH_SUBSCRIBER_MAX_AGE
                                    })
                                    .then_some(handle.clone())
                            })
                            .collect::<Vec<_>>();
                        expired
                            .into_iter()
                            .filter_map(|handle| subs.remove(&handle))
                            .collect::<Vec<_>>()
                    };
                    let purged = removed.len();
                    for subscriber in removed {
                        subscriber.close_modern_resource();
                    }
                    if purged > 0 {
                        let remaining = subs_for_purge.lock().await.len();
                        tracing::info!(
                            purged,
                            remaining,
                            max_age_secs = push_subscribers::PUSH_SUBSCRIBER_MAX_AGE.as_secs(),
                            "v1.4.39 Phase 5: purged stale push subscribers (> 4h registered)"
                        );
                    }
                }
            });
        }

        // Reviewer M-4: release the single-flight gate before the network
        // re-registration so concurrent callers are not serialised behind it;
        // `restore_push_registration` is idempotent.
        drop(_connect_gate);
        if generation > 1 {
            self.restore_push_registration(&arc).await;
        }

        Ok(arc)
    }

    /// Whether a failed post-reconnect push re-registration is pending a retry.
    #[cfg(test)]
    pub fn push_registration_dirty(&self) -> bool {
        self.push_registration_dirty
            .load(std::sync::atomic::Ordering::Acquire)
    }

    /// Number of failed post-reconnect `TRD_SUB_ACC_PUSH` re-registrations
    /// (production path logs each failure at warn; the counter is a test seam).
    #[cfg(test)]
    pub fn push_reregister_failures(&self) -> u64 {
        self.push_reregister_failures
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// issue #54：daemon 侧 push 订阅按 conn_id 存，旧连接断开时已被 daemon
    /// `on_disconnect(conn_id)` 清空；重连后把本地仍在册订阅者的 acc_ids 并集
    /// 重发 `TRD_SUB_ACC_PUSH`(2008)，否则 subscriber 留在本地却收不到任何 push。
    /// 失败只 warn，不阻塞 `client()` 返回（调用方可重新 `futu_sub_acc_push`）。
    /// QOT 行情订阅 MCP 没有本地记录，不在此恢复。
    async fn restore_push_registration(&self, client: &Arc<FutuClient>) {
        let acc_ids = {
            let subs = self.push_subscribers.lock().await;
            if subs.is_empty() {
                return;
            }
            let mut acc_ids = subs
                .values()
                .flat_map(|sub| sub.acc_ids.iter().copied())
                .collect::<Vec<u64>>();
            acc_ids.sort_unstable();
            acc_ids.dedup();
            acc_ids
        };
        match crate::handlers::trade::sub_acc_push(client, &acc_ids).await {
            Ok(_) => {
                self.push_registration_dirty
                    .store(false, std::sync::atomic::Ordering::Release);
                tracing::info!(
                    accounts = ?RedactedAccountIdSet::new(&acc_ids),
                    "mcp push subscription re-registered"
                );
            }
            Err(err) => {
                self.push_registration_dirty
                    .store(true, std::sync::atomic::Ordering::Release);
                self.push_reregister_failures
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                tracing::warn!(
                    error = ?RedactedMcpError::new(&err),
                    accounts = ?RedactedAccountIdSet::new(&acc_ids),
                    "mcp push subscription re-register failed after gateway reconnect; \
                     callers must re-run futu_sub_acc_push"
                );
            }
        }
    }

    /// Open a transient C++-compatible internal-UI transport for the pre-login
    /// Verification tool. The ordinary cached MCP client must keep normal
    /// InitConnect semantics and wait for a real login identity.
    pub async fn verification_client(&self) -> Result<Arc<FutuClient>> {
        let gateway = self.gateway().await;
        let config = ClientConfig {
            addr: gateway.clone(),
            client_ver: env!("CARGO_PKG_VERSION").to_string(),
            client_id: futu_core::INTERNAL_UI_CLIENT_ID.to_string(),
            recv_notify: false,
            rsa_key: None,
        };
        let policy =
            ReconnectPolicy::new(MCP_CONNECT_RETRY_DELAY, MCP_CONNECT_RETRY_DELAY, Some(1));
        let mut reconnector = ReconnectingClient::new(config).with_policy(policy);
        let connect_result =
            tokio::time::timeout(MCP_CONNECT_TOTAL_TIMEOUT, reconnector.connect()).await;
        let (client, _push_rx, _info) = match connect_result {
            Ok(result) => result.with_context(|| {
                format!("connect Verification transport to futu gateway at {gateway}")
            })?,
            Err(_) => {
                return Err(anyhow!(
                    "connect Verification transport to futu gateway at {gateway} timed out after {}s",
                    MCP_CONNECT_TOTAL_TIMEOUT.as_secs()
                ));
            }
        };
        Ok(Arc::new(client))
    }
}

// ========== symbol 解析 ==========

pub fn parse_symbol(s: &str) -> Result<Security> {
    let parsed = qot_symbol::parse_qot_symbol_parts(s).map_err(|err| anyhow!("{err}"))?;
    Ok(qot_sdk_adapter::security_from_parsed_symbol(parsed))
}

/// 格式化 Security 为 "MARKET.CODE"
pub fn format_symbol(sec: &Security) -> String {
    qot_symbol::format_qot_symbol(sec.market as i32, &sec.code)
}

/// v1.4.90 P2-C: audit log Option<T> 序列化助手。
///
/// **背景**：之前 audit log 把 `Option<f64>` 用 `?req.price`（tracing 的 Debug
/// shorthand）记录，渲染成 JSON 字符串 `"Some(400.0)"` / `"None"`，下游 jq /
/// DuckDB 数值聚合炸（aggregator 期望 `400.0` number 或 `null`）。
///
/// **修法**：用 NaN sentinel 把 `Option<f64>` flatten 成 `f64`，tracing-subscriber
/// 的 JSON formatter 内部走 `serde_json::Value::from(f64::NAN)` →
/// `Number::from_f64(NaN) = None` → `Value::Null`。
/// 整数 / 字符串同理（i32 → f64 NaN sentinel；&str → "" 哨兵）。
///
/// 验证依据：
/// - `tracing_subscriber::fmt::format::json` line 501 `record_f64` 直接调
///   `serde_json::Value::from(value)`
/// - `serde_json::Value::from(f64)` impl: `Number::from_f64(f).map_or(Value::Null, Value::Number)`
pub mod audit_fmt {
    /// `Option<f64>` → `f64`（None → NaN）。tracing JSON 渲染 NaN 为 `null`。
    #[inline]
    pub fn opt_f64(v: Option<f64>) -> f64 {
        v.unwrap_or(f64::NAN)
    }

    /// `Option<i32>` → `f64`（None → NaN，Some(n) → n as f64）。
    /// i32 ≤ 2^31 < 2^52 mantissa，无精度损失。
    #[inline]
    pub fn opt_i32(v: Option<i32>) -> f64 {
        v.map(f64::from).unwrap_or(f64::NAN)
    }

    /// `Option<&str>` → `&str`（None → ""）。"" 哨兵在 audit 上下文里足以区分
    /// 不传 vs 传空（因为 Symbol / owner 等业务字段不会是空字符串）。
    #[inline]
    pub fn opt_str(v: Option<&str>) -> &str {
        v.unwrap_or("")
    }
}