ek_ctp/

plugin.rs

//! # CtpPlugin —— ek-core 宿主注册的 CTP 网关插件。
//!
//! 谁发谁收:
//! - **发**:驱动输出 → `ctx.put` 上总线(纯生产者:**裸线键** `eTick.`/`eContract.`/
//!   `eOrder.`/`eTrade.`/`ePosition.`/`eAccount.`/
//!   `ePositionDone.`/`eLog`;0.4.0 D-EMIT 起裸键,网关身份归载荷 `gateway_name`;
//!   线值与 ek-oms 的 wire 模块同表同值)。
//! - **收**:`eTimer`(ek-event 秒针,2s 轮转驱动 + 行情日期刷新)与自身寻址
//!   `eReq.<名>`(订阅/发单/撤单三项服务,对齐 oms 的 ask 面形制)。
//!
//! 传输层(`CtpTransport`)是 native 桥的接缝:默认构建无传输(可测纯逻辑与
//! 请求面),native 特性(0.2.0)以官方 SDK 二进制实现该接口。
//!
//! 约定:
//! - 原稿 `CtpGateway.count`(224-233 的 2 秒 gate)在此为插件内计数器;
//! - `md.update_date` 与查询轮转同 gate(共节拍);
//! - `EVENT_POSITION_QUERY_DONE` 落线值 `ePositionDone.<gw>`(与 ek-oms 相位门同表)。

use crate::driver::{self, MdDriver, MdEvent, MdOut, TdDriver, TdEvent, TdOut};
use crate::wire::InputOrder;
use ek_core::context::PluginContext;
use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
use ek_core::plugin::Plugin;
use ek_core::section::{ApplyReport, Configurable, Field, SectionMap};
use ek_core::service::ServiceDesc;
use ek_data::cancel_request::CancelRequest;
use ek_data::contract::ContractData;
use ek_data::log::LogData;
// 只被 native 面的 `transport_log` 使用(默认构建没有它,故按特性引入)。
#[cfg(feature = "native")]
use ek_data::log::LEVEL_ERROR;
// 0.9.0 断开口落一条 INFO(默认构建也要能编译,故不挂特性)。
use ek_data::log::LEVEL_INFO;
use ek_data::order_request::OrderRequest;
use ek_event::Event;
use std::borrow::Cow;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex, OnceLock};

// ================ 总线线值(注意:与 ek-oms 的 wire 模块同表同值) ================

const W_TICK: &str = "eTick.";
const W_ORDER: &str = "eOrder.";
const W_TRADE: &str = "eTrade.";
const W_POSITION: &str = "ePosition.";
const W_ACCOUNT: &str = "eAccount.";
const W_CONTRACT: &str = "eContract.";

/// 合约流分批上限(0.11.0 第4刀 R5):**每 2000 行发一批**,收尾发尾巴。
///
/// 为什么不是"整批一次发":那会把「查询流渐进上屏」这个用户可见行为改掉(合约窗
/// 从"一段段长出来"变成"憋 8 秒一次性出现")。分批保住了渐进显形,同时把一次
/// 1.5 万行的洪峰切成 8 帧(每帧 2000 行),消费方邮箱里的帧数降两个数量级。
///
/// 线名不变(`eContract.`),载荷**新增批形**:`Vec<ContractData>`(批)。
/// **单条形 `ContractData` 继续受理**(兼容:测试注入与旧消费方),消费方两条都吃。
pub const CONTRACT_BATCH_ROWS: usize = 2000;
/// 原稿 EVENT_POSITION_QUERY_DONE(event.py:17):**裸键,无 gw 后缀**,gw 放载荷
/// ——ek-oms 侧以该裸键精确订阅(CtpPlugin emit 面),此处保持一字面。
const W_POS_DONE: &str = "ePositionDone.";
/// 资金查询完成(0.9.5):三条查询路谁收到 last 谁就推自己的结束事件,资金这一路原来没有。
const W_ACCOUNT_DONE: &str = "eAccountDone.";
/// 合约流完成线(0.10.5;**全局线,无 gw 后缀**,载荷=网关名以便多源辨源,D-CFIN;
/// 0.10.5 起导出单源常量 —— ui/station 等消费方一律改引,禁再造裸字面)。
pub const EVENT_CONTRACT_FINISH: &str = "eContractFinish";
/// 网关通用状态面后缀(0.10.5 单源导出;**`{gw}` 参数化面** —— 消费方按
/// skill §3 全名公式组 `format!("{gw}.{SVC_STATUS}")`,分隔符是语法非事实)。
pub const SVC_STATUS: &str = "status";
/// 日志线单源(ek-data::event::EVENT_LOG = "eLog",不再本地复定义)。
use ek_data::event::EVENT_LOG as W_LOG;
/// ek-event 秒针广播键(引用共享常量,不再重复定义字面值)。
/// 由 [ek_event::EVENT_TIMER] 单源提供。
pub use ek_event::EVENT_TIMER as ROUTE_TIMER;

// ================ 服务请求/回包契约(ask 载荷类型) ================

/// 订阅行情请求。
#[derive(Debug, Clone)]
pub struct SubscribeReq {
    /// 合约代码symbols(逐笔)
    pub symbols: Vec<String>,
}

/// 订阅请求回包。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeReply {
    /// 是否受理(true = 已登记/已发起)
    pub ok: bool,
    /// 说明
    pub msg: String,
}

/// 发单请求。
#[derive(Debug, Clone)]
pub struct SendOrderReq {
    /// 下单请求
    pub req: OrderRequest,
}

/// 发单回包。
#[derive(Debug, Clone, PartialEq)]
pub struct SendOrderReply {
    /// 是否受理(原稿 `""` 面 → false)
    pub ok: bool,
    /// 成功 = `"{gw}.{front}_{session}_{ref}"`(OrderData.en_orderid 同式)
    pub en_orderid: String,
    /// 拒绝时说明
    pub msg: String,
}

/// 撤单请求。
#[derive(Debug, Clone)]
pub struct CancelOrderReq {
    /// 撤单请求
    pub req: CancelRequest,
}

/// 撤单回包。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CancelOrderReply {
    /// 是否受理(true = 命令已交付)
    pub ok: bool,
    /// 说明
    pub msg: String,
}

/// 费率/保证金率查询请求(附加接口面,超出原稿)。
///
/// `kind`: "margin" = 保证金率(hedge 过滤参与查询);"commission" = 手续费率
/// (无 hedge 过滤)。查询结果经 TD 回调 → 驱动 eLog 落线(不新增事件实体)。
#[derive(Debug, Clone)]
pub struct QueryRateReq {
    /// 查询类型("margin" / "commission")
    pub kind: String,
    /// 合约代码
    pub instrument_id: String,
    /// 投机套保档(仅 margin 查询用;'1'投机)
    pub hedge_flag: String,
    /// 交易所代码(过滤;可空)
    pub exchange_id: String,
}

/// 费率查询回包(是否已交付 + RT 返回码文本)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueryRateReply {
    /// 是否受理(true = 命令已交付 native)
    pub ok: bool,
    /// 说明(交付失败时含错误代码)
    pub msg: String,
}

/// 网关配置请求(**插件自带配置面**,2026-10-17 定盘:配置归插件自身,
/// 界面/宿主手工输入经 ask 进来;插件永不读配置文件)。
///
/// 语义 = 更新账号/前置 + 复位连接守卫 + 立即发起连接(先 TD 后 MD,与
/// `connect` 同式)。`md_user`/`md_password` 留空回落 `td_user`/`td_password`
/// (SimNow 同账号常态,界面少填两格)。
///
/// Debug 手工掩码(0.10.0,BD-CFG-18):`td_password`/`md_password` 永远
/// `****` —— 0.9.x 这一路派生 `Debug`,日志/panic 报文可能带出明文口令。
#[derive(Clone, Default)]
pub struct ConfigureReq {
    /// 经纪公司代码
    pub broker_id: String,
    /// 交易前置(空 = 不连 TD;与 md_addr 至少一个非空)
    pub td_addr: String,
    /// 交易用户
    pub td_user: String,
    /// 交易密码
    pub td_password: String,
    /// 授权编码(空 = 直接登录)
    pub td_auth_code: String,
    /// 产品名称(AppID)
    pub td_app_id: String,
    /// 行情前置(空 = 不连 MD)
    pub md_addr: String,
    /// 行情用户(空回落 td_user)
    pub md_user: String,
    /// 行情密码(空回落 td_password)
    pub md_password: String,
    /// 收录的产品类白名单(CTP 单字符码,如 `["1"]`;空 = 回落默认「期货」)。
    /// 界面勾选面:期货 `'1'` / 期权 `'2'`(现货期权 `'6'` 可配但不在默认勾选里)。
    pub product_classes: Vec<String>,
}

impl std::fmt::Debug for ConfigureReq {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConfigureReq")
            .field("broker_id", &self.broker_id)
            .field("td_addr", &self.td_addr)
            // 秘面两键永远掩码(BD-CFG-18)。
            .field("td_password", &"****")
            .field("td_auth_code", &self.td_auth_code)
            .field("td_app_id", &self.td_app_id)
            .field("md_addr", &self.md_addr)
            .field("md_user", &self.md_user)
            .field("md_password", &"****")
            .field("product_classes", &self.product_classes)
            .finish()
    }
}

/// 配置回包(ok = 已应用并发起连接;false = 校验未过,msg 报字段)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigureReply {
    /// 是否受理
    pub ok: bool,
    /// 说明
    pub msg: String,
    /// 实际生效的产品类白名单(已归一化;空请求回落默认「期货」`'1'`)。
    /// 校验未过时回带**当前仍在生效**的那一份 —— 不是空表:空表会让界面误清空勾选,
    /// 而网关其实还跑在旧勾选上。
    pub product_classes: Vec<String>,
}

/// 状态查询请求(0.9.0 `{gw}.status`;无参数,单位结构体只为契约类型名占位)。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StatusReq;

/// 网关状态读数(0.9.0 新增;`{gw}.status` 的返回体 = 界面状态机的唯一事实源)。
///
/// 口径边界(消费侧文案必须照这个说,不得越界):
/// - `*_handle_up` = API 句柄已建并 registerFront,**不等于在线**(掉线时这一位仍为真);
/// - `*_logged_in` = 最近一次登录成功且未断开 —— 「在线/掉线」的唯一硬证据;
/// - 「查询断供」= `td_query_reject_streak` 达阈 **或** 距 `td_last_account_ms` 超时;
///   阈值归消费侧(ek-ui)配置,网关只给原料不裁决(它不知道界面多久刷一次)。
///   0.9.1 起合约查询被拒也进同一条连续串(消费侧要的是「查询这条路连续被拒几次」,
///   不分路);要分路看 `td_instrument_rejects` 这个分项。
/// - 计数与「最近时刻」跨重连保留:掉过几次线是历史事实,不因重连清零;
/// - 从未发生过 = `None`(界面须显示「无证据」,不得显示 0 秒前)。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct GatewayStatus {
    /// 网关注册名(派生名 `ctp_<broker>_<账号>`)
    pub gateway: String,
    /// 读数时刻(纪元毫秒;消费侧据此算「距最后一次回报多久」)
    pub sampled_ms: i64,
    // ---- 交易侧 ----
    /// TD 句柄已建(不等于在线)
    pub td_handle_up: bool,
    /// TD 已登录
    pub td_logged_in: bool,
    /// 最近一次 TD 登录失败(成功后清零)
    pub td_login_failed: bool,
    /// TD 认证通过
    pub td_auth_ok: bool,
    /// TD 认证失败
    pub td_auth_failed: bool,
    /// 登录回报带的 FrontID(0 = 从未登录成功)
    pub td_front_id: i64,
    /// 登录回报带的 SessionID
    pub td_session_id: i64,
    /// TD 断开累计次数
    pub td_disconnects: u32,
    /// 最近一次 TD 登录成功时刻
    pub td_last_login_ok_ms: Option<i64>,
    /// 最近一次资金回报时刻
    pub td_last_account_ms: Option<i64>,
    /// 最近一次持仓回报时刻
    pub td_last_position_ms: Option<i64>,
    /// 查询被拒累计次数(合约/持仓/资金三条查询路,0.9.1 起含合约)
    pub td_query_rejects: u32,
    /// 连续被拒次数(成功回报清零)
    pub td_query_reject_streak: u32,
    /// 合约查询被拒累计次数(0.9.1;`td_query_rejects` 的分项,不是另一本账)
    pub td_instrument_rejects: u32,
    /// 最近一次 TD 错误回报
    pub td_last_error: Option<String>,
    // ---- 行情侧 ----
    /// MD 句柄已建(不等于在线)
    pub md_handle_up: bool,
    /// MD 已登录
    pub md_logged_in: bool,
    /// MD 断开累计次数
    pub md_disconnects: u32,
    /// 最近一次 MD 登录成功时刻
    pub md_last_login_ok_ms: Option<i64>,
    /// 最近一条行情回报时刻
    pub md_last_tick_ms: Option<i64>,
    /// 订阅被拒累计次数
    pub md_sub_rejects: u32,
    /// 最近一次 MD 错误回报
    pub md_last_error: Option<String>,
    /// 已登记订阅数(断开不清订阅集)
    pub md_subscribed: usize,
}

/// 状态变化种类(0.11.0 第3刀):网关状态机里**值得显形的那一刻**。
///
/// 口径(与 `GatewayStatus` 的分工钉死):
/// - `GatewayStatus` = **状态面**(一直在着的读数:在不在、第几次、多久没回);
/// - `GatewayChange` = **变化面**(那一刻发生了什么的类型名);
/// - `GatewayStatusEvent` 把两者**一次到货**:变化发生时随附整份快照 → 消费者零回查。
///
/// 15 种是**穷举清单**,不是枚举可扩展的开放集:界面按 kind 去重告警、
/// 判据按 kind 断言;加一种要同时动界面文案与判据,所以新增须走文档。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GatewayChange {
    /// TD 前置已连接(会话还没起,下一步是认证/登录)
    TdConnected,
    /// TD 前置断开(会话作废,查询阶段回 Idle)
    TdDisconnected,
    /// TD 授权(认证)通过
    TdAuthOk,
    /// TD 授权失败
    TdAuthFailed,
    /// TD 登录成功(带 FrontID/SessionID)
    TdLoginOk,
    /// TD 登录失败
    TdLoginFailed,
    /// 结算单已确认(接下来问合约)
    SettlementConfirmed,
    /// 合约已到手(接下来轮询资金/持仓)
    ContractArrived,
    /// 合约查询连续未获 → 本轮查询停止(不再问资金与持仓)
    ContractQueryStopped,
    /// 查询被拒(资金/持仓/合约任一路;连续计数在状态面里)
    QueryRejected,
    /// MD 前置已连接
    MdConnected,
    /// MD 前置断开
    MdDisconnected,
    /// MD 登录成功(订阅集自会重推)
    MdLoginOk,
    /// MD 登录失败
    MdLoginFailed,
    /// MD 订阅被拒
    SubscribeRejected,
}

impl GatewayChange {
    /// 机读短名(界面去重键、判据断言都吃这个,不吃中文文案)。
    pub fn kind(&self) -> &'static str {
        match self {
            Self::TdConnected => "td-connected",
            Self::TdDisconnected => "td-disconnected",
            Self::TdAuthOk => "td-auth-ok",
            Self::TdAuthFailed => "td-auth-failed",
            Self::TdLoginOk => "td-login-ok",
            Self::TdLoginFailed => "td-login-failed",
            Self::SettlementConfirmed => "settlement-confirmed",
            Self::ContractArrived => "contract-arrived",
            Self::ContractQueryStopped => "contract-query-stopped",
            Self::QueryRejected => "query-rejected",
            Self::MdConnected => "md-connected",
            Self::MdDisconnected => "md-disconnected",
            Self::MdLoginOk => "md-login-ok",
            Self::MdLoginFailed => "md-login-failed",
            Self::SubscribeRejected => "subscribe-rejected",
        }
    }

    /// 人话文案(**生产者侧生成**,0.11.0 第3刀「二合一」):界面 `log_line(ev.describe())`
    /// 逐字照打,不再自己拼句子 —— 人看的那行与机器读的结构同源同次到货。
    ///
    /// 前缀沿用 0.9.8 的 `[链] `(界面与既有判据的观感不断档);数字一律取自
    /// 随附快照(同一份事实,不会出现"文案说第 3 次、状态列说第 2 次"这种分家)。
    pub fn describe(&self, gw: &str, st: &GatewayStatus) -> String {
        let p = driver::CHAIN_STATE_PREFIX;
        match self {
            Self::TdConnected => format!("{p}{gw} 交易前置已连接(下一步:认证)"),
            Self::TdDisconnected => format!(
                "{p}{gw} 交易前置断开(累计 {} 次;会话作废,查询阶段回 Idle)",
                st.td_disconnects
            ),
            Self::TdAuthOk => format!("{p}{gw} 授权通过(下一步:登录)"),
            Self::TdAuthFailed => format!(
                "{p}{gw} 授权失败:{}",
                st.td_last_error.as_deref().unwrap_or("无错误文本")
            ),
            Self::TdLoginOk => format!(
                "{p}{gw} 登录成功(Front {} / Session {})",
                st.td_front_id, st.td_session_id
            ),
            Self::TdLoginFailed => format!(
                "{p}{gw} 登录失败:{}",
                st.td_last_error.as_deref().unwrap_or("无错误文本")
            ),
            Self::SettlementConfirmed => {
                format!("{p}{gw} 结算单已确认(开始问合约)")
            }
            Self::ContractArrived => {
                format!("{p}{gw} 合约已到手,开始资金/持仓轮询")
            }
            Self::ContractQueryStopped => {
                format!("{p}{gw} 合约查询问满:本轮查询已停止(不再问资金与持仓)")
            }
            Self::QueryRejected => format!(
                "{p}{gw} 查询被拒(连续 {} 次 / 累计 {} 次):{}",
                st.td_query_reject_streak,
                st.td_query_rejects,
                st.td_last_error.as_deref().unwrap_or("无错误文本")
            ),
            Self::MdConnected => format!("{p}{gw} 行情前置已连接(下一步:登录)"),
            Self::MdDisconnected => format!("{p}{gw} 行情前置断开(累计 {} 次)", st.md_disconnects),
            Self::MdLoginOk => format!(
                "{p}{gw} 行情登录成功(已登记订阅 {} 项,自会重推)",
                st.md_subscribed
            ),
            Self::MdLoginFailed => format!(
                "{p}{gw} 行情登录失败:{}",
                st.md_last_error.as_deref().unwrap_or("无错误文本")
            ),
            Self::SubscribeRejected => format!(
                "{p}{gw} 行情订阅被拒(累计 {} 次):{}",
                st.md_sub_rejects,
                st.md_last_error.as_deref().unwrap_or("无错误文本")
            ),
        }
    }
}

/// 状态变化事件(0.11.0 第3刀;线名 `eStatus.<派生网关名>`)。
///
/// 一条事件 = 变化 + 整份快照(**二合一**):界面一次到货做三件事
/// (打文案 / 回写状态列 / 去重告警),**零回查**、**零轮询**。
/// 发布纪律:SDK 回调 → `with_ingress` → `TdOut::Status`/`MdOut::Status` →
/// 插件 worker `publish`;回调线程不许裸发(与数据面同一条规矩)。
#[derive(Debug, Clone)]
pub struct GatewayStatusEvent {
    /// 网关注册名(与 `eStatus.` 后缀一致)
    pub gateway: String,
    /// 变化发生(= 快照采样)时刻,纪元毫秒
    pub at_ms: i64,
    /// 变化种类
    pub change: GatewayChange,
    /// 变化发生那一刻的整份状态快照
    pub status: Box<GatewayStatus>,
}

impl GatewayStatusEvent {
    /// 人话文案(逐字给界面打日志环用;文案由生产者生成)。
    pub fn describe(&self) -> String {
        self.change.describe(&self.gateway, &self.status)
    }
}

/// 状态变化线的前缀(线名 = `eStatus.<派生网关名>`,与 `eReq.` 同式后缀型)。
pub const EVENT_STATUS_PREFIX: &str = "eStatus.";

/// 主动断开请求(0.9.0 `{gw}.disconnect`;裁定 1「断开 = 停会话,实例与订阅保留」)。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct DisconnectReq;

/// 断开回带(ok = 会话守卫已复位;msg 说明放掉了什么、保留了什么)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DisconnectReply {
    /// 是否执行
    pub ok: bool,
    /// 说明
    pub msg: String,
}

// ================ 传输层接缝(native 桥实现/dev 以测试桩实现) ================

/// 传输出口:驱动动作落到真实 CTP 会话(0.2.0 native 才有官方实现)。
pub trait CtpTransport: Send + Sync {
    /// 创建行情 API 并 registerFront/init。
    fn md_connect(&self, addr: &str, flow_path: &str);
    /// 行情登录。
    fn md_login(&self, user: &str, password: &str, broker: &str);
    /// 行情订阅。
    fn md_subscribe(&self, symbols: &[String]);
    /// 创建交易 API 并 registerFront/init。
    fn td_connect(&self, addr: &str, flow_path: &str);
    /// 认证。
    fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str);
    /// 登录。
    fn td_login(&self, user: &str, password: &str, broker: &str);
    /// 结算单确认。
    fn td_settlement_confirm(&self, broker: &str, user: &str);
    /// 合约查询;返回 `reqQryInstrument` 的 rt 返回码(0.9.3)。原稿 519-533 在这里
    /// `while rc != 0 { sleep(1); 重发 }` —— 阻塞的是 CTP 的回调线程,而且每秒一发
    /// 正好把轮转查询挤出台阶。现在**不排重发**:下一问的位置由 `TdDriver::on_timer` 那个
    /// 协调点的周期给出(最多 3 次,任一合约信息到手即退出),被拒进同一条查询连续串。
    fn td_qry_instrument(&self) -> i32;
    /// 资金查询;返回 `reqQryTradingAccount` 的 rt 返回码(0.9.2)。原稿丢弃该返回码(876),
    /// 于是发送侧被拒在计数与状态口里全隐形;非 0 现由插件喂进同一条查询连续串
    /// (`TdDriver::note_query_send_failure`)。
    fn td_qry_account(&self, broker: &str, user: &str) -> i32;
    /// 持仓查询;返回 `reqQryInvestorPosition` 的 rt 返回码(0.9.2,同上,原稿 886)。
    fn td_qry_position(&self, broker: &str, user: &str) -> i32;
    /// 插单;返回 `reqOrderInsert` 的 rt 返回码(0 = 受理;非 0 = 原稿 846-848 的
    /// 失败面:记「委托请求发送失败,错误代码:n」且**不回 SUBMITTING 单**)。
    fn td_insert(&self, order: &InputOrder) -> i32;
    /// 撤单;返回 `reqOrderAction` 的 rt 返回码(原稿不检,原样透传给 native 自定)。
    fn td_cancel(&self, action: &crate::wire::InputOrderAction) -> i32;
    /// 合约保证金率查询(附加接口面;返回 `ReqQryInstrumentMarginRate` rt 返回码)。
    fn td_qry_margin_rate(&self, query: &crate::wire::QryInstrumentMarginRate) -> i32;
    /// 合约手续费率查询(附加接口面;返回 `ReqQryInstrumentCommissionRate` rt 返回码)。
    fn td_qry_commission_rate(&self, query: &crate::wire::QryInstrumentCommissionRate) -> i32;

    // 交易状态没有查询面(SDK 6.7.7 无 ReqQryInstrumentStatus):该件纯被动,
    // 由公有流推送经 TD 回调直达驱动,传输层不提供主动入口。

    /// 关闭原生会话(宿主停插件时调用)。
    ///
    /// 默认空实现 = 无资源可放。官方实现里这一步是必需的:`Release()` 之后 SDK
    /// **不再回调**,回调入口(ctx)才允许被放下 —— 顺序反了就是 use-after-free。
    fn shutdown(&self) {}

    /// 断开交易会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
    /// configure 强制重连面(0.5.0)用:旧登录态先断,新账号/密码才有显形机会。
    fn td_release(&self) {}
    /// 断开行情会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
    fn md_release(&self) {}
}

// ================ 网关配置 ================

/// 网关配置(对齐原稿 connect setting 141-171;地址补前缀由使用方完成——
/// 原稿在 connect 里对缺 "tcp://"/"ssl://" 的地址自动补 "tcp://",
/// 本 crate 要求调用方先行规整,保持驱动数据纯净)。
///
/// Debug 手工掩码(0.10.0,BD-CFG-18):`td_password`/`md_password` 永远 `****`。
#[derive(Clone, Default)]
pub struct CtpConfig {
    /// 网关/插件注册名(唯一;默认 "ctp")
    pub gateway_name: String,
    /// 经纪公司代码
    pub broker_id: String,
    /// 交易前置
    pub td_addr: String,
    /// 交易用户
    pub td_user: String,
    /// 交易密码
    pub td_password: String,
    /// 授权编码(空 = 直接登录,原稿 448-451)
    pub td_auth_code: String,
    /// 产品名称(AppID)
    pub td_app_id: String,
    /// 行情前置
    pub md_addr: String,
    /// 行情用户
    pub md_user: String,
    /// 行情密码
    pub md_password: String,
    /// 收录的产品类白名单(D-PC-FILTER;空 = 回落默认「期货」`'1'`)。
    /// 查询侧不过滤,收录侧按它严格丢弃(不进共享表、不上抛 `eContract`)。
    pub product_classes: crate::map::ProductClasses,
}

impl std::fmt::Debug for CtpConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CtpConfig")
            .field("gateway_name", &self.gateway_name)
            .field("broker_id", &self.broker_id)
            .field("td_addr", &self.td_addr)
            .field("td_user", &self.td_user)
            // 秘面两键永远掩码(BD-CFG-18)。
            .field("td_password", &"****")
            .field("td_auth_code", &self.td_auth_code)
            .field("td_app_id", &self.td_app_id)
            .field("md_addr", &self.md_addr)
            .field("md_user", &self.md_user)
            .field("md_password", &"****")
            .field("product_classes", &self.product_classes)
            .finish()
    }
}

impl CtpConfig {
    /// 缺省名 "ctp" 的兜底。
    pub fn gateway_name_or_default(&self) -> String {
        if self.gateway_name.is_empty() {
            "ctp".to_string()
        } else {
            self.gateway_name.clone()
        }
    }
}

/// 凭据档位快照行(0.10.0;`{gw}.credentials_state` ask 回包的元素;运维窗用)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialEntry {
    /// 字段名(`td_password`/`md_password`)。
    pub field: String,
    /// 档位:`REFERENCED`(引用已登记)/`RESOLVED`(内存已解密,连接可用)/
    /// `ABSENT`(待手配;连接面按原有空值守卫跳过)。
    pub state: String,
    /// 引用的秘键(空 = 无)。
    pub ref_key: String,
}

/// 凭据状态请求(无载荷)。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CredentialStateReq;

/// 多账号面的注册名派生(**界面唯一入口**;配置面不存网关名)。
///
/// 规则 `ctp_<broker>_<账号>`,例 `ctp_9999_278153`。为什么必须是这个形状:
///
/// - 宿主名字规则 `[A-Za-z][A-Za-z0-9_-]{0,31}`(ek-core `valid_plugin_name`)
///   ——**不能含点号**,总长 ≤32;
/// - `en_orderid = "{gw}.{front}_{session}_{ref}"` 的反解用 `split_once('.')`
///   (ek-converter `holding.rs`)——gw 里带点号会把反解串成 `"ctp"`;
/// - 派生而非手填:手填会重名/打错;同 broker+账号 重复添加时派生名相同,
///   由宿主 `Duplicate` 直接挡下(不需要额外的账号注册表)。
///
/// 返回 `None` = 输入不可用(broker 或账号为空、含 `[0-9A-Za-z]` 之外的字符、
/// 或派生名超 32 字符)。调用方(界面)须拒绝并说明,**不得静默替换**。
///
/// 0.10.4 边角收紧:字段**统一小写归一** —— 唯一键即 `ctp+broker+账号`
/// 字符串恒等(大小写变体 = 同键 = 添加面重复拦截)。
#[must_use]
pub fn derived_gateway_name(broker_id: &str, account: &str) -> Option<String> {
    let broker = sanitize_id_field(broker_id)?;
    let account = sanitize_id_field(account)?;
    let name = format!("ctp_{broker}_{account}");
    // 宿主上限 32(`[A-Za-z][A-Za-z0-9_-]{0,31}`);超限界面提前拦。
    if name.len() > 32 {
        return None;
    }
    Some(name)
}

/// 账号位清洗:去首尾空白,只允许 `[0-9A-Za-z]`,**统一小写归一**
/// (0.10.4 边角收紧:"I130"与"i130"是同一键 —— 变体 = 重复,不是两家;
/// 与流文件目录/秘库/配置 ns 共用这一处大小写真值,别处不再各自改写)。
/// 空或含非法字符 → `None`。
fn sanitize_id_field(raw: &str) -> Option<String> {
    let trimmed = raw.trim();
    if trimmed.is_empty() || !trimmed.chars().all(|c| c.is_ascii_alphanumeric()) {
        return None;
    }
    Some(trimmed.to_ascii_lowercase())
}

/// 注册名形状钉(0.10.4):命名实例只收**派生名形状** —— `ctp_<broker>_<账号>`,
/// 两段非空且仅 `[0-9a-z]`,总长 ≤32,小写。宿主工厂面在收 svctl 命名请求时
/// 以本口前置闸(任意名/变体名/超长名在源头拒,不再给出孤儿档案开口);
/// 界面添加面走 `derived_gateway_name`,同样只产本形状。
#[must_use]
pub fn is_derived_gateway_name(name: &str) -> bool {
    let Some(rest) = name.strip_prefix("ctp_") else {
        return false;
    };
    if name.len() > 32 {
        return false;
    }
    let ok = |s: &str| {
        !s.is_empty()
            && s.chars()
                .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
    };
    match rest.split_once('_') {
        Some((broker, account)) => ok(broker) && ok(account),
        None => false,
    }
}

// ================ 插件 ================

/// 锁克隆辅助(poison 恢复语义与插件既有 `unwrap_or_else(into_inner)` 同式)。
fn lock_clone<T: Clone>(l: &std::sync::Mutex<T>) -> T {
    match l.lock() {
        Ok(g) => g.clone(),
        Err(e) => e.into_inner().clone(),
    }
}

/// CTP 网关插件(纯生产者 + 自服务)。
pub struct CtpPlugin {
    gw: String,
    /// 配置托底段 ns(0.10.0:缺省实例 `ctp`,命名实例 `ctp.<网关注册名>`)。
    ns: String,
    /// 配置段已见最大 rev(dump 单调递增的底;BD-CFG-4 语义)。
    cfg_rev: AtomicU32,
    /// 已回灌的秘面引用行((字段名, 秘键)) —— 解密在 on_start(worker 面,
    /// BD-CFG-16);apply_config 自身禁取秘。
    sec_refs: Mutex<Vec<(String, String)>>,
    /// 当前网关配置(connect 空地址守卫与费率查询的单一事实源)。
    cfg: Mutex<CtpConfig>,
    /// 经纪公司代码(费率查询的 broker_id;configure 可换)。
    broker: Mutex<String>,
    /// 投资者代码(费率查询的 investor_id;configure 可换)。
    user: Mutex<String>,
    md: Mutex<MdDriver>,
    td: Mutex<TdDriver>,
    /// 2s gate 计数器(原稿 self.count)
    timer_count: Mutex<u32>,
    /// 传输层(默认构建 None;native/测试注入)
    trans: Option<Box<dyn CtpTransport>>,
    ctx: OnceLock<PluginContext>,
    /// 「未启用传输」提示只发一次
    trans_warned: Mutex<bool>,
    /// 直写托底兜底审计行只发一次(BD-CFG-11)
    save_fallback: Mutex<bool>,
    /// 合约流分批缓冲(0.11.0 第4刀 R5):攒到 `CONTRACT_BATCH_ROWS` 就发一批,
    /// 收尾(`ContractFinish`)/会话复位时把不满一批的尾巴发掉。
    contract_batch: Mutex<Vec<ContractData>>,
}

impl CtpPlugin {
    /// 纯逻辑构造(native 尚未接传输)。
    pub fn new(cfg: CtpConfig) -> Self {
        let gw = cfg.gateway_name_or_default();
        Self::with_parts(cfg, gw, None, None)
    }

    /// 注入传输层(native 桥或测试桩)。
    pub fn with_transport(cfg: CtpConfig, trans: Box<dyn CtpTransport>) -> Self {
        let gw = cfg.gateway_name_or_default();
        Self::with_parts(cfg, gw, Some(trans), None)
    }

    /// 注入传输层 + 驱动单调钟(0.9.6,**测试接缝**)。
    ///
    /// 协调点的护栏按**真实时间**放行:真钟下测试里连续调 `on_timer` 之间 elapsed≈0,会被
    /// 全挡。驱动级有 `TdDriver::with_clock`,但插件级钉(`plugin_full_bus_roundtrip` 那三条)
    /// 走的是插件内部自建的驱动 —— 没有这个接缝,它们只能改成真 `sleep(1s)`:网关测试套从
    /// 1 秒拖成 6 秒,而且变成时间依赖(以后会偶发红)。
    pub fn with_transport_and_clock(
        cfg: CtpConfig,
        trans: Box<dyn CtpTransport>,
        now_ms: Arc<dyn Fn() -> u64 + Send + Sync>,
    ) -> Self {
        let gw = cfg.gateway_name_or_default();
        Self::with_parts(cfg, gw, Some(trans), Some(now_ms))
    }

    fn with_parts(
        cfg: CtpConfig,
        gw: String,
        trans: Option<Box<dyn CtpTransport>>,
        clock: Option<Arc<dyn Fn() -> u64 + Send + Sync>>,
    ) -> Self {
        let contracts = driver::contracts_empty();
        let cfg_broker = cfg.broker_id.clone();
        let cfg_user = cfg.td_user.clone();
        let md_cfg = driver::MdConfig {
            addr: cfg.md_addr.clone(),
            user: cfg.md_user.clone(),
            password: cfg.md_password.clone(),
            broker: cfg.broker_id.clone(),
        };
        let td_cfg = driver::TdConfig {
            addr: cfg.td_addr.clone(),
            user: cfg.td_user.clone(),
            password: cfg.td_password.clone(),
            broker: cfg.broker_id.clone(),
            auth_code: cfg.td_auth_code.clone(),
            app_id: cfg.td_app_id.clone(),
            product_classes: cfg.product_classes.clone(),
        };
        // 原稿 md.current_date 由 datetime.now() 起手;这里初值 = 启动时的上海日期
        let md = Mutex::new(MdDriver::new(
            &gw,
            md_cfg,
            contracts.clone(),
            crate::map::shanghai_now_date(),
        ));
        let td = Mutex::new(match clock {
            Some(f) => TdDriver::with_clock(&gw, td_cfg, contracts.clone(), f),
            None => TdDriver::new(&gw, td_cfg, contracts.clone()),
        });
        let ns = if gw == "ctp" {
            // 缺省实例 ns = "ctp"(与工厂名一致)。
            "ctp".to_owned()
        } else {
            // 命名实例 ns = `ctp.<网关注册名>`(多账号天然分档)。
            format!("ctp.{gw}")
        };
        Self {
            gw,
            ns,
            cfg_rev: AtomicU32::new(0),
            sec_refs: Mutex::new(Vec::new()),
            cfg: Mutex::new(cfg),
            broker: Mutex::new(cfg_broker),
            user: Mutex::new(cfg_user),
            md,
            td,
            timer_count: Mutex::new(0),
            trans,
            ctx: OnceLock::new(),
            trans_warned: Mutex::new(false),
            save_fallback: Mutex::new(false),
            contract_batch: Mutex::new(Vec::with_capacity(CONTRACT_BATCH_ROWS)),
        }
    }

    /// native 构造:装配官方 SDK 传输层并完成回调接线(见 `native::transport` 文档)。
    ///
    /// 两相接线:① 先建回调入口 + 传输层(传输层持有 `Arc<Ingress>`);② 插件建好后
    /// 回填 `Weak<CtpPlugin>` —— 必须用弱引用断环,否则 插件 ↔ 传输层 互相持有,
    /// 两者永不析构(回调也不会停)。
    ///
    /// `flow_root` = CTP 流文件落盘根目录(D-FLOWPATH)。构造本身**不碰磁盘**:
    /// 目录在真正连接时才按 `<flow_root>/<网关名小写>/{Md,Td}` 创建。
    #[cfg(feature = "native")]
    pub fn with_native(cfg: CtpConfig, flow_root: impl Into<std::path::PathBuf>) -> Arc<Self> {
        let (trans, ingress) = crate::native::transport::native_parts(flow_root);
        let plugin = Arc::new(Self::with_transport(cfg, trans));
        crate::native::transport::attach_plugin(&ingress, &plugin);
        plugin
    }

    /// 连接(原稿 `CtpGateway.connect` 162-190:先 TD 后 MD;`init_query` 只是挂秒针)。
    ///
    /// **连接触发点住在本插件里**(不放到核心/公共层,每个插件自持自己的连接):
    /// 宿主 `start()` 会经 [`Plugin::on_start`] 调它一次;要显式控制(先配后连、
    /// 断线后重连)也可以直接调。幂等由驱动守卫裁决 —— 原稿 MD 重复调用静默,
    /// TD 重复调用转认证(776-777),两者都在 `MdDriver::connect`/`TdDriver::connect`。
    ///
    /// 原稿末尾的 `init_query` 只是把查询轮转注册到 `EVENT_TIMER`(237-240);
    /// 本 crate 的对应面是宿主 eTimer → [`CtpPlugin::on_timer`],这里无需动作。
    pub fn connect(&self) {
        // 空地址守卫(0.3.0):td/md 前置都空 → 不瞎连(on_start 空配置形态),
        // 等 `{gw}.configure` 手工注入后再发起(插件自带配置面,定盘口径)。
        let (td_empty, md_empty) = {
            let c = self.cfg.lock().unwrap();
            (c.td_addr.is_empty(), c.md_addr.is_empty())
        };
        if td_empty && md_empty {
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: "未配置前置地址,跳过连接(等待 configure)".to_string(),
                level: 20,
            });
            return;
        }
        // 先 TD 后 MD:与原稿 connect 的调用次序一致。
        // 锁在各自语句结束时释放 —— emit 可能落到会阻塞的原生调用(如合约查询的
        // 流控重发),绝不能在持驱动锁时做。
        let td_outs = if td_empty {
            Vec::new()
        } else {
            self.td.lock().unwrap().connect()
        };
        let md_outs = if md_empty {
            Vec::new()
        } else {
            self.md.lock().unwrap().connect()
        };
        for o in td_outs {
            self.emit_td(o);
        }
        for o in md_outs {
            self.emit_md(o);
        }
    }

    /// 传输层级失败上报:native 桥在回调面之外没有别的出口。
    ///
    /// 只用在本插件/原生层自己的故障上(建不出 API、建不出流目录)。原稿没有对应面
    /// ——那里失败会直接抛异常,进程级可见;Rust 侧不能抛,所以落一条 ERROR 日志。
    /// **请求返回码不在此列**:原稿除下单外一律丢弃返回码,那些日志会是原稿没有的噪音。
    #[cfg(feature = "native")]
    pub(crate) fn transport_log(&self, msg: String) {
        self.put_log(LogData {
            gateway_name: self.gw.clone(),
            msg,
            level: LEVEL_ERROR,
        });
    }

    /// 原生适配入口(native 桥的回调直接调这四个):MD 回报步进 + 总线上抛。
    pub fn on_md_event(&self, ev: MdEvent) {
        let outs = self.md.lock().unwrap().step(ev);
        for o in outs {
            self.emit_md(o);
        }
    }

    /// 原生适配入口(native 桥的回调直接调这四个):TD 回报步进 + 总线上抛。
    pub fn on_td_event(&self, ev: TdEvent) {
        let outs = self.td.lock().unwrap().step(ev);
        for o in outs {
            self.emit_td(o);
        }
    }

    /// eTimer 一拍(原稿 gateway.process_timer_event)。
    ///
    /// 0.9.4 分工:**每一拍都问驱动** —— [`TdDriver::on_timer`] 是三条查询路(合约重发 /
    /// 资金轮 / 持仓轮)的唯一协调点,它自己按拍决定这一拍发哪一条、最多发一条。
    /// 插件这里保留原稿的 2 秒 gate 只给行情日期刷新(`md.update_date` 走的是另一条
    /// 连接,不占交易前置的查询名额)。
    pub fn on_timer(&self) {
        let outs = self.td.lock().unwrap().on_timer();
        for o in outs {
            self.emit_td(o);
        }
        let gate_pass = {
            let mut c = self.timer_count.lock().unwrap();
            *c += 1;
            if *c < 2 {
                false
            } else {
                *c = 0;
                true
            }
        };
        if !gate_pass {
            return;
        }
        // 原稿 231-235 次序:先查询,再 md_api.update_date
        let date = crate::map::shanghai_now_date();
        self.md.lock().unwrap().update_date(date);
    }

    fn emit_md(&self, o: MdOut) {
        match o {
            MdOut::Log(l) => self.put_log(l),
            // 0.11.0 第3刀:状态变化 → eStatus.<gw>(worker 侧 publish,回调线程不裸发)
            MdOut::Status(ch) => self.publish_status(ch),
            MdOut::Tick(t) => self.put_data(W_TICK, t),
            MdOut::ReqConnect { addr, flow_path } => {
                if let Some(t) = &self.trans {
                    t.md_connect(&addr, &flow_path);
                } else {
                    self.warn_no_transport();
                }
            }
            MdOut::ReqLogin {
                user,
                password,
                broker,
            } => {
                if let Some(t) = &self.trans {
                    t.md_login(&user, &password, &broker);
                } else {
                    self.warn_no_transport();
                }
            }
            MdOut::ReqSubscribe(symbols) => {
                if let Some(t) = &self.trans {
                    t.md_subscribe(&symbols);
                } else {
                    self.warn_no_transport();
                }
            }
        }
    }

    fn emit_td(&self, o: TdOut) {
        match o {
            TdOut::Log(l) => self.put_log(l),
            // 0.11.0 第3刀:状态变化 → eStatus.<gw>(随附整份状态快照,消费者零回查)
            TdOut::Status(ch) => self.publish_status(ch),
            TdOut::Contract(c) => self.emit_contract(c),
            TdOut::Order(o) => self.publish_enriched(W_ORDER, o),
            TdOut::Trade(t) => self.publish_enriched(W_TRADE, t),
            TdOut::Position(p) => self.publish_enriched(W_POSITION, p),
            TdOut::Account(a) => self.put_data(W_ACCOUNT, a),
            TdOut::PositionQueryDone(gw) => {
                // 原稿 593:on_event(EVENT_POSITION_QUERY_DONE, self.gateway_name)
                // → 裸键 "ePositionDone." + 载荷 gw(ek-oms 以此精确订阅)
                if let Some(ctx) = self.ctx.get() {
                    let _ = ctx.put(Event::with_data(W_POS_DONE, Arc::new(gw)));
                }
            }
            TdOut::AccountQueryDone(gw) => {
                // 0.9.5:与持仓那一路同式 —— 裸键 "eAccountDone." + 载荷 gw。
                if let Some(ctx) = self.ctx.get() {
                    let _ = ctx.put(Event::with_data(W_ACCOUNT_DONE, Arc::new(gw)));
                }
            }
            // 原稿 EVENT_CONTRACT_FINISH(0.10.5 起本 crate 导出常量):全局线无 gw 后缀;
            // 载荷放网关名(多网关辨源,D-CFIN)。
            TdOut::ContractFinish(gw) => {
                // 0.11.0 第4刀:先把不满一批的尾巴发掉,再收尾(收尾语义不变)
                self.flush_contract();
                if let Some(ctx) = self.ctx.get() {
                    let _ = ctx.put(Event::with_data(EVENT_CONTRACT_FINISH, Arc::new(gw)));
                }
            }
            TdOut::ReqConnect { addr, flow_path } => {
                if let Some(t) = &self.trans {
                    t.td_connect(&addr, &flow_path);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqAuthenticate {
                user,
                broker,
                auth_code,
                app_id,
            } => {
                if let Some(t) = &self.trans {
                    t.td_auth(&user, &broker, &auth_code, &app_id);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqLogin {
                user,
                password,
                broker,
            } => {
                if let Some(t) = &self.trans {
                    t.td_login(&user, &password, &broker);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqSettlementConfirm { broker, user } => {
                if let Some(t) = &self.trans {
                    t.td_settlement_confirm(&broker, &user);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryInstrument => {
                if let Some(t) = &self.trans {
                    let n = t.td_qry_instrument();
                    // 0.9.3:返回码不再丢(原稿在这里阻塞重发)。被拒 → 进连续串 +
                    // 进合约分项 + 排下一拍重发,全在驱动那一个方法里。
                    for l in self.td.lock().unwrap().note_instrument_send_failure(n) {
                        self.put_log(l);
                    }
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryAccount { broker, user } => {
                if let Some(t) = &self.trans {
                    let n = t.td_qry_account(&broker, &user);
                    for l in self.td.lock().unwrap().note_query_send_failure("资金", n) {
                        self.put_log(l);
                    }
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::ReqQryPosition { broker, user } => {
                if let Some(t) = &self.trans {
                    let n = t.td_qry_position(&broker, &user);
                    for l in self.td.lock().unwrap().note_query_send_failure("持仓", n) {
                        self.put_log(l);
                    }
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::Insert(o) => {
                if let Some(t) = &self.trans {
                    t.td_insert(&o);
                } else {
                    self.warn_no_transport();
                }
            }
            TdOut::Cancel(a) => {
                if let Some(t) = &self.trans {
                    t.td_cancel(&a);
                } else {
                    self.warn_no_transport();
                }
            }
        }
    }

    /// 业务日志统一出口(LogData 挂 "eLog" 固定键;ctx 文档同款)。
    fn put_log(&self, l: LogData) {
        if let Some(ctx) = self.ctx.get() {
            let _ = ctx.put(Event::with_data(W_LOG, Arc::new(l)));
        }
    }

    /// 数据上抛(**裸线键 `eXxx.` + 载荷**;0.4.0 集成修正)。
    ///
    /// 原稿 emit 面带 gw 后缀(`eOrder.ctp`),但 ek 总线派发是精确匹配
    /// (ek-event register 精确键,ek-core 照 routes 原样注册),而消费侧
    /// dm/oms 注册的是裸前缀(`eTick.`/`eOrder.`)——带后缀的行**没有任何
    /// 订阅者**,整条 ctp 数据流在装配里被静默丢弃(T5 实连揭出)。网关身份
    /// 本就在载荷 `gateway_name` 字段里,消费侧按它分流足够;sim 侧同款裸键
    /// 先例(ui_sim 全绿)钉死此契约。见 DESIGN D-EMIT。
    fn put_data<T: Send + Sync + 'static>(&self, prefix: &str, payload: T) {
        if let Some(ctx) = self.ctx.get() {
            let _ = ctx.put(Event::with_data(prefix, Arc::new(payload)));
        }
    }

    /// 状态变化上抛(0.11.0 第3刀;线名 = 裸前缀 `eStatus.`,网关名在载荷里)。
    ///
    /// 为什么不是 `eStatus.<gw>` 带后缀:本总线是**精确匹配**
    /// (`ek_event::HandlerRegistry::snapshot` 只查 `typed.get(event_type)`),
    /// 带后缀的线只有"知道完整网关名的订阅者"收得到,而消费侧(routes 在插件启动时
    /// 一次性注册)没法为将来添加的账户预注册线。数据面早就是同一条契约
    /// (D-EMIT:裸线键 + 载荷带 `gateway_name`),状态面照它走,消费侧按载荷分流。
    /// 合约行入账并发批(0.11.0 第4刀 R5)。
    ///
    /// 攒满 [`CONTRACT_BATCH_ROWS`] 就发一帧 `Vec<ContractData>`;不满一批**不发**
    /// (等下一行/收尾),所以一次 1.5 万行的查询只产生 8 帧而不是 1.5 万帧。
    fn emit_contract(&self, c: ContractData) {
        let mut batch = self.contract_batch.lock().unwrap();
        batch.push(c);
        if batch.len() >= CONTRACT_BATCH_ROWS {
            let payload = std::mem::take(&mut *batch);
            drop(batch);
            self.put_data(W_CONTRACT, payload);
        }
    }

    /// 发掉不满一批的尾巴(收尾 / 会话复位 / 摘插件时调用)。空则什么都不做。
    pub fn flush_contract(&self) {
        let payload = {
            let mut batch = self.contract_batch.lock().unwrap();
            std::mem::take(&mut *batch)
        };
        if !payload.is_empty() {
            self.put_data(W_CONTRACT, payload);
        }
    }

    /// 发一行并补名称(0.11.0 第5刀 R6)。
    ///
    /// 网关侧富化:行自带名字就用它的(柜台哪天给名字了就不覆盖),空了才查共享
    /// 合约表;**查不到照发**(红线①:绝不因为缺名字丢行)。
    fn publish_enriched<T>(&self, line: &str, mut row: T)
    where
        T: EnrichName + Send + Sync + 'static,
    {
        if row.name_mut().is_empty() {
            let symbol = row.symbol().to_string();
            if let Some(n) = self.td.lock().unwrap().contract_name(&symbol) {
                *row.name_mut() = n;
            }
        }
        self.put_data(line, row);
    }

    fn publish_status(&self, change: GatewayChange) {
        let st = self.status();
        self.put_data(
            EVENT_STATUS_PREFIX,
            GatewayStatusEvent {
                gateway: self.gw.clone(),
                at_ms: st.sampled_ms,
                change,
                status: Box::new(st),
            },
        );
    }

    /// 无传输时的静默风告(只发一次;info 级防刷屏)。
    fn warn_no_transport(&self) {
        let mut warned = self.trans_warned.lock().unwrap();
        if !*warned {
            *warned = true;
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: "传输层未接入(默认构建):连接/查询/发送动作已被吸收".to_string(),
                level: 20,
            });
        }
    }

    /// 服务面:`eReq.<gw>` 分派(arm 范式与 oms 同)。
    fn on_request(&self, event: &Event) {
        let Some(env) = event.data.downcast_ref::<Envelope>() else {
            return;
        };
        macro_rules! arm {
            ($ty:ty, $res:ty, $body:expr) => {
                if let Some(req) = env.payload.downcast_ref::<$ty>() {
                    let res: $res = $body(req);
                    let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
                    return;
                }
            };
        }
        arm!(SubscribeReq, SubscribeReply, |r: &SubscribeReq| self
            .do_subscribe(r));
        arm!(SendOrderReq, SendOrderReply, |r: &SendOrderReq| self
            .do_send_order(r));
        arm!(CancelOrderReq, CancelOrderReply, |r: &CancelOrderReq| self
            .do_cancel_order(r));
        arm!(QueryRateReq, QueryRateReply, |r: &QueryRateReq| self
            .do_query_rate(r));
        arm!(ConfigureReq, ConfigureReply, |r: &ConfigureReq| self
            .do_configure(r));
        arm!(StatusReq, GatewayStatus, |r: &StatusReq| self.do_status(r));
        arm!(DisconnectReq, DisconnectReply, |r: &DisconnectReq| self
            .do_disconnect(r));
        // 0.10.0(BD-CFG 环节7):凭据档位只读口(运维窗「已存/未存」)。
        arm!(
            CredentialStateReq,
            Vec<CredentialEntry>,
            |_: &CredentialStateReq| self.credentials_state()
        );
        // 载荷不是任何已声明契约 → 协议错误(宿主既有语义)
        let _ = env
            .reply
            .send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
    }

    fn do_subscribe(&self, r: &SubscribeReq) -> SubscribeReply {
        let mut outs = Vec::new();
        {
            let mut md = self.md.lock().unwrap();
            for s in &r.symbols {
                outs.extend(md.subscribe(s));
            }
        }
        for o in outs {
            self.emit_md(o);
        }
        SubscribeReply {
            ok: true,
            msg: String::new(),
        }
    }

    fn do_send_order(&self, r: &SendOrderReq) -> SendOrderReply {
        let outcome = {
            let mut td = self.td.lock().unwrap();
            td.send_order(&r.req)
        };
        if let Some(log) = &outcome.error {
            self.put_log(log.clone());
        }
        // 原稿 844-848:插单命令的 rt 返回码由传输层裁决;n != 0 →
        // 日志「委托请求发送失败,错误代码:n」且回 ""(不回 SUBMITTING 单)。
        // 其余命令面照常下发。
        let mut submit_order: Option<ek_data::order::OrderData> = None;
        let mut send_failed: Option<String> = None;
        for o in outcome.outs {
            match o {
                TdOut::Insert(input) => match &self.trans {
                    Some(t) => {
                        let n = t.td_insert(&input);
                        if n != 0 {
                            send_failed = Some(format!("委托请求发送失败,错误代码:{n}"));
                        }
                    }
                    None => self.warn_no_transport(),
                },
                TdOut::Order(od) => submit_order = Some(od),
                other => self.emit_td(other),
            }
        }
        if let Some(msg) = send_failed {
            let log = LogData {
                gateway_name: self.gw.clone(),
                msg: msg.clone(),
                level: 20,
            };
            self.put_log(log);
            return SendOrderReply {
                ok: false,
                en_orderid: String::new(),
                msg,
            };
        }
        if let Some(od) = submit_order {
            self.emit_td(TdOut::Order(od));
        }
        if outcome.en_orderid.is_empty() {
            let msg = outcome.error.map(|l| l.msg).unwrap_or_default();
            SendOrderReply {
                ok: false,
                en_orderid: String::new(),
                msg,
            }
        } else if self.trans.is_none() {
            self.warn_no_transport();
            SendOrderReply {
                ok: false,
                en_orderid: outcome.en_orderid,
                msg: "传输层未接入(默认构建)".to_string(),
            }
        } else {
            SendOrderReply {
                ok: true,
                en_orderid: outcome.en_orderid,
                msg: String::new(),
            }
        }
    }

    /// 费率查询实现(附加接口面)。margin → `td_qry_margin_rate`,
    /// commission → `td_qry_commission_rate`;status 无 Req,不在此线
    /// (纯被动,官文见 DESIGN)。RT 非零命中原稿式发送失败日志并回值。
    fn do_query_rate(&self, r: &QueryRateReq) -> QueryRateReply {
        let broker = self.broker.lock().unwrap().clone();
        let user = self.user.lock().unwrap().clone();
        let is_margin = r.kind == "margin";
        let is_commission = r.kind == "commission";
        if !is_margin && !is_commission {
            self.put_log(LogData {
                gateway_name: self.gw.clone(),
                msg: format!("费率查询类型未知:{}", r.kind),
                level: 20,
            });
            return QueryRateReply {
                ok: false,
                msg: format!("费率查询类型未知:{}", r.kind),
            };
        }
        if let Some(t) = &self.trans {
            let n = if is_margin {
                let query = crate::wire::QryInstrumentMarginRate {
                    broker_id: broker.clone(),
                    investor_id: user.clone(),
                    hedge_flag: r.hedge_flag.clone(),
                    exchange_id: r.exchange_id.clone(),
                    invest_unit_id: String::new(),
                    instrument_id: r.instrument_id.clone(),
                };
                t.td_qry_margin_rate(&query)
            } else {
                let query = crate::wire::QryInstrumentCommissionRate {
                    broker_id: broker.clone(),
                    investor_id: user.clone(),
                    exchange_id: r.exchange_id.clone(),
                    invest_unit_id: String::new(),
                    instrument_id: r.instrument_id.clone(),
                };
                t.td_qry_commission_rate(&query)
            };
            let label = if is_margin {
                "保证金率"
            } else {
                "手续费率"
            };
            if n != 0 {
                // 0.9.6(用户定口径):费率**不自动查询、不自动重发** —— 这一口只负责把 CTP
                // 两个费率查询封装好,供别的插件按需调用,谁调谁负责自己的节律。
                // 原稿示例第 11 步那套「RT != 0 就隔 1.05s 重发一次」在这里有两个害处:
                // ① 它睡在**总线唯一的分发线程**上(ek-event engine.rs `run_dispatch` 单线程),
                //    一次被拒 = 全进程所有插件的下单/撤单/状态回包停 1.05 秒,并把排队的
                //    eTimer 拍挤到一起处理;② 它自己就是第二发,与协调点的周期查询落在同一秒
                //    里正好凑成 `-3`。
                // **不进 query_rejects / query_reject_streak**(用户定):那个数是网关自己那条
                // 查询路的体检数,ek-ui 拿它数「查询这条路连续被拒几次」;费率是别人按需调的,
                // 混进去就是替别人记账。**日志级别保持 INFO**(用户定:不与三条查询那套对齐)。
                let msg =
                    format!("{label}查询请求发送失败,错误代码:{n}(不自动重发:费率查询按需调用)");
                self.put_log(LogData {
                    gateway_name: self.gw.clone(),
                    msg: msg.clone(),
                    level: 20,
                });
                return QueryRateReply { ok: false, msg };
            }
            QueryRateReply {
                ok: true,
                msg: String::new(),
            }
        } else {
            self.warn_no_transport();
            QueryRateReply {
                ok: false,
                msg: "传输层未接入(默认构建)".to_string(),
            }
        }
    }

    fn do_cancel_order(&self, r: &CancelOrderReq) -> CancelOrderReply {
        let outs = {
            let mut td = self.td.lock().unwrap();
            td.cancel_order(&r.req)
        };
        let mut rejected_log: Option<LogData> = None;
        let outs: Vec<TdOut> = outs
            .into_iter()
            .filter(|o| match o {
                TdOut::Log(l) => {
                    rejected_log = Some(l.clone());
                    false
                }
                _ => true,
            })
            .collect();
        if let Some(l) = rejected_log {
            let msg = l.msg.clone();
            self.put_log(l);
            return CancelOrderReply { ok: false, msg };
        }
        for o in outs {
            self.emit_td(o);
        }
        if self.trans.is_none() {
            self.warn_no_transport();
            return CancelOrderReply {
                ok: false,
                msg: "传输层未接入(默认构建)".to_string(),
            };
        }
        CancelOrderReply {
            ok: true,
            msg: String::new(),
        }
    }

    /// 配置实现(插件自带配置面,0.3.0;定盘:界面/宿主手工输入经 ask,
    /// 插件不读配置文件)。
    ///
    /// 校验 → 应用(驱动 reconfigure 复位连接/认证/登录守卫 + 插件侧
    /// cfg/broker/user 换血)→ **强制重连**(0.5.0:先 `td_release`/`md_release`
    /// 断开旧会话——传输层幂等 connect 下,不断开就没有新登录显形,换账号/
    /// 改密码永不生效;Release 归 RAII Drop,可能阻塞到回调收尾)→ `connect()`
    /// (先 TD 后 MD)。纯逻辑面 release 默认空,行为不变。
    fn do_configure(&self, r: &ConfigureReq) -> ConfigureReply {
        self.configure(r)
    }

    fn do_status(&self, _r: &StatusReq) -> GatewayStatus {
        self.status()
    }

    fn do_disconnect(&self, _r: &DisconnectReq) -> DisconnectReply {
        self.disconnect()
    }

    /// 状态口(0.9.0 新增;公开与 `configure` 同级;ask 臂 `{gw}.status` 走它)。
    ///
    /// 纪律:**只读**——不改任何守卫、不触发重连、不写日志(界面按秒级刷,
    /// 每次刷都写日志会把 eLog 环冲爆,512 行取证环就是这么饱和的)。
    pub fn status(&self) -> GatewayStatus {
        let td = self.td.lock().unwrap().status_view();
        let md = self.md.lock().unwrap().status_view();
        GatewayStatus {
            gateway: self.gw.clone(),
            sampled_ms: crate::map::now_epoch_ms(),
            td_handle_up: td.handle_up,
            td_logged_in: td.logged_in,
            td_login_failed: td.login_failed,
            td_auth_ok: td.auth_ok,
            td_auth_failed: td.auth_failed,
            td_front_id: td.front_id,
            td_session_id: td.session_id,
            td_disconnects: td.disconnects,
            td_last_login_ok_ms: td.last_login_ok_ms,
            td_last_account_ms: td.last_account_ms,
            td_last_position_ms: td.last_position_ms,
            td_query_rejects: td.query_rejects,
            td_query_reject_streak: td.query_reject_streak,
            td_instrument_rejects: td.instrument_rejects,
            td_last_error: td.last_error,
            md_handle_up: md.handle_up,
            md_logged_in: md.logged_in,
            md_disconnects: md.disconnects,
            md_last_login_ok_ms: md.last_login_ok_ms,
            md_last_tick_ms: md.last_tick_ms,
            md_sub_rejects: md.sub_rejects,
            md_last_error: md.last_error,
            md_subscribed: md.subscribed,
        }
    }

    /// 断开口(0.9.0 新增;裁定 1「断开 = 停会话,实例与订阅保留」)。
    ///
    /// 放掉:两侧会话(传输层 release + 驱动守卫复位,查询阶段回 `Idle`)。
    /// 保留:插件实例、账号配置、订阅集、断开计数与「最近时刻」、持仓合并缓存、
    /// OrderSysID 映射。恢复走『连接』(= `configure` 重走全链,登录后自动重订阅)。
    pub fn disconnect(&self) -> DisconnectReply {
        if let Some(t) = &self.trans {
            t.td_release();
            t.md_release();
        }
        {
            self.td.lock().unwrap().reset_session();
        }
        {
            self.md.lock().unwrap().reset_session();
        }
        // 0.11.0 第4刀:会话复位 = 这一轮查询结束,把没发出去的合约尾巴发掉
        // (下一轮从头攒,不能让半批压在锁里)
        self.flush_contract();
        self.put_log(LogData {
            gateway_name: self.gw.clone(),
            msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
            level: LEVEL_INFO,
        });
        // 0.9.8 链状态行:账户表「阶段」列跟着回到「尚未开始查询」。
        self.put_log(LogData {
            gateway_name: self.gw.clone(),
            msg: format!("{}会话已复位", driver::CHAIN_STATE_PREFIX),
            level: LEVEL_INFO,
        });
        DisconnectReply {
            ok: true,
            msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
        }
    }

    /// 当前生效的产品类勾选(D-PC-FILTER)。
    ///
    /// 校验被拒时也回带它,而不是空表 —— 空表会让界面把勾选框清空,而网关其实还跑在
    /// 旧勾选上,那是假信息。
    fn effective_classes(&self) -> Vec<String> {
        self.td.lock().unwrap().product_classes()
    }

    /// 配置口(公开,与 `connect` 同级的显式控制面;ask 臂 `{gw}.configure`
    /// 走的也是它)。
    pub fn configure(&self, r: &ConfigureReq) -> ConfigureReply {
        if r.td_addr.is_empty() && r.md_addr.is_empty() {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: td_addr/md_addr 至少一个非空".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        if !r.td_addr.is_empty()
            && (r.broker_id.is_empty() || r.td_user.is_empty() || r.td_password.is_empty())
        {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: TD 侧需 broker_id/td_user/td_password 齐".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        let md_user = if !r.md_user.is_empty() {
            r.md_user.clone()
        } else {
            r.td_user.clone()
        };
        let md_password = if !r.md_password.is_empty() {
            r.md_password.clone()
        } else {
            r.td_password.clone()
        };
        if !r.md_addr.is_empty() && (md_user.is_empty() || md_password.is_empty()) {
            return ConfigureReply {
                ok: false,
                msg: "配置校验: MD 侧需 md_user/md_password(或 td 侧回落)".to_string(),
                product_classes: self.effective_classes(),
            };
        }
        // 产品类勾选校验(D-PC-FILTER):CTP 产品类码是单字符。非法项不静默接受,
        // 直接拒并列出;空串项视为「没填」,由归一化忽略(全空 → 回落默认期货)。
        let bad_codes: Vec<String> = r
            .product_classes
            .iter()
            .map(|c| c.trim().to_string())
            .filter(|c| !c.is_empty() && c.chars().count() != 1)
            .collect();
        if !bad_codes.is_empty() {
            return ConfigureReply {
                ok: false,
                msg: format!(
                    "配置校验: product_classes 需为单字符 CTP 码(非法项:{})",
                    bad_codes.join(",")
                ),
                product_classes: self.effective_classes(),
            };
        }
        let classes = crate::map::ProductClasses(r.product_classes.clone());
        let effective_classes = classes.normalize();
        {
            self.td.lock().unwrap().reconfigure(driver::TdConfig {
                addr: r.td_addr.clone(),
                user: r.td_user.clone(),
                password: r.td_password.clone(),
                broker: r.broker_id.clone(),
                auth_code: r.td_auth_code.clone(),
                app_id: r.td_app_id.clone(),
                product_classes: classes.clone(),
            });
        }
        {
            self.md.lock().unwrap().reconfigure(driver::MdConfig {
                addr: r.md_addr.clone(),
                user: md_user,
                password: md_password,
                broker: r.broker_id.clone(),
            });
        }
        *self.broker.lock().unwrap() = r.broker_id.clone();
        *self.user.lock().unwrap() = r.td_user.clone();
        *self.cfg.lock().unwrap() = CtpConfig {
            gateway_name: self.gw.clone(),
            broker_id: r.broker_id.clone(),
            td_addr: r.td_addr.clone(),
            td_user: r.td_user.clone(),
            td_password: r.td_password.clone(),
            td_auth_code: r.td_auth_code.clone(),
            td_app_id: r.td_app_id.clone(),
            md_addr: r.md_addr.clone(),
            md_user: r.md_user.clone(),
            md_password: r.md_password.clone(),
            product_classes: classes,
        };
        // 强制重连(0.5.0):旧会话先断,connect() 才会新建句柄走全链登录。
        if let Some(t) = &self.trans {
            t.td_release();
            t.md_release();
        }
        self.connect();
        // 受理三行序(BD-CFG-8/14;0.10.1):受理成功 = 唯一持久面。秘面登记/
        // 引用化在 dump_config 内完成,ask `cfg.save` 落手;cfg 缺席 → 直写
        // 库兜底(BD-CFG-10,审计行 BD-CFG-11 一次性)。回执零改(UX 零变)。
        self.persist_config();
        ConfigureReply {
            ok: true,
            msg: format!(
                "已配置并发起连接(先 TD 后 MD);收录产品类:{}",
                effective_classes.join(" ")
            ),
            product_classes: effective_classes,
        }
    }

    /// 凭据档位快照(只读;`{gw}.credentials_state` ask 臂与运维窗呈现共用,
    /// BD-CFG-17 的呈现语义:以文件引用为准)。
    #[must_use]
    pub fn credentials_state(&self) -> Vec<CredentialEntry> {
        let refs = lock_clone(&self.sec_refs);
        let cfg = lock_clone(&self.cfg);
        let mut out = Vec::with_capacity(2);
        for field in ["td_password", "md_password"] {
            let sk = refs
                .iter()
                .find(|(f, _)| f == field)
                .map(|(_, k)| k.clone())
                .unwrap_or_default();
            let resolved = if field == "td_password" {
                !cfg.td_password.is_empty()
            } else {
                !cfg.md_password.is_empty()
            };
            let state = if resolved {
                "RESOLVED"
            } else if sk.is_empty() {
                "ABSENT"
            } else {
                "REFERENCED"
            };
            out.push(CredentialEntry {
                field: field.to_owned(),
                state: state.to_owned(),
                ref_key: sk,
            });
        }
        out
    }

    /// 解密已回灌的引用行(**只允许 worker 时机调用** —— on_start;
    /// BD-CFG-16:宿主管理线程禁调秘取)。取不到 = 待手配,不改连接守卫
    /// (连接空值守卫原有语义原样);库值缺 / 非文本字节都当场 eLog 显形,
    /// 不过这两个显形挡翻倍:每次实例启动至多各一行(info 级防刷屏)。
    fn resolve_pending_secrets(&self) {
        let refs = lock_clone(&self.sec_refs);
        if refs.is_empty() {
            return;
        }
        let mut cfg = lock_clone(&self.cfg);
        let mut hit = false;
        for (field, sk) in &refs {
            match ek_cfg::secret::get(&self.ns, sk) {
                Some(v) => match std::str::from_utf8(v.as_bytes()) {
                    Ok(text) => {
                        if field == "td_password" {
                            cfg.td_password = text.to_owned();
                            hit = true;
                        } else if field == "md_password" {
                            cfg.md_password = text.to_owned();
                            hit = true;
                        }
                    }
                    Err(_) => {
                        // 坏字节(keyring 属文本面走不到;env 注入面可能发生)
                        // → 该键空置待手配,显形不静默(BD-U4)。
                        self.put_log(LogData {
                            gateway_name: self.gw.clone(),
                            msg: format!("{field} 的秘值不是文本(键 {sk}),按待手配处理"),
                            level: 40,
                        });
                    }
                },
                None => {
                    self.put_log(LogData {
                        gateway_name: self.gw.clone(),
                        msg: format!("{field} 的引用在库值缺(键 {sk}),待手配"),
                        level: 20,
                    });
                }
            }
        }
        if !hit {
            return;
        }
        // 解密结果落驱动(与 configure 的 reconfigure 同式但不 release/不
        // connect —— 连接时机仍归 on_start 的 connect,UX 行为零变)。
        let td_cfg = driver::TdConfig {
            addr: cfg.td_addr.clone(),
            user: cfg.td_user.clone(),
            password: cfg.td_password.clone(),
            broker: cfg.broker_id.clone(),
            auth_code: cfg.td_auth_code.clone(),
            app_id: cfg.td_app_id.clone(),
            product_classes: cfg.product_classes.clone(),
        };
        let md_cfg = driver::MdConfig {
            addr: cfg.md_addr.clone(),
            user: if cfg.md_user.is_empty() {
                cfg.td_user.clone()
            } else {
                cfg.md_user.clone()
            },
            password: if cfg.md_password.is_empty() {
                cfg.td_password.clone()
            } else {
                cfg.md_password.clone()
            },
            broker: cfg.broker_id.clone(),
        };
        self.td.lock().unwrap().reconfigure(td_cfg);
        self.md.lock().unwrap().reconfigure(md_cfg);
        *self.cfg.lock().unwrap() = cfg;
    }

    /// 受理三行序的落手段(BD-CFG-14):`dump_config`(秘面引用在这一步登记
    /// 进秘库)→ ask `cfg.save`;cfg 实例缺席(BD-CFG-9:UnknownService/
    /// TargetGone/超时同闸单分支)→ 直写 `Store`(BD-CFG-10 软集中,兜底
    /// 审计行 BD-CFG-11 每插件一次性)。成败都不改 configure 回执(UX 零变),
    /// 问题走 eLog 显形。**调用面 = 插件 worker 线程**(on_request →
    /// do_configure;secret::set 的 keyring 写在此许可,BD-CFG-16)。
    fn persist_config(&self) {
        // dump 一次取定(0.10.2 自审修正:原先兜底分支再 dump 一遍 —— 秘面
        // 重复登记无害但 rev 平白多爬一级;现在同一 map 走两条路)。
        let map = self.dump_config();
        if map.is_empty() {
            return; // BD-CFG-15:不输出 = 不持久(理应不会到;受理校验保底)
        }
        let ns = Configurable::config_ns(self).to_owned();
        let req = ek_cfg::wire::SaveReq {
            ns: ns.clone(),
            map: map.clone(),
        };
        let ctx = self.ctx.get();
        let res = match ctx {
            Some(c) => c.ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
                "cfg.save",
                req,
                std::time::Duration::from_millis(1500),
            ),
            // on_request 面 ctx 必在(on_init 先行);纯逻辑测试兜直库面。
            None => Err(ek_core::error::AskError::UnknownService {
                service: "cfg.save".to_owned(),
            }),
        };
        match res {
            Ok(r) if r.ok => {
                self.put_log(LogData {
                    gateway_name: self.gw.clone(),
                    msg: format!("配置已持久化(段 {});警告{:?}", ns, r.warns),
                    level: 20,
                });
            }
            Ok(r) => {
                self.put_log(LogData {
                    gateway_name: self.gw.clone(),
                    msg: format!("配置落盘失败(段 {}):{}", ns, r.msg),
                    level: 40,
                });
            }
            Err(_) => {
                // cfg 缺席 → 直接落库(BD-CFG-10);审计行一次性(BD-CFG-11)。
                let once = {
                    let mut f = self.save_fallback.lock().unwrap();
                    let first = !*f;
                    *f = true;
                    first
                };
                if once {
                    self.put_log(LogData {
                        gateway_name: self.gw.clone(),
                        msg: "cfg 缺席,已直写落盘".to_string(),
                        level: 20,
                    });
                }
                let ok = Self::store_direct(&ns, &map);
                if !ok {
                    self.put_log(LogData {
                        gateway_name: self.gw.clone(),
                        msg: format!("直写托底也失败(段 {})", ns),
                        level: 40,
                    });
                }
            }
        }
    }

    /// 直写库面(兜底;根同 cfg 实例的 eco → 本地回退口径)。
    /// `EK_CFG_FALLBACK_ROOT` = 部署/测试显缝(设了就走该根,不碰真配置目录)。
    fn store_direct(ns: &str, map: &SectionMap) -> bool {
        let store = match std::env::var("EK_CFG_FALLBACK_ROOT") {
            Ok(v) if !v.is_empty() => ek_cfg::Store::with_root(std::path::PathBuf::from(v)),
            _ => ek_cfg::Store::eco()
                .unwrap_or_else(|_| ek_cfg::Store::with_root("./ekcfg-local".into())),
        };
        store.store(ns, map).is_ok()
    }
}

// ================ 配置托底契约(BD-CFG 环节7) ================

impl Configurable for CtpPlugin {
    fn config_ns(&self) -> &str {
        &self.ns
    }

    /// 键位与段语义(0.10.0 v1;apply 为**补丁合入**,段缺键 = 沿用现值)。
    /// **本方法禁调 `secret::get`**(只被 boot/管理线程调用,BD-CFG-16)——
    /// 秘面只登记引用行,解密在 on_start(worker 面)。
    fn apply_config(&self, m: &SectionMap) -> Result<ApplyReport, String> {
        if m.rev > self.cfg_rev.load(Ordering::Relaxed) {
            self.cfg_rev.store(m.rev, Ordering::Relaxed);
        }
        let mut rep = ApplyReport::all_adopted();
        let mut patch = lock_clone(&self.cfg);
        let mut new_refs: Vec<(String, String)> = Vec::new();
        for (k, f) in m.iter() {
            match k {
                "broker_id" => {
                    patch.broker_id = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "td_addr" => {
                    patch.td_addr = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "td_user" => {
                    patch.td_user = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "td_auth_code" => {
                    patch.td_auth_code = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "td_app_id" => {
                    patch.td_app_id = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "md_addr" => {
                    patch.md_addr = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "md_user" => {
                    patch.md_user = f.plain_value().unwrap_or("").to_owned();
                    rep.adopted.push(k.to_owned());
                }
                "product_classes" => {
                    patch.product_classes = crate::map::ProductClasses(
                        f.plain_value()
                            .unwrap_or("")
                            .split(',')
                            .map(str::trim)
                            .filter(|c| !c.is_empty())
                            .map(str::to_owned)
                            .collect(),
                    );
                    rep.adopted.push(k.to_owned());
                }
                "td_password" | "md_password" => {
                    if let Some(sk) = f.secret_key() {
                        new_refs.push((k.to_owned(), sk.to_owned())); // 只登记引用
                        rep.adopted.push(k.to_owned());
                    } else {
                        rep.deferred.push(k.to_owned()); // 段里不该有明文
                    }
                }
                _ => rep.deferred.push(k.to_owned()), // 未知键:不采纳不覆盖
            }
        }
        *self.sec_refs.lock().unwrap() = new_refs;
        // 驱动同步(同 configure 的 reconfigure 形,但不 release/不 connect ——
        // 连接时机仍是 on_start 的 connect;本通道不改 UX 行为)。
        let td_cfg = driver::TdConfig {
            addr: patch.td_addr.clone(),
            user: patch.td_user.clone(),
            password: patch.td_password.clone(),
            broker: patch.broker_id.clone(),
            auth_code: patch.td_auth_code.clone(),
            app_id: patch.td_app_id.clone(),
            product_classes: patch.product_classes.clone(),
        };
        let md_cfg = driver::MdConfig {
            addr: patch.md_addr.clone(),
            user: if patch.md_user.is_empty() {
                patch.td_user.clone()
            } else {
                patch.md_user.clone()
            },
            password: if patch.md_password.is_empty() {
                patch.td_password.clone()
            } else {
                patch.md_password.clone()
            },
            broker: patch.broker_id.clone(),
        };
        self.td
            .lock()
            .map_err(|e| e.to_string())?
            .reconfigure(td_cfg);
        self.md
            .lock()
            .map_err(|e| e.to_string())?
            .reconfigure(md_cfg);
        *self.broker.lock().unwrap() = patch.broker_id.clone();
        *self.user.lock().unwrap() = patch.td_user.clone();
        *self.cfg.lock().unwrap() = patch;
        Ok(rep)
    }

    /// 导出(BD-CFG-15):非秘键 plain 出(**该键非空才落**),秘键只在
    /// 引用已登记时出引用形态;内存明文无引用 → 在本次落库登记(保存动作
    /// 线程;keyring 写毫秒级可忍)并出引用。**值永不落文件**(BD-CFG-12)。
    fn dump_config(&self) -> SectionMap {
        let next = self.cfg_rev.load(Ordering::Relaxed) + 1;
        self.cfg_rev.store(next, Ordering::Relaxed);
        let cfg = lock_clone(&self.cfg);
        let mut map = SectionMap::new(next);
        for (k, v) in [
            ("broker_id", &cfg.broker_id),
            ("td_addr", &cfg.td_addr),
            ("td_user", &cfg.td_user),
            ("td_auth_code", &cfg.td_auth_code),
            ("td_app_id", &cfg.td_app_id),
            ("md_addr", &cfg.md_addr),
            ("md_user", &cfg.md_user),
        ] {
            if !v.is_empty() {
                map = map.set(k, Field::plain(v));
            }
        }
        if !cfg.product_classes.0.is_empty() {
            map = map.set(
                "product_classes",
                Field::plain(cfg.product_classes.0.join(",")),
            );
        }
        for (field, value) in [
            ("td_password", &cfg.td_password),
            ("md_password", &cfg.md_password),
        ] {
            let refs = lock_clone(&self.sec_refs);
            if let Some((_, sk)) = refs.iter().find(|(f, _)| f == field) {
                // 既有引用:库值**幂等覆写为当前受理值**(受理面真相 = 连接面
                // 当前用着的密码;0.10.1 修正 —— 只换明文不覆写库值时,下一次
                // 启动会解回旧密码)。空明文 = 引用行原样(不碰库)。
                if !value.is_empty() {
                    let _ = ek_cfg::secret::set(
                        &self.ns,
                        sk,
                        ek_cfg::SecretValue::from_bytes(value.clone()),
                    );
                }
                map = map.set(field, Field::secret(sk.as_str()));
            } else if !value.is_empty() {
                // 明文登记落库 + 出引用(值永不出本 crate 边界)。
                let sk = format!("gw_{field}");
                if ek_cfg::secret::set(
                    &self.ns,
                    &sk,
                    ek_cfg::SecretValue::from_bytes(value.clone()),
                )
                .is_ok()
                {
                    self.sec_refs
                        .lock()
                        .unwrap()
                        .push((field.to_owned(), sk.clone()));
                    map = map.set(field, Field::secret(sk.as_str()));
                }
                // 存库失败(缺平台后端面):跳过该键不落盘。
            }
            // 空值:跳过(不持久空形)。
        }
        map
    }
}

impl Default for CtpPlugin {
    fn default() -> Self {
        Self::new(CtpConfig::default())
    }
}

impl Plugin for CtpPlugin {
    fn name(&self) -> &str {
        &self.gw
    }

    fn routes(&self) -> Vec<Cow<'static, str>> {
        vec![Cow::Owned(e_req_type(&self.gw)), Cow::Borrowed(ROUTE_TIMER)]
    }

    fn version(&self) -> Cow<'static, str> {
        Cow::Borrowed(env!("CARGO_PKG_VERSION"))
    }

    fn services(&self) -> Vec<ServiceDesc> {
        vec![
            ServiceDesc::describe::<SubscribeReq, SubscribeReply>(
                format!("{}.subscribe", self.gw),
                "登记并(已登录时)下发行情订阅",
            ),
            ServiceDesc::describe::<SendOrderReq, SendOrderReply>(
                format!("{}.send_order", self.gw),
                "下单(InputOrder + SUBMITTING 单已上总线)",
            ),
            ServiceDesc::describe::<CancelOrderReq, CancelOrderReply>(
                format!("{}.cancel_order", self.gw),
                "撤单(三段订单号解析)",
            ),
            ServiceDesc::describe::<QueryRateReq, QueryRateReply>(
                format!("{}.query_rate", self.gw),
                "费率查询(margin=保证金率/commission=手续费率;状态为纯被动推送)",
            ),
            ServiceDesc::describe::<ConfigureReq, ConfigureReply>(
                format!("{}.configure", self.gw),
                "网关配置(插件自带配置面;手工输入账号/前置 → 复位守卫并发起连接)",
            ),
            ServiceDesc::describe::<StatusReq, GatewayStatus>(
                format!("{}.{}", self.gw, SVC_STATUS),
                "状态读数(0.9.0;登录态 / 断开计数 / 查询被拒计数 / 最近回报时刻;只读、不写日志)",
            ),
            ServiceDesc::describe::<DisconnectReq, DisconnectReply>(
                format!("{}.disconnect", self.gw),
                "停会话(0.9.0;实例与订阅保留,『连接』可恢复)",
            ),
            // 0.10.0(BD-CFG 环节7):凭据档位只读口(运维窗「已存/未存」)。
            ServiceDesc::describe::<CredentialStateReq, Vec<CredentialEntry>>(
                format!("{}.credentials_state", self.gw),
                "凭据档位(REFERENCED/RESOLVED/ABSENT;只读)",
            ),
        ]
    }

    fn on_init(&self, ctx: &PluginContext) {
        let _ = self.ctx.set(ctx.clone());
    }

    fn on_start(&self) {
        // 0.10.0(BD-CFG-16):先在 worker 时机解密已回灌的引用行(管理线程
        // 禁取秘;on_start 属插件 worker 面),再按原有 connect 语义发起 ——
        // 拿不到的键留在「待手配」,连接空值守卫原有语义不变(UX 零变)。
        self.resolve_pending_secrets();
        // 传输层已接入 → 按配置发起连接(原稿由 app 显式调 CtpGateway.connect)。
        // 无传输层的纯逻辑构建保持原状:emit 走 warn_no_transport 提示一次。
        self.connect();
    }

    fn on_stop(&self) {
        // 先释放原生会话:之后 SDK 不再回调,回调入口才会随插件一起被放下。
        if let Some(t) = &self.trans {
            t.shutdown();
        }
        // 0.11.0 第4刀:摘插件前把攒着的合约尾巴发掉(半批不随插件进坟墓)
        self.flush_contract();
    }

    fn on_mail(&self, event: &Event) {
        if event.event_type == ROUTE_TIMER {
            self.on_timer();
            return;
        }
        if event.event_type == e_req_type(&self.gw) {
            self.on_request(event);
        }
    }

    fn as_configurable(&self) -> Option<&dyn Configurable> {
        // 0.10.0(BD-CFG 环节7):CTP 有配置面 → 显式 opt-in。
        Some(self)
    }
}

/// 名称富化面(0.11.0 第5刀 R6):委托 / 成交 / 持仓三张行共用同一个动作。
///
/// 刻意做成插件侧私有 trait —— 这是**发送前的网关侧富化**,不是映射层的事实;
/// `map.rs` 保持"柜台给了什么就写什么"(0.7.0 甲的纪律:映射层不吃合约参数)。
trait EnrichName {
    /// 查表键(en_symbol 同形:合约表就以它为键)
    fn symbol(&self) -> &str;
    /// 名称栏(空 = 待补)
    fn name_mut(&mut self) -> &mut String;
}

impl EnrichName for ek_data::order::OrderData {
    fn symbol(&self) -> &str {
        &self.symbol
    }
    fn name_mut(&mut self) -> &mut String {
        &mut self.name
    }
}

impl EnrichName for ek_data::trade::TradeData {
    fn symbol(&self) -> &str {
        &self.symbol
    }
    fn name_mut(&mut self) -> &mut String {
        &mut self.name
    }
}

impl EnrichName for ek_data::position::PositionData {
    fn symbol(&self) -> &str {
        &self.symbol
    }
    fn name_mut(&mut self) -> &mut String {
        &mut self.name
    }
}