futu_server/

ws_listener.rs

// WebSocket 监听器：接受 WebSocket 连接，复用 TCP 的请求路由和连接池
//
// 每个 WebSocket 二进制消息 = 一个完整的 FutuAPI 帧（44 字节帧头 + body）。
// 与 TCP 共享同一个 connections DashMap、RequestRouter、SubscriptionManager。
//
// ## v1.0 鉴权
//
// 握手阶段（accept_hdr_async）校验 HTTP `Authorization: Bearer <token>` 或
// `?token=<plaintext>` query —— 通过 `KeyStore::verify` 得到 `KeyRecord`，
// 把 scope 集合和 key_id 存到 `ClientConn`。每条消息进 `ws_process_requests`
// 时按 `futu_auth::scope_for_proto_id(proto_id)` 查所需 scope，不匹配 → 不 dispatch、
// 记 audit reject。`trade:real` 额外跑 `check_and_commit` 过一道 rate + hours
// 全局闸门。未注入 KeyStore（TCP listener 或 legacy 模式）→ scopes 空集被
// 解释为"全放行"，保持向后兼容。
//
// ## v1.4.104 阶段 3
//
// 把 inline scope gate / rate gate / body-aware acc_id check 替换为
// `futu_auth_pipeline::authenticate_request` 单一调用. 流程:
//   1. AES decrypt (handshake protocol INIT_CONNECT 跳过)
//   2. (非 INIT_CONNECT + 非 1xxx 系统协议) → 调 pipeline
//      Credential::PreVerified(rec_from_get_by_id) 拿 SIGHUP-aware 最新 rec.
//      Reject → drop. Allow → 取 allowed_acc_ids 给 response filter.
//   3. dispatch (router / handle_init_connect / handle_keepalive)
//   4. response filter (proto 2001 TRD_GET_ACC_LIST 等)
//
// 删除: 旧 `ws_body_aware_check` (功能进 pipeline body_aware::build_check_ctxs).
// 删除: 旧 inline scope gate + rate gate + audit allow/reject. 全 pipeline 一处.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Instant;

use dashmap::DashMap;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, watch};

mod connection;
mod handshake;

use connection::prepare_ws_connection;
use handshake::AuthResult;

use futu_auth::{KeyStore, RuntimeCounters};
use futu_auth_pipeline::{
    AuthDecision, AuthEnvelope, Credential, Endpoint, FilterRegistry, RejectKind, SurfaceId,
    authenticate_request,
};

/// v1.4.106 D1 5c: WS surface adapter — `AuthDecision::Reject` 翻成 silent drop.
///
/// **历史**: WS 在 v1.4.103 / v1.4.104 阶段 3 都按 silent drop 处理 reject
/// (与 v1.4.103 行为一致, 防 timing 探测 — 给客户端任何 wire response 都让
/// 它知道帧被读了 vs 没读). v1.4.106 D1 把这层"翻译为 unit"也走 trait, 让 4
/// surface SurfaceAdapter 一致.
///
/// **WireResponse = ()**: WS 不发任何东西回 client; 调用方拿 `Option<()>`
/// 知道是 reject (Some) 还是 allow (None) 即可继续 dispatch / 丢弃.
///
/// **不变量**: `translate_reject` 不能写日志 (pipeline 已 audit::reject 一次,
/// 不要重复). 只 drain reason / kind 让它进 `_`.
pub struct WsAdapter;

impl futu_auth_pipeline::SurfaceAdapter for WsAdapter {
    type WireResponse = ();

    fn surface_id() -> SurfaceId {
        SurfaceId::Ws
    }

    fn translate_reject(_kind: RejectKind, _reason: String) -> Self::WireResponse {
        // 与 v1.4.103/104 行为一致: silent drop. pipeline 已 audit reject,
        // 不再写 log; reject fields intentionally unused to avoid leaking daemon state.
    }
}
use futu_codec::header::ProtoFmtType;
use futu_core::proto_id;
use futu_core::server_time::ServerTimeAnchorStore;

use crate::conn::{ClientConn, ConnState, DisconnectNotify, IncomingRequest};
use crate::listener::{
    MAX_CONNECTIONS, PER_CONNECTION_REQUEST_QUEUE_CAPACITY, ServerConfig,
    default_server_time_store, server_now_ts,
};
use crate::listener_status::{
    ListenerBindEventSender, ListenerSurface, notify_listener_failed, notify_listener_opened,
};
use crate::router::RequestRouter;

/// WebSocket 服务端
pub struct WsServer {
    listen_addr: String,
    config: ServerConfig,
    connections: Arc<DashMap<u64, ClientConn>>,
    router: Arc<RequestRouter>,
    subscriptions: Option<Arc<crate::subscription::SubscriptionManager>>,
    /// v1.0：握手时做 Bearer token 鉴权。None 或 `!is_configured()` → legacy 模式放行
    key_store: Option<Arc<KeyStore>>,
    /// v1.0：跨 REST / gRPC / WS 共享的限额 counters
    counters: Option<Arc<RuntimeCounters>>,
    /// v1.4.104 阶段 3: 跨 surface 共享的 response filter registry (proto 2001
    /// TRD_GET_ACC_LIST 默认装入). None 时 fallback 到内置 with_defaults.
    filter_registry: Option<Arc<FilterRegistry>>,
    server_time_store: Arc<ServerTimeAnchorStore>,
}

/// Shared dependencies for the WebSocket server.
///
/// These are the same runtime objects the raw TCP server owns. Keeping them in a
/// bundle avoids every constructor carrying a growing positional list as auth /
/// counters / filters evolve.
pub struct WsServerDeps {
    connections: Arc<DashMap<u64, ClientConn>>,
    router: Arc<RequestRouter>,
    subscriptions: Option<Arc<crate::subscription::SubscriptionManager>>,
}

impl WsServerDeps {
    pub fn new(
        connections: Arc<DashMap<u64, ClientConn>>,
        router: Arc<RequestRouter>,
        subscriptions: Option<Arc<crate::subscription::SubscriptionManager>>,
    ) -> Self {
        Self {
            connections,
            router,
            subscriptions,
        }
    }
}

impl WsServer {
    /// 创建 WsServer，共享 TCP 的连接池、路由器、订阅管理器（无鉴权，向后兼容）
    pub fn new(
        listen_addr: String,
        config: ServerConfig,
        connections: Arc<DashMap<u64, ClientConn>>,
        router: Arc<RequestRouter>,
        subscriptions: Option<Arc<crate::subscription::SubscriptionManager>>,
    ) -> Self {
        Self::with_auth(
            listen_addr,
            config,
            WsServerDeps::new(connections, router, subscriptions),
            None,
            None,
        )
    }

    /// v1.0 入口：同时接入 KeyStore + 共享 RuntimeCounters 做握手鉴权和 per-message
    /// scope / 限额检查。`key_store = None` 或未配置时保持 legacy（全放行）。
    pub fn with_auth(
        listen_addr: String,
        config: ServerConfig,
        deps: WsServerDeps,
        key_store: Option<Arc<KeyStore>>,
        counters: Option<Arc<RuntimeCounters>>,
    ) -> Self {
        Self {
            listen_addr,
            config,
            connections: deps.connections,
            router: deps.router,
            subscriptions: deps.subscriptions,
            key_store,
            counters,
            filter_registry: None,
            server_time_store: default_server_time_store(),
        }
    }

    /// Inject the shared backend server-time anchor for SDK-facing WS fields.
    pub fn with_server_time_store(mut self, store: Arc<ServerTimeAnchorStore>) -> Self {
        self.server_time_store = store;
        self
    }

    /// v1.4.104 阶段 3: 注入显式 FilterRegistry (跨 surface 共享同一份).
    /// 不调用此 setter 则 run() 时 fallback 到 `FilterRegistry::with_defaults()`.
    pub fn with_filter_registry(mut self, registry: Arc<FilterRegistry>) -> Self {
        self.filter_registry = Some(registry);
        self
    }

    /// 启动 WebSocket 服务端监听
    pub async fn run(&self) -> anyhow::Result<()> {
        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
        self.run_until_shutdown(shutdown_rx).await
    }

    /// 启动 WebSocket 服务端监听，并在 shutdown 信号到来时停止接受新连接。
    pub async fn run_until_shutdown(
        &self,
        shutdown_rx: watch::Receiver<bool>,
    ) -> anyhow::Result<()> {
        self.run_until_shutdown_with_listener_events(shutdown_rx, None)
            .await
    }

    /// Run until shutdown and report the exact socket bind result to startup.
    pub async fn run_until_shutdown_with_listener_events(
        &self,
        mut shutdown_rx: watch::Receiver<bool>,
        listener_events: Option<ListenerBindEventSender>,
    ) -> anyhow::Result<()> {
        let listener = TcpListener::bind(&self.listen_addr)
            .await
            .map_err(|error| {
                notify_listener_failed(&listener_events, ListenerSurface::WebSocket);
                anyhow::Error::new(crate::bind_hint::io_bind_error(
                    "WebSocket",
                    "--websocket-port",
                    &self.listen_addr,
                    error,
                ))
            })?;
        tracing::info!(addr = %self.listen_addr, "WebSocket server listening");

        // The disconnect cleanup signal is intentionally unbounded: each item
        // is a tiny `conn_id`, and blocking this path behind request-queue
        // backpressure would risk leaking connection/subscription state.
        let (disconnect_tx, mut disconnect_rx) = mpsc::unbounded_channel::<DisconnectNotify>();

        // v1.4.104 阶段 3: KeyStore + RuntimeCounters 总是材料化 (legacy mode 用
        // empty / new()), 让 ws_process_requests 拿 non-Option Arc 直接调 pipeline.
        // 行为与 v1.4.103 等价: KeyStore::empty().is_configured() = false, pipeline
        // 走 legacy short-circuit (Allow{rec:None} 不 audit, body-aware 不 enforce).
        let key_store_for_process = self
            .key_store
            .clone()
            .unwrap_or_else(|| Arc::new(KeyStore::empty()));
        let counters_for_process = self
            .counters
            .clone()
            .unwrap_or_else(|| Arc::new(RuntimeCounters::new()));
        let filter_registry_for_process = self
            .filter_registry
            .clone()
            .unwrap_or_else(|| Arc::new(FilterRegistry::with_defaults()));
        // 启动连接清理任务
        let cleanup_connections = Arc::clone(&self.connections);
        let cleanup_subs = self.subscriptions.clone();
        tokio::spawn(async move {
            while let Some(notify) = disconnect_rx.recv().await {
                let removed = cleanup_connections.remove(&notify.conn_id);
                if removed.is_some() {
                    if let Some(ref subs) = cleanup_subs {
                        subs.on_disconnect(notify.conn_id);
                    }
                    tracing::info!(
                        conn_id = notify.conn_id,
                        remaining = cleanup_connections.len(),
                        "ws connection removed from pool"
                    );
                }
            }
        });

        // 接受连接循环
        let connections = Arc::clone(&self.connections);
        let key_store_accept = self.key_store.clone();
        // v1.4.104 阶段 3: scope_mode 局部计算 (KeyStore configured = 启用 auth).
        let scope_mode = self.key_store.as_ref().is_some_and(|ks| ks.is_configured());
        if !scope_mode {
            // v1.4.93 P0-5 (NEW-C-02): 加强 legacy mode loud WARN —
            // 对齐 REST mutating-blocked policy 的 startup signal。任何
            // 未授权客户端都可 handshake + 接收 push（HTTP 101 OK）。本版
            // **不 reject**（保持向后兼容），未来 v2 默认 reject。
            tracing::warn!("{}", legacy_mode_warn_tracing_message());
            eprintln!("{}", legacy_mode_warn_stderr_message());
        }
        let _serving =
            notify_listener_opened(&listener_events, ListenerSurface::WebSocket, &shutdown_rx)
                .await?;
        drop(listener_events);

        loop {
            let (stream, peer_addr) = tokio::select! {
                _ = crate::listener::shutdown_requested(&mut shutdown_rx) => {
                    tracing::info!("WebSocket server accept loop stopped by shutdown signal");
                    break;
                }
                accepted = listener.accept() => accepted?,
            };

            if connections.len() >= MAX_CONNECTIONS {
                tracing::warn!(
                    peer = %peer_addr,
                    "max connections reached ({}), rejecting ws client",
                    MAX_CONNECTIONS,
                );
                drop(stream);
                continue;
            }

            let conn_id = ClientConn::generate_conn_id();
            let session_generation = ClientConn::generate_session_generation();
            let aes_key = ClientConn::generate_aes_key();
            crate::listener::set_nodelay_with_log(&stream, peer_addr, "ws");

            tracing::info!(
                conn_id = conn_id,
                peer = %peer_addr,
                total = connections.len() + 1,
                "ws client connected"
            );

            let (req_tx, req_rx) =
                mpsc::channel::<IncomingRequest>(PER_CONNECTION_REQUEST_QUEUE_CAPACITY);
            let (tx, authed, close_control, io_start) = prepare_ws_connection(
                stream,
                peer_addr,
                conn_id,
                aes_key,
                req_tx,
                disconnect_tx.clone(),
                shutdown_rx.clone(),
                key_store_accept.clone(),
            )
            .await;

            // 握手鉴权失败 → run_ws_connection 已经 drop 连接；这里什么都不做
            let Some(authed) = authed else {
                continue;
            };
            let (key_id, scopes, allowed_markets, allowed_acc_ids) = match authed {
                AuthResult::Authenticated(rec) => (
                    Some(rec.id.clone()),
                    rec.scopes.clone(),
                    // v1.4.105 D3 (Phase 4) T-B2: 拷贝 caller key 的 allowed_markets
                    // 到 ClientConn 让 PushDispatcher::push_trd_acc Layer 3 用.
                    rec.allowed_markets
                        .as_ref()
                        .map(|s| std::sync::Arc::new(s.clone())),
                    // codex round 1 F4 (P2) v1.4.105: 拷贝 caller key 的
                    // allowed_acc_ids 到 ClientConn 让 PushDispatcher::push_trd_acc
                    // Layer 1 push-time 硬过滤. 防 stale subscription /
                    // KeyRecord reload 后 acc 范围窄化 时 push leak.
                    rec.allowed_acc_ids
                        .as_ref()
                        .map(|s| std::sync::Arc::new(s.clone())),
                ),
                AuthResult::Legacy => (None, HashSet::new(), None, None),
            };

            let conn = ClientConn {
                conn_id,
                session_generation,
                state: ConnState::Connected,
                aes_key,
                aes_encrypt_enabled: false,
                proto_fmt_type: ProtoFmtType::Protobuf,
                last_keepalive: Instant::now(),
                recv_notify: false,
                ai_type: 0,
                keepalive_count: std::sync::atomic::AtomicU32::new(0),
                tx,
                key_id,
                scopes,
                allowed_markets,
                allowed_acc_ids,
            };

            connections.insert(conn_id, conn);
            if let Some(ref subscriptions) = self.subscriptions {
                subscriptions.on_connect(conn_id, session_generation);
                subscriptions.register_client_close_control(conn_id, close_control);
            }
            tokio::spawn(ws_process_requests(
                req_rx,
                Arc::clone(&connections),
                Arc::clone(&self.router),
                self.subscriptions.clone(),
                self.config.clone(),
                Arc::clone(&counters_for_process),
                Arc::clone(&key_store_for_process),
                Arc::clone(&filter_registry_for_process),
                Arc::clone(&self.server_time_store),
                peer_addr.ip().is_loopback(),
            ));
            io_start.start();
        }

        Ok(())
    }
}

/// v1.4.93 P0-5 (NEW-C-02): legacy mode 的 `tracing::warn!` 内容。
///
/// 抽出 const fn 以便单测验证 warn 消息携带 "v2"/"reject" 等关键提示词，
/// 防止后续被误删（同模式 v1.4.86 SEC-003 Q4 已沉淀）。
pub(crate) const fn legacy_mode_warn_tracing_message() -> &'static str {
    "WS server running WITHOUT API key auth (legacy mode); \
     all WS clients accept unauthenticated handshake (no-token / \
     wrong-bearer / bogus-query all return success). \
     Pass KeyStore via with_auth() to enable. \
     v2 will default-reject; migrate to --rest-keys-file / --ws-keys-file for production."
}

/// v1.4.93 P0-5 (NEW-C-02): legacy mode 的 stderr 用户可见消息。
///
/// 比 tracing::warn 更短，方便 systemd / docker logs 一行抓住。
pub(crate) const fn legacy_mode_warn_stderr_message() -> &'static str {
    "⚠️  WS server (legacy mode, no --ws-keys-file): \
     unauthenticated handshakes accepted. v2 will default-reject. \
     Migrate to --ws-keys-file for production."
}

/// 处理 WebSocket 连接的请求（逻辑与 TCP 的 process_requests 相同，额外做 scope / 限额）
///
/// v1.4.104 阶段 3 重构: inline scope gate + trade:real rate gate +
/// `ws_body_aware_check` 三段折叠为单一 `authenticate_request` pipeline 调用.
///
/// **流程**:
/// 1. AES decrypt (handshake INIT_CONNECT 跳过 — body 是明文 RSA 加密 InitReq)
/// 2. 非 INIT_CONNECT + scope_for_proto_id != None → 调 pipeline:
///    - Credential::PreVerified(`KeyStore::get_by_id(key_id)`) 拿 SIGHUP-aware
///      最新 rec, 复用 handshake 已 verify 的身份 (skip re-verify).
///    - Endpoint::Proto(proto_id), commit_rate=true (per-msg rate 闸门).
///    - Reject → drop request (audit 已 emit). Allow → 拿 allowed_acc_ids.
/// 3. dispatch (router / handle_init_connect / handle_keepalive)
/// 4. response filter (proto 2001 等, 通过 FilterRegistry::apply).
///
/// **legacy mode 行为**: KeyStore::empty().is_configured() = false, pipeline
/// 走 legacy short-circuit (Allow{rec:None}, 不 audit, body-aware 不 enforce).
/// 1xxx 系统协议 (INIT_CONNECT / KEEP_ALIVE / GET_GLOBAL_STATE) 与 v1.4.103
/// 一致跳过 pipeline 直接 dispatch (handshake / heartbeat / 公开协议无 scope).
async fn ws_process_requests(
    mut req_rx: mpsc::Receiver<IncomingRequest>,
    connections: Arc<DashMap<u64, ClientConn>>,
    router: Arc<RequestRouter>,
    subscriptions: Option<Arc<crate::subscription::SubscriptionManager>>,
    config: ServerConfig,
    counters: Arc<RuntimeCounters>,
    key_store: Arc<KeyStore>,
    filter_registry: Arc<FilterRegistry>,
    server_time_store: Arc<ServerTimeAnchorStore>,
    peer_is_loopback: bool,
) {
    use crate::listener::ApiServer;

    let mut pending_init_connect: Option<tokio::task::JoinHandle<()>> = None;
    while let Some(mut req) = req_rx.recv().await {
        req.caller_is_loopback = peer_is_loopback;
        req.caller_legacy_local_mode = !key_store.is_configured();
        let Some(session_generation) = connections
            .get(&req.conn_id)
            .map(|conn| conn.session_generation)
        else {
            continue;
        };
        req.session_generation = session_generation;
        req.caller_has_auth_setup_scope = connections
            .get(&req.conn_id)
            .is_some_and(|conn| conn.scopes.contains(&futu_auth::Scope::AuthSetup));
        let conn_id = req.conn_id;
        let proto_id_val = req.proto_id;
        let serial_no = req.serial_no;

        // 更新 last_keepalive（任何包都算活跃）
        if let Some(mut conn) = connections.get_mut(&conn_id) {
            conn.last_keepalive = Instant::now();
        }

        // ── Step 0: v1.4.106 codex 0532 F3 (P2): daemon-internal proto_id
        // (高位 0x8000_0000 bit) 绝不应从 raw WS 公开 surface 进入 — 仅 REST
        // handler 内部合成给 router. 在 AES decrypt 前显式 reject + log,
        // 防探测 daemon 内部 routing.
        if futu_auth::is_internal_proto_id(proto_id_val) {
            tracing::warn!(
                conn_id,
                proto_id = proto_id_val,
                "rejecting daemon-internal proto_id at raw WS public surface (audit 0532 F3)"
            );
            continue;
        }

        if proto_id_val != proto_id::INIT_CONNECT
            && connections
                .get(&conn_id)
                .is_some_and(|conn| conn.state == crate::conn::ConnState::Connected)
            && !crate::identity::StartupReadiness::is_prelogin_proto(proto_id_val)
        {
            if router.startup_readiness().snapshot().state != crate::identity::StartupState::Ready {
                if let Some(body) = router.dispatch(conn_id, &req).await
                    && ApiServer::send_response(
                        connections.as_ref(),
                        conn_id,
                        proto_id_val,
                        serial_no,
                        body,
                    )
                    .await
                {
                    req.mark_response_committed();
                }
                continue;
            }
            let Some(pending) = pending_init_connect.take() else {
                tracing::warn!(
                    conn_id,
                    proto_id = proto_id_val,
                    "dropping WS request from Ready connection with no completed InitConnect"
                );
                continue;
            };
            if let Err(error) = pending.await {
                tracing::warn!(conn_id, error = %error, "pending WS InitConnect task failed");
                continue;
            }
            if connections
                .get(&conn_id)
                .is_none_or(|conn| conn.state == crate::conn::ConnState::Connected)
            {
                continue;
            }
        }

        // ── Step 1: AES decrypt (always for non-INIT_CONNECT) ─────────────────
        // pipeline + body-aware 都需 plaintext body. INIT_CONNECT body 是 RSA
        // 加密的 ConnInitReq, 由 handle_init_connect 自行 RSA decrypt.
        if proto_id_val != proto_id::INIT_CONNECT
            && let Some(conn) = connections.get(&conn_id)
            && conn.aes_encrypt_enabled
        {
            match conn.decrypt_body(&req.body) {
                Ok(decrypted) => {
                    req.body = bytes::Bytes::from(decrypted);
                }
                Err(e) => {
                    tracing::warn!(
                        conn_id = conn_id,
                        proto_id = proto_id_val,
                        error = %e,
                        "ws AES decrypt request failed, dropping"
                    );
                    continue;
                }
            }
        }

        // ── Step 2: Pipeline auth ─────────────────────────────────────────────
        // INIT_CONNECT (handshake) + 1xxx 系统协议 (scope_for_proto_id == None)
        // 跳 pipeline. 与 v1.4.103 行为一致: 系统协议不审 / 不限频 / 直 dispatch.
        let needed_scope = futu_auth_pipeline::capability::scope_for_proto_id(proto_id_val);
        // codex 0522 F1 v1.4.106: 提前抓 conn.key_id 快照让 dispatch IncomingRequest
        // 也能填 caller_key_id (per-call snapshot, 与 caller_allowed_acc_ids 同源).
        // 即使 INIT_CONNECT / 1xxx 系统协议 (跳 pipeline) 也带上 — handler 可基于
        // key_id 做 per-key 订阅配额 / cleanup / 审计.
        let dispatch_caller_key_id: Option<String> =
            connections.get(&conn_id).and_then(|c| c.key_id.clone());
        let allowed_acc_ids_for_resp_filter: Option<HashSet<u64>> =
            if proto_id_val == proto_id::INIT_CONNECT || needed_scope.is_none() {
                None
            } else {
                // 从 conn 取 key_id 快照, 然后 KeyStore::get_by_id 拿 SIGHUP-aware
                // 最新 rec (limits / scopes / allowed_acc_ids 都跟最新). 找不到 ↦
                // Credential::None (legacy mode 放行 / scope mode reject Unauth).
                let key_id_snap = dispatch_caller_key_id.clone();
                let rec_opt = key_id_snap.as_ref().and_then(|id| key_store.get_by_id(id));
                let credential = match rec_opt {
                    Some(rec) => Credential::PreVerified(rec),
                    None => Credential::None,
                };

                let env = AuthEnvelope {
                    surface: SurfaceId::Ws,
                    endpoint: Endpoint::Proto(proto_id_val),
                    needed_scope,
                    credential,
                    proto_id: Some(proto_id_val),
                    body: &req.body,
                    explicit_acc_id: None,
                    explicit_ctx: None,
                    commit_rate: true, // WS per-msg 是 trade:real 唯一 rate gate
                    audit_emit: true,
                };
                let session_id = conn_id.to_string();
                let audit_ctx =
                    futu_auth::audit::AuditContext::new(None::<&str>, Some(session_id.as_str()));

                // v1.4.106 D1 5c: 走 SurfaceAdapter trait
                // (`WsAdapter::translate_decision`), 与 4 surface 一致.
                // Allow → Some(allowed_acc_ids), Reject → None + silent drop.
                use futu_auth_pipeline::SurfaceAdapter;
                match futu_auth::audit::with_context(audit_ctx.clone(), || {
                    authenticate_request(&key_store, &counters, env)
                }) {
                    AuthDecision::Allow {
                        allowed_acc_ids, ..
                    } => allowed_acc_ids,
                    decision @ AuthDecision::Reject { .. } => {
                        // WsAdapter::translate_decision 返 Some(()) 表 reject
                        // (silent drop). pipeline 已 audit reject, 这里不打 log.
                        let silent_drop = WsAdapter::translate_decision(decision);
                        debug_assert!(silent_drop.is_some());
                        continue;
                    }
                }
            };

        // ── Step 3: Dispatch ──────────────────────────────────────────────────
        let response_body = match proto_id_val {
            proto_id::INIT_CONNECT => match crate::conn::start_init_connect_for_startup(
                Arc::clone(&connections),
                router.startup_readiness().clone(),
                conn_id,
                &req.body,
                serial_no,
                config.server_ver,
                config.keepalive_interval,
                config.rsa_private_key.clone(),
            ) {
                Ok(crate::conn::InitConnectStart::Immediate(prepared)) => {
                    if let Err(error) = crate::conn::complete_prepared_init_connect(
                        connections.as_ref(),
                        conn_id,
                        serial_no,
                        prepared,
                    )
                    .await
                    {
                        tracing::warn!(conn_id, error = %error, "ws InitConnect response failed");
                    }
                    None
                }
                Ok(crate::conn::InitConnectStart::Deferred(task)) => {
                    if let Some(previous) = pending_init_connect.replace(task) {
                        previous.abort();
                    }
                    None
                }
                Err(error) => {
                    tracing::warn!(
                        conn_id,
                        proto_id = proto_id_val,
                        error = %error,
                        "ws InitConnect handling failed"
                    );
                    None
                }
            },
            proto_id::KEEP_ALIVE => match connections.get(&conn_id) {
                Some(conn) => {
                    match conn.handle_keepalive_at(&req.body, server_now_ts(&server_time_store)) {
                        Ok(body) => Some(body),
                        Err(error) => {
                            tracing::warn!(
                                conn_id,
                                proto_id = proto_id_val,
                                error = %error,
                                "ws KeepAlive handling failed"
                            );
                            None
                        }
                    }
                }
                None => {
                    tracing::warn!(
                        conn_id,
                        proto_id = proto_id_val,
                        "ws KeepAlive request received for missing connection"
                    );
                    None
                }
            },
            _ => {
                // v1.4.105 D2 T-A1 fix: caller_allowed_acc_ids 从 pipeline allow
                // decision 真填进 IncomingRequest, 让 dispatch handler (e.g.
                // SubAccPushHandler) 端 enforce per-acc whitelist defense-in-depth.
                // codex 0522 F1 v1.4.106: 同步填 caller_key_id (per-call snapshot
                // 来自 conn.key_id), 让 cross-surface handler 都能识别 caller.
                let dispatch_req = IncomingRequest::builder(
                    req.conn_id,
                    req.proto_id,
                    req.serial_no,
                    req.proto_fmt_type,
                    req.body.clone(),
                )
                .with_response_commit_from(&req)
                .with_transport(req.transport)
                .with_session_generation(req.session_generation)
                .with_idempotency_key(req.idempotency_key.clone())
                .with_caller_scope(
                    allowed_acc_ids_for_resp_filter
                        .as_ref()
                        .map(|s| std::sync::Arc::new(s.clone())),
                    dispatch_caller_key_id.clone(),
                )
                .build();
                router.dispatch(conn_id, &dispatch_req).await
            }
        };

        crate::listener::reconcile_post_dispatch_connection_state(
            &connections,
            subscriptions.as_deref(),
            conn_id,
        );

        // ── Step 4: Response filter (TRD_GET_ACC_LIST 等) ────────────────────
        if let Some(body) = response_body {
            // FilterRegistry::apply(): proto_id 未注册 → 原 body 不动 (no-op).
            // 注册了 (e.g. proto 2001) → filter by allowed_acc_ids.
            let filtered =
                filter_registry.apply(proto_id_val, body, allowed_acc_ids_for_resp_filter.as_ref());
            if ApiServer::send_response(&connections, conn_id, proto_id_val, serial_no, filtered)
                .await
            {
                req.mark_response_committed();
            }
        }
    }
    if let Some(pending) = pending_init_connect {
        pending.abort();
    }
}

#[cfg(test)]
mod tests;