futu_mcp/

state.rs

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

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::{Context, Result, anyhow};
use futu_auth::{KeyRecord, KeyStore, RuntimeCounters};
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;

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

use push_filter::{
    TradePushDecode, classify_trade_push, subscriber_visible_to_caller, 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,
};

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);

/// 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)
}

/// v1.4.38 Phase 5: 订阅了 push 通知的 MCP 客户端 session 记录。
///
/// 每个 session 用 `futu_sub_acc_push` 注册时登记一条。daemon push 事件来到
/// MCP server 时，按 `acc_ids` 过滤后用 `peer.notify_logging_message()` 推回。
///
/// v1.4.38 100%：`acc_ids` 过滤已生效（state.rs drain loop 实装）。
/// caller key ownership / scope 快照在注册时解析并保存，后续 push 分发不再
/// 重新读取 bearer 明文。
#[derive(Clone)]
pub struct PushSubscriber {
    /// rmcp 对 MCP session 的抽象。clone 便宜（内部 Arc）。
    pub peer: Peer<RoleServer>,
    /// 该 session 关心的 acc_id 列表。空集合表示"不过滤"（接收所有 acc 的 push）。
    pub acc_ids: std::collections::HashSet<u64>,
    /// v1.4.39 per-key acc_id 白名单**注册时快照**（非 live-reload）。
    ///
    /// Some(set) + non-empty → push 的 acc_id 必须在 set 里才推。
    /// Some(empty) / None → 不做 key 级过滤（兼容无 allowed_acc_ids 约束的 key
    /// 或 stdio / legacy 模式）。
    ///
    /// 快照语义：注册后 SIGHUP 重载 keys.json 修改 allowed_acc_ids 不会立即
    /// 反映到已注册订阅者。用户需重新 `futu_sub_acc_push` 才应用新 scope。
    /// 这是 defense-in-depth 层（主 auth 在 tool 调用时 guard.rs），可接受。
    pub allowed_acc_ids_snapshot: Option<std::collections::HashSet<u64>>,
    /// v1.4.105 T-C2: per-key `allowed_markets` 注册时快照, Layer 3 trade push gate.
    ///
    /// `Some(set)` + non-empty → push event 的 `trd_market` 必须 ∈ set 才推 (按
    /// `Trd_Common.TrdMarket` int → 字符串映射, 与 `keys.json::allowed_markets`
    /// 配置字符串一致, e.g. "HK"/"US"/"FUTURES").
    /// `None` / `Some(empty)` → 不做 market gate (兼容 stdio / legacy / 未配
    /// allowed_markets 的 key).
    ///
    /// 与 `allowed_acc_ids_snapshot` 同样**注册时快照**, SIGHUP 重载不影响
    /// 已注册订阅者. 用户需重新 `futu_sub_acc_push` 才应用新 scope.
    /// 配套 main auth (guard.rs / require_acc_read_with_acc_id) 仍在 tool 调用
    /// 时 enforce, 这是 defense-in-depth 层 (push 走 server-initiated channel
    /// 绕过 tool 调用 → 必须独立 enforce).
    pub allowed_markets_snapshot: Option<std::collections::HashSet<String>>,
    /// 注册时间（用于 session 硬上限清理，4h 默认 TTL）
    pub registered_at: std::time::Instant,
    /// v1.4.103 (codex 50 F6 / 53 F4 — B8): owner key id (KeyRecord.id).
    /// 注册时填的 caller key id (HTTP Bearer 或 startup key); 用于 unsub
    /// ownership check — 任何 caller 拿到 session_id 后想 unsub 必须 key id
    /// 匹配 owner_key_id (admin scope 例外).
    ///
    /// None = legacy / stdio 模式无 key (ownership 退化为 "anyone can unsub",
    /// 与本来 v1.4.102 行为一致).
    pub owner_key_id: Option<String>,
}

struct PushDelivery {
    peer: Peer<RoleServer>,
    data: serde_json::Value,
    session_id: String,
    owner_key_id: Option<String>,
    proto_id: u32,
}

/// 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>>,
    /// 交易密码所属登录账号。用于 `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>>>,
}

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

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,
            })),
            enable_trading: false,
            allow_real_trading: false,
            key_store: Arc::new(KeyStore::empty()),
            authed_key: None,
            trade_pwd_account: None,
            counters: Arc::new(RuntimeCounters::new()),
            push_subscribers: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// v1.4.38 Phase 5: 注册当前 session 接收指定 acc_id 的 push。返回 session
    /// UUID（调用方存着，后续可 unregister）。
    ///
    /// v1.4.38: 已 wire 到 `futu_sub_acc_push` tool。tool 被调用时拿到
    /// `RequestContext.peer`，`acc_ids` 从工具 args 解析，注册完成后
    /// state.rs 的 push drain loop 会按 acc_ids filter 转 notify 给该 peer。
    /// v1.4.103 (codex 50 F5 / 53 F2 / 58 F3 — B7) + (codex 50 F6 / 53 F4 — B8):
    /// 注册当前 session 接收指定 acc_id 的 push。
    ///
    /// `owner_key_id_override` 由 caller 传入 (例如从 HTTP Bearer 解析得到 key id);
    /// 若 None → fall back 到 bearer_token 解析 → fall back 到 startup `authed_key.id`.
    /// `allowed_acc_ids_snapshot` 同样 fall back 链: bearer → startup.
    pub async fn register_push_subscriber_with_owner(
        &self,
        peer: Peer<RoleServer>,
        acc_ids: std::collections::HashSet<u64>,
        bearer_token: Option<String>,
        owner_key_id_override: Option<String>,
    ) -> String {
        use std::time::Instant;
        let session_id = format!("sub-{}", rand::random::<u64>());

        // 解析 caller-specific KeyRecord (HTTP Bearer 优先, fallback startup key).
        // 用于 (a) allowed_acc_ids_snapshot (B7) (b) owner_key_id (B8).
        //
        // v1.4.106 codex 0608 F2 (P1): startup fallback 路径用
        // `get_by_id_for_current_machine` 替代裸 `get_by_id`, 与 verify (Bearer 路径
        // 自带 machine 校验) 行为对称, 让 SIGHUP 收紧 allowed_machines 后能立即拒绝.
        let bearer_key_rec = bearer_token
            .as_deref()
            .filter(|pt| !pt.is_empty())
            .and_then(|pt| self.key_store.verify(pt));
        let startup_key_rec = self
            .authed_key
            .as_ref()
            .and_then(|k| self.key_store.get_by_id_for_current_machine(&k.id));
        let effective_key_rec = bearer_key_rec.as_ref().or(startup_key_rec.as_ref());

        // v1.4.103 (B7): allowed_acc_ids_snapshot 优先 HTTP Bearer 解析,
        // 否则 fall back startup key.allowed_acc_ids; 都无 → None (无限制).
        let allowed_acc_ids_snapshot =
            effective_key_rec.and_then(|rec| rec.allowed_acc_ids.clone());

        // v1.4.105 T-C2: allowed_markets_snapshot 同链 — 与 acc_ids_snapshot
        // 同样从 effective_key_rec (Bearer / startup) 取 allowed_markets.
        // 用于 push event Layer 3 market gate (state.rs::subscriber_should_receive_with_market).
        let allowed_markets_snapshot =
            effective_key_rec.and_then(|rec| rec.allowed_markets.clone());

        // v1.4.103 (B8): owner_key_id 优先 caller 显式传入, 否则 effective_key_rec.id.
        let owner_key_id =
            owner_key_id_override.or_else(|| effective_key_rec.map(|rec| rec.id.clone()));

        let subscriber = PushSubscriber {
            peer,
            acc_ids,
            allowed_acc_ids_snapshot,
            allowed_markets_snapshot,
            registered_at: Instant::now(),
            owner_key_id,
        };
        self.push_subscribers
            .lock()
            .await
            .insert(session_id.clone(), subscriber);
        session_id
    }

    /// v1.4.103 (codex 50 F6 / 53 F4 — B8): unsub session ownership check.
    ///
    /// 行为:
    /// - 无 caller_key_id (legacy / stdio): 退化为旧行为 (任何 caller 可 unsub).
    /// - 有 caller_key_id + subscriber.owner_key_id 匹配: 删除, 返 Ok(true).
    /// - 有 caller_key_id + subscriber.owner_key_id 不匹配: **拒绝**, 返
    ///   Err(reason) — 防其他 caller 拿可见 session_id 强制 unsub.
    /// - session_id 不存在: 返 Ok(false) (idempotent, 不报错).
    /// - subscriber.owner_key_id = None (legacy 注册): 退化为旧行为 — 任何 caller
    ///   可 unsub (向后兼容).
    pub async fn unregister_push_subscriber_with_owner_check(
        &self,
        session_id: &str,
        caller_key_id: Option<&str>,
    ) -> Result<bool, String> {
        let mut subs = self.push_subscribers.lock().await;
        // 不存在 → idempotent Ok(false), 不报错
        let Some(sub) = subs.get(session_id) else {
            return Ok(false);
        };
        // ownership check
        match (caller_key_id, sub.owner_key_id.as_deref()) {
            (None, _) => {
                // 无 caller key: legacy / stdio 模式, 退化旧行为
                subs.remove(session_id);
                Ok(true)
            }
            (Some(caller), None) => {
                // session 注册时无 owner_key_id (legacy): 任何 caller 可 unsub
                tracing::warn!(
                    session_id,
                    caller,
                    "v1.4.103 B8: unsub legacy session (no owner_key_id) — \
                     allowed for backward-compat"
                );
                subs.remove(session_id);
                Ok(true)
            }
            (Some(caller), Some(owner)) if caller == owner => {
                subs.remove(session_id);
                Ok(true)
            }
            (Some(caller), Some(owner)) => {
                // ownership 不匹配: reject. 当前 MCP subscription ownership contract
                // 没有 admin override surface；若要扩展，必须先在 spec / auth
                // pipeline / integration tests 中定义清楚。
                Err(format!(
                    "session_id {session_id:?} owned by key_id {owner:?}, \
                     caller key_id {caller:?} not allowed to unsub"
                ))
            }
        }
    }

    /// 当前活跃订阅数。生产诊断使用 `push_subscribers_summary`，这里只作为
    /// state 单测的轻量断言入口保留。
    #[cfg(test)]
    pub async fn push_subscriber_count(&self) -> usize {
        self.push_subscribers.lock().await.len()
    }

    /// v1.4.58 Phase A: 列出所有 push 订阅 summary（tool diagnostic 用）。
    ///
    /// 返 Vec<(session_id, acc_ids, age_secs)>。
    ///
    /// **MED-NEW-3 修（2nd review）**：加 `caller_allowed_acc_ids` 参数做
    /// scope-mode 多租过滤。当 caller 的 key 有 `allowed_acc_ids` 白名单时，
    /// **只返** subscription 的 `acc_ids` 与 caller 白名单有交集的条目。
    /// 避免 agent A（acc_ids=[100, 200]）通过本 tool 看到 agent B 订阅的
    /// acc_id=[300, 400]。
    ///
    /// `caller_allowed_acc_ids=None` / empty → 不过滤（legacy mode / no-scope key）。
    ///
    /// **rmcp 版本兼容**：rmcp 1.4.0 `Peer<RoleServer>` 不实装 `PartialEq`，
    /// 无法按 peer 身份直接过滤。若未来 rmcp 加 PartialEq，可切到更精确的
    /// per-session-owner 过滤（当前只能靠 acc_id 权限交集近似）。
    pub async fn push_subscribers_summary(
        &self,
        caller_allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
    ) -> Vec<(String, std::collections::HashSet<u64>, u64)> {
        let subs = self.push_subscribers.lock().await;
        let now = std::time::Instant::now();
        subs.iter()
            .filter(|(_, sub)| subscriber_visible_to_caller(&sub.acc_ids, caller_allowed_acc_ids))
            .map(|(id, sub)| {
                let age = now
                    .checked_duration_since(sub.registered_at)
                    .map(|d| d.as_secs())
                    .unwrap_or(0);
                // LOW-3RD-1（3rd code review）：scope mode 下返回的 acc_ids 要与
                // caller allowed 求交集 — 防止 caller=[100] 看到 sub=[100, 999]
                // 时知道 999 这个 acc 存在。sub.acc_ids 空集（subscribe-all）不做
                // 交集（概念上 caller 看到的是"有个 catch-all subscriber"，不泄漏
                // 具体账户信息）。
                let visible_accs = match caller_allowed_acc_ids {
                    Some(allowed) if !allowed.is_empty() && !sub.acc_ids.is_empty() => {
                        sub.acc_ids.intersection(allowed).copied().collect()
                    }
                    _ => sub.acc_ids.clone(),
                };
                (id.clone(), visible_accs, age)
            })
            .collect()
    }

    /// 启用交易写工具（构造器式链式设置）
    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
    }

    /// 设置交易密码所属登录账号（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()
    }

    /// 获取（或懒加载）网关客户端
    pub async fn client(&self) -> Result<Arc<FutuClient>> {
        let gateway = {
            let guard = self.inner.lock().await;
            if let Some(c) = &guard.client {
                return Ok(c.clone());
            }
            guard.gateway.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 mut guard = self.inner.lock().await;
            if let Some(c) = &guard.client {
                return Ok(c.clone());
            }
            guard.client = Some(arc.clone());
        }

        // 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);
        // 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 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 => (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() {
                        // v1.4.106 codex 0932 F6 [P2]: trade decode-failed + restricted
                        // key (allowed_acc_ids 非 None) → DROP. 不能让 restricted key
                        // 看到无 acc gate 信息的 trade body 透传 (绕过 ACL).
                        // unrestricted key (allowed_acc_ids None / 空) 仍透传带 decode_status="failed".
                        if matches!(decode_result, TradePushDecode::DecodeFailed) {
                            let restricted = sub
                                .allowed_acc_ids_snapshot
                                .as_ref()
                                .map(|s| !s.is_empty())
                                .unwrap_or(false);
                            if restricted {
                                let key_id = sub.owner_key_id.as_deref().unwrap_or("<none>");
                                // 复用 cross-surface metric — reason="trade_decode_failed"
                                futu_auth::metrics::bump_ws_filtered("trade_decode_failed", key_id);
                                tracing::warn!(
                                    proto_id = push.proto_id,
                                    key_id,
                                    "v1.4.106 audit 0932 F6: trade push body decode failed; \
                                     dropped for restricted key (allowed_acc_ids set, \
                                     cannot ACL-gate body without acc_id)"
                                );
                                continue;
                            }
                            // unrestricted: fall through, broadcast 带 decode_status="failed"
                        }
                        // v1.4.105 F5 fix: 改用 FilterRegistry::should_drop_event.
                        // 行为对齐 4 surface — sub.acc_ids (MCP 显式订阅 list) 喂给
                        // ctx.sub_state (REST sub-acc-push state 同语义); sub.allowed_*
                        // _snapshot 喂给 ctx.allowed_* (caller key 限额).
                        //
                        // **行为微差** (与老 inline fn 一致, 不破老行为):
                        // - sub.acc_ids 空 = MCP 老语义"无限制订阅" → 传 None
                        //   (避免 REST sub_state 空集 tombstone 语义触发 drop-all)
                        // - sub.acc_ids 非空 → 传 Some(&sub.acc_ids), 跟 REST 一致
                        let sub_state_for_ctx = if sub.acc_ids.is_empty() {
                            None
                        } else {
                            Some(&sub.acc_ids)
                        };
                        let ctx = futu_auth_pipeline::PushEventCtx {
                            event_type,
                            event_acc: push_acc_id,
                            allowed_acc_ids: sub.allowed_acc_ids_snapshot.as_ref(),
                            sub_state: sub_state_for_ctx,
                            event_trd_market: push_trd_market_str,
                            allowed_markets: sub.allowed_markets_snapshot.as_ref(),
                        };
                        if filter_registry.should_drop_event(&ctx) {
                            // v1.4.105 T-C2 + F3 (codex review C4): bump filtered metric.
                            // **统一 label 命名** 跟 4 surface 一致 — gRPC subscribe_push
                            // / push_trd_acc / 都用 "trade_market" 标 Layer 3 (allowed_markets)
                            // 拒. 老 "push.trade" 命名是 surface-specific (T-C2 sole), 改成
                            // canonical "trade_market" 让跨 surface metrics jq aggregate 一致.
                            let key_id = sub.owner_key_id.as_deref().unwrap_or("<none>");
                            futu_auth::metrics::bump_ws_filtered("trade_market", key_id);
                            continue;
                        }
                        deliveries.push(PushDelivery {
                            peer: sub.peer.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 {
                    tokio::spawn(async move {
                        let params = rmcp::model::LoggingMessageNotificationParam {
                            level: rmcp::model::LoggingLevel::Info,
                            logger: Some("futu_push".to_string()),
                            data: delivery.data,
                        };
                        if let Err(err) = delivery.peer.notify_logging_message(params).await {
                            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"
                            );
                        }
                    });
                }
            }
        });

        // v1.4.39 Phase 5 stale cleanup: 5 分钟跑一次，移除 registered_at > 4h
        // 的订阅者。避免长跑 daemon 累积陈旧 subscriber（客户端断开 /  rmcp
        // session gone 但没显式 unregister 的情况）。
        let subs_for_purge = Arc::downgrade(&self.push_subscribers);
        tokio::spawn(async move {
            use std::time::Duration;
            const PURGE_INTERVAL: Duration = Duration::from_secs(5 * 60);
            const MAX_AGE: Duration = Duration::from_secs(4 * 3600);
            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 mut subs = subs_for_purge.lock().await;
                let before = subs.len();
                subs.retain(|_, sub| {
                    now.checked_duration_since(sub.registered_at)
                        .map(|age| age < MAX_AGE)
                        .unwrap_or(true)
                });
                let purged = before - subs.len();
                if purged > 0 {
                    tracing::info!(
                        purged,
                        remaining = subs.len(),
                        max_age_secs = MAX_AGE.as_secs(),
                        "v1.4.39 Phase 5: purged stale push subscribers (> 4h registered)"
                    );
                }
            }
        });

        Ok(arc)
    }
}

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

pub fn parse_symbol(s: &str) -> Result<Security> {
    futu_qot::symbol::parse_symbol(s)
}

/// 格式化 Security 为 "MARKET.CODE"
pub fn format_symbol(sec: &Security) -> String {
    futu_qot::symbol::format_symbol(sec)
}

/// 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("")
    }
}