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::{QotMarket, Security};
use rmcp::{RoleServer, service::Peer};
use tokio::sync::Mutex;

#[cfg(test)]
mod tests;

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.105 T-C2: 从 daemon push 的 raw body 里解 `(acc_id, trd_market)`，
/// 用于按 caller key `allowed_markets` 白名单过滤 trade push event。
///
/// 只处理 trade push（TRD_UPDATE_ORDER / TRD_UPDATE_ORDER_FILL），proto Header
/// 同时含 `acc_id` (u64) 与 `trd_market` (i32 enum)。返 `Some((acc_id, trd_market))`
/// 仅当 decode 成功且 s2c.header 存在.
///
/// 行情 push (QOT_UPDATE_*) 不含 trd_market 概念 → 返 None → 调用方按
/// `event_acc_id=None` 路径不做 acc/market gate, 直接 broadcast.
fn extract_acc_id_and_market_from_push(proto_id: u32, body: &[u8]) -> Option<(u64, i32)> {
    use prost::Message;
    match proto_id {
        TRD_UPDATE_ORDER_PROTO_ID => {
            let rsp = futu_proto::trd_update_order::Response::decode(body).ok()?;
            let h = rsp.s2c?.header;
            Some((h.acc_id, h.trd_market))
        }
        TRD_UPDATE_ORDER_FILL_PROTO_ID => {
            let rsp = futu_proto::trd_update_order_fill::Response::decode(body).ok()?;
            let h = rsp.s2c?.header;
            Some((h.acc_id, h.trd_market))
        }
        _ => None,
    }
}

/// v1.4.106 codex 0932 F6/F7: trade push proto_ids (set membership 测试).
const TRD_UPDATE_ORDER_PROTO_ID: u32 = 2208;
const TRD_UPDATE_ORDER_FILL_PROTO_ID: u32 = 2218;

/// v1.4.106 codex 0932 F6 [P2]: 仅按 proto_id 判断是否 trade push 类.
///
/// 修 F6 silent-misclassify bug: 旧实装靠 `extract_acc_id_and_market_from_push`
/// 返 None 推断 "非 trade" — 但 trade push (proto_id 2208/2218) 若 body decode
/// 失败也返 None → 误归 quote → restricted key (有 allowed_acc_ids 限额) 看到
/// 该 push 时按 quote 路径直接 broadcast (绕过 trade-market filter 的 ACL).
///
/// **正确语义**: event_type 由 proto_id 决定 (set membership), 与 body 完整性
/// 无关. body 解码状态另用 `decode_status` 字段表达 (F7).
fn is_trade_push_proto_id(proto_id: u32) -> bool {
    matches!(
        proto_id,
        TRD_UPDATE_ORDER_PROTO_ID | TRD_UPDATE_ORDER_FILL_PROTO_ID
    )
}

/// v1.4.106 codex 0932 F6 [P2]: trade push 的解码结果 + 分发决策.
#[derive(Debug, Clone)]
enum TradePushDecode {
    /// proto_id 不在 trade set — 非 trade push (quote / notify / 其他).
    NotTrade,
    /// trade push body decode 成功 — 含 (acc_id, trd_market).
    Decoded { acc_id: u64, trd_market: i32 },
    /// trade push body decode 失败 — caller 必须按 trade 语义处理 (event_type="trade")
    /// 但无 acc_id / market gate 信息. restricted key 应 drop, unrestricted
    /// 应透传带 `decode_status="failed"`.
    DecodeFailed,
}

fn classify_trade_push(proto_id: u32, body: &[u8]) -> TradePushDecode {
    if !is_trade_push_proto_id(proto_id) {
        return TradePushDecode::NotTrade;
    }
    match extract_acc_id_and_market_from_push(proto_id, body) {
        Some((acc_id, trd_market)) => TradePushDecode::Decoded { acc_id, trd_market },
        None => TradePushDecode::DecodeFailed,
    }
}

/// v1.4.105 T-C2: `Trd_Common.TrdMarket` enum int → 字符串 (与 `keys.json`
/// 配 `allowed_markets` 中字符串一致). 实际映射由 `futu_trd::market`
/// 统一维护，避免 MCP push 过滤、CLI 展示和 trade read projection 漂移.
///
/// 来源: `Trd_Common.proto::TrdMarket` enum.
fn trd_market_int_to_str(i: i32) -> &'static str {
    futu_trd::market::trd_market_label(i).unwrap_or("")
}

/// v1.4.39 Phase 5 filter 核心决策（pure function，便于单测）：
///
/// 判断 push 是否应推给某个订阅者。2 层 gate：
/// 1. **subscriber acc_ids**：`futu_sub_acc_push` 参数里用户指定的 acc_id 列表
///    （空集 = subscribe-all，即不做 subscriber 级过滤）
/// 2. **key allowed_acc_ids** (v1.4.39)：注册时快照的 per-key 白名单。defense-in-depth
///    层，防止 agent 订阅了 key 无权限的 acc_id（主 auth 在 guard.rs tool 调用
///    时 enforce，但 push 是服务端主动发起绕过 tool 调用）
///
/// 两层都 pass 才推。行情类 push (`push_acc_id=None`) 跳过所有 acc_id gate。
/// v1.4.58 MED-NEW-3（2nd code review）: summary 过滤决策 pure function。
///
/// 用于 `push_subscribers_summary` 的 cross-tenant filter —— scope mode 下 caller
/// 只能看到**自己 allowed_acc_ids 有交集的订阅**（防止跨租户泄漏其他 agent 的
/// acc_ids）。
///
/// 规则：
/// - `caller_allowed = None` 或空集（legacy / unrestricted）→ 全可见
/// - 被查订阅的 `sub_acc_ids` 空集（subscribe-all）→ 全可见（没有 specific 账户可泄漏）
/// - 否则 → 只要 `sub_acc_ids` 与 `caller_allowed` 有**任一交集** → 可见
pub(crate) fn subscriber_visible_to_caller(
    sub_acc_ids: &std::collections::HashSet<u64>,
    caller_allowed: Option<&std::collections::HashSet<u64>>,
) -> bool {
    match caller_allowed {
        Some(allowed) if !allowed.is_empty() => {
            sub_acc_ids.is_empty() || sub_acc_ids.iter().any(|a| allowed.contains(a))
        }
        _ => true,
    }
}

/// v1.4.105 T-C2: 生产路径已 migrate 到 `subscriber_should_receive_with_market`
/// (含 market gate Layer 3). 本 helper 保留作向后兼容入口 (老 6 个 filter_*
/// pure-fn 测仍引用), 不入生产路径.
///
/// **v1.4.105 F5 fix**: production migrated to `FilterRegistry::should_drop_event`,
/// 本 fn 跟 `subscriber_should_receive_with_market` 同 `#[cfg(test)]` gate 防
/// dead_code 警告.
#[cfg(test)]
pub(crate) fn subscriber_should_receive(
    push_acc_id: Option<u64>,
    sub_acc_ids: &std::collections::HashSet<u64>,
    key_allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
) -> bool {
    // v1.4.105 T-C2: 旧 API 委托新 API (无 market 信息 → 不做 market gate).
    // 保留为向后兼容入口 (单测仍以 sub_acc_ids + allowed_acc_ids 双 gate 模式验证).
    subscriber_should_receive_with_market(
        push_acc_id,
        None, // 无 market context → 不做 market gate
        sub_acc_ids,
        key_allowed_acc_ids,
        None, // 无 allowed_markets snapshot → 不做 market gate
    )
}

/// v1.4.105 T-C2: subscriber_should_receive 的扩展版 — 加 market gate (Layer 3).
///
/// 完整 3-layer trade push filter:
/// 1. **subscriber acc_ids** (`futu_sub_acc_push` 参数): 空 = subscribe-all
/// 2. **key allowed_acc_ids** (caller key 注册时快照): 硬白名单
/// 3. **key allowed_markets** (caller key 注册时快照, v1.4.105 加): 硬白名单
///
/// 三 gate 都 pass 才推. `event_trd_market=None` (decode 失败 / 行情 push) 跳过
/// market gate. `key_allowed_markets=None` / 空 = 不做 market gate (stdio /
/// legacy / 未配 allowed_markets 的 key).
///
/// 行情类 push (`push_acc_id=None`) 仍按 v1.4.39 行为全推 — 行情无 acc / market
/// 概念, 不参与 trade event filter.
///
/// **v1.4.105 F5 fix (codex review C4)**: production 路径已 migrate 到
/// `FilterRegistry::should_drop_event` (跟 4 surface 一致). 本 fn 现仅作
/// `#[cfg(test)]` unit test target — 直接验证 logic, 不再 production 调用.
/// 保留 + cfg-gate 让 12 既有 test 继续 lock 行为契约 (logic 等价两套断言:
/// 这里 pure-fn assert + cross_surface_invariants FilterRegistry 路径 assert).
#[cfg(test)]
pub(crate) fn subscriber_should_receive_with_market(
    push_acc_id: Option<u64>,
    event_trd_market: Option<i32>,
    sub_acc_ids: &std::collections::HashSet<u64>,
    key_allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
    key_allowed_markets: Option<&std::collections::HashSet<String>>,
) -> bool {
    let Some(aid) = push_acc_id else {
        return true; // 非 trade push，全推（行情类无 key 级 gate 概念）
    };
    // Layer 1: 订阅者 acc_ids 过滤
    let sub_gate = sub_acc_ids.is_empty() || sub_acc_ids.contains(&aid);
    // Layer 2: caller key allowed_acc_ids 硬白名单
    let key_acc_gate = match key_allowed_acc_ids {
        Some(allowed) if !allowed.is_empty() => allowed.contains(&aid),
        _ => true, // 无 key 级约束（allowed_acc_ids=None 或空）/ stdio 模式
    };
    // Layer 3 (v1.4.105 T-C2): caller key allowed_markets 硬白名单
    //
    // 行为契约:
    // - `key_allowed_markets=None` 或空集 → 不做 market gate (向后兼容 + stdio)
    // - `event_trd_market=None` → 不做 market gate (decode 失败保守 — 不 silent
    //   drop, 让 caller 看到原始 push, downstream 单测覆盖 decode 失败路径)
    // - `event_trd_market=Some(int)` → 转字符串与 allowed 集合比对
    //   (注: int=0/未知 → trd_market_int_to_str 返 "" → 不在任何非空 allowed
    //    集合内 → drop. 这是 fail-closed 语义: 未知 market 配 restricted key
    //    应 drop 而非 silently leak)
    let key_market_gate = match key_allowed_markets {
        Some(allowed) if !allowed.is_empty() => match event_trd_market {
            Some(int) => allowed.contains(trd_market_int_to_str(int)),
            None => true, // decode 失败 → 不 drop (向后兼容 — 单独 metric 提示)
        },
        _ => true, // 无 market 限制
    };
    sub_gate && key_acc_gate && key_market_gate
}

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

/// MCP server 运行时状态
#[derive(Clone)]
pub struct ServerState {
    /// [`Inner`] 共享可变状态（gateway 地址 + 懒加载的 [`FutuClient`]）
    pub inner: Arc<Mutex<Inner>>,
    /// 是否启用交易写工具（place/modify/cancel）。默认 false。旧开关，仅当
    /// `key_store.is_configured() == false` 时生效。
    pub enable_trading: bool,
    /// 是否允许对 real 环境下单。默认 false。旧开关，同上。
    pub allow_real_trading: bool,
    /// keys.json 加载的 KeyStore。`is_configured()` 为 true 时走 scope 授权模式。
    pub key_store: Arc<KeyStore>,
    /// 调用方传入的 API Key 对应的记录；None 表示未提供 key。
    pub authed_key: Option<Arc<KeyRecord>>,
    /// 交易密码所属登录账号。用于 `futu_unlock_trade` 从账号级 keychain
    /// `trade-password.<login-account>` 读取密码；None 时走 legacy/global/env 兼容路径。
    pub trade_pwd_account: Option<String>,
    /// 限额运行时（日累计计数器）
    pub 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）。
    pub push_subscribers: Arc<Mutex<HashMap<String, PushSubscriber>>>,
}

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

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

        let config = ClientConfig {
            addr: guard.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 gateway = guard.gateway.clone();
        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()
                ));
            }
        };

        // 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 = self.push_subscribers.clone();
        // 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 subs = subs_for_push.lock().await;
                if subs.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),
                });
                for sub in subs.values() {
                    // 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 codex 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;
                    }
                    let peer = sub.peer.clone();
                    let data = payload.clone();
                    tokio::spawn(async move {
                        let params = rmcp::model::LoggingMessageNotificationParam {
                            level: rmcp::model::LoggingLevel::Info,
                            logger: Some("futu_push".to_string()),
                            data,
                        };
                        let _ = peer.notify_logging_message(params).await;
                    });
                }
            }
        });

        // v1.4.39 Phase 5 stale cleanup: 5 分钟跑一次，移除 registered_at > 4h
        // 的订阅者。避免长跑 daemon 累积陈旧 subscriber（客户端断开 /  rmcp
        // session gone 但没显式 unregister 的情况）。
        let subs_for_purge = self.push_subscribers.clone();
        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 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)"
                    );
                }
            }
        });

        let arc = Arc::new(client);
        guard.client = Some(arc.clone());
        Ok(arc)
    }
}

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

/// 解析 "MARKET.CODE" 格式的 symbol
pub fn parse_symbol(s: &str) -> Result<Security> {
    let (market_str, code) = s.split_once('.').ok_or_else(|| {
        anyhow!("invalid symbol {s:?}: expected MARKET.CODE (e.g. HK.00700, US.AAPL, SH.600519)")
    })?;
    if code.is_empty() {
        return Err(anyhow!("invalid symbol {s:?}: code part is empty"));
    }
    let market = match market_str.to_ascii_uppercase().as_str() {
        "HK" => QotMarket::HkSecurity,
        "HK_FUTURE" => QotMarket::HkFuture,
        "US" => QotMarket::UsSecurity,
        "SH" => QotMarket::CnshSecurity,
        "SZ" => QotMarket::CnszSecurity,
        "SG" => QotMarket::SgSecurity,
        "JP" => QotMarket::JpSecurity,
        "AU" => QotMarket::AuSecurity,
        "MY" => QotMarket::MySecurity,
        "CA" => QotMarket::CaSecurity,
        "FX" => QotMarket::FxSecurity,
        "CRYPTO" | "CC" => QotMarket::Crypto,
        other => {
            return Err(anyhow!(
                "invalid symbol {s:?}: unknown market {other:?} (HK|HK_FUTURE|US|SH|SZ|SG|JP|AU|MY|CA|FX|CRYPTO|CC)"
            ));
        }
    };
    Ok(Security::new(market, code))
}

/// 格式化 Security 为 "MARKET.CODE"
pub fn format_symbol(sec: &Security) -> String {
    let m = match sec.market {
        QotMarket::HkSecurity => "HK",
        QotMarket::HkFuture => "HK_FUTURE",
        QotMarket::UsSecurity => "US",
        QotMarket::CnshSecurity => "SH",
        QotMarket::CnszSecurity => "SZ",
        QotMarket::SgSecurity => "SG",
        QotMarket::JpSecurity => "JP",
        QotMarket::AuSecurity => "AU",
        QotMarket::MySecurity => "MY",
        QotMarket::CaSecurity => "CA",
        QotMarket::FxSecurity => "FX",
        QotMarket::Crypto => "CRYPTO",
        QotMarket::Unknown => "UNKNOWN",
        _ => "UNKNOWN",
    };
    format!("{m}.{}", 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("")
    }
}