ek_ui/

plugin.rs

//! `UiPlugin` —— 总线面:`eLog` 入环形日志、`ePlugUp/Down` 记事件 + rev、
//! `eBar`./`eBarUpdate.` 记行情库(K线窗数据面)、`eTick.` 记 tick 库(v4 行情带
//! 唯一源,S7 上屏地面)、`eContract.`/`eContractFinish` 记合约表,
//! **0.5 M1 起 `eOrder.`/`eTrade.`/`ePosition.`/`eAccount.` 四条推送线入交易账本**
//! (基线 BaseMonitor 同形:后台只 put,界面自己 process_event)。
//! `eTimer`(1s 宿主自带)那一拍只做三件事:管制面快照(`svctl.list`,每
//! `LIST_POLL_TICKS` 拍)、摊平的状态轮询(每拍最多 `STATUS_ASKS_PER_TICK` 个账户)、
//! 被标记时才问一次的装配显形(`oms.pipeline_status`)。**这一拍不再问交易数据,
//! 也不再辞行操作指令** —— 指令由命令线程执行(0.5 M5,点击即发)。
//!
//! 纪律(D1):UI 不周期取数;唯一例外是观测面 `svctl.list`(基线里对应物是同进程
//! 直读,不是推送)。纪律(D2):后台只 put、不驱动界面 —— 入账 worker 里没有任何 ask。唯一管制面是 ask 契约(svctl / dm.reload /
//! oms 查询与 submit_order·cancel_order / ctp 配置订阅,全部与
//! headless 同源;`PluginContext` 只在 worker 可用)。0.1.1 BD-UI-30:
//! 下单/撤单直发网关的旁路口删除——风控与守卫不可绕过。0.2.10 BD-UI-38:
//! sim 纸面网关退场(流控查询口随之撤)。

use std::sync::OnceLock;
use std::time::Duration;

use ek_core::context::PluginContext;
use ek_core::plugin::Plugin;
use ek_core::service::{PlugNotice, ServiceDesc};
use ek_core::{EVENT_PLUG_DOWN, EVENT_PLUG_UP};
use ek_ctp::plugin::{
    ConfigureReply, ConfigureReq, DisconnectReply, DisconnectReq, GatewayStatus, StatusReq,
    SubscribeReply as CtpSubscribeReply, SubscribeReq as CtpSubscribeReq,
};
use ek_data::account::AccountData;
use ek_data::event::{EVENT_BAR, EVENT_CONTRACT, EVENT_LOG, EVENT_TICK};
use ek_data::log::LogData;
use ek_data::order::OrderData;
use ek_data::position::PositionData;
use ek_data::trade::TradeData;
use ek_dm::plugin::{EVENT_BAR_UPDATE, ReloadReq};
use ek_event::EVENT_TIMER;
use ek_oms::plugin::{
    GetAllAccountsReq, GetAllOrdersReq, GetAllPositionsReq, GetAllTradesReq, OmsCancelOrderReq,
    OmsCancelReply, PipelineStatusReq, SubmitOrderReq,
};
use ek_oms::{SubmitReply, SubmitStatus};
use ek_station::wire::{
    ListReq, PlugRow, RemoveReq, RemoveRes, SVC_LIST, SVC_REMOVE, SVC_START, StartReq, StartRes,
};

use crate::state::{BusSnap, UiShared};

/// 注册名 = 工厂名 = `"ui"`(BD-U5:两者一致)。
pub const UI_NAME: &str = "ui";

/// 界面邮箱容量(0.5 M6:取代对账机制)。
///
/// ek-event 的总线队列是无界的(照 Python `queue.Queue`,engine.rs:21-22),会丢事件的
/// 只有 ek-core 这一跳有界邮箱 —— 而界面不再每秒自己问四次之后,它邮箱里就只剩
/// 「别人推来的事件」。1024 格在 30 条/秒的行情下只够 34 秒积压;10 万格 ≈ 55 分钟,
/// 溢出从常态变成异常。溢出仍然当场显形(0.4.5 P7d 的运维窗上色 + toast),但不再触发
/// 自动修复:基线没有对账,因为它不丢事件。
pub const UI_MAILBOX_CAP: usize = 100_000;

/// 交易四表的推送线(基线 `event.py:13-18`;ek-ctp 0.4.0 起发裸键,ek-core 精确匹配)。
/// 名字单源取 ek-oms::wire —— 同一条线在链上只有一份拼写。
const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;

/// 常驻观测插件(窗口面在 [`crate::runtime`] 持 [`UiShared`] 渲染)。
pub struct UiPlugin {
    shared: std::sync::Arc<UiShared>,
    ctx: OnceLock<PluginContext>,
    /// 命令线程的停线位(0.5 M5;on_stop 置真,线程自己退)。
    cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl UiPlugin {
    /// 与窗口面共用的共享状态句柄。
    pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
        Self {
            shared,
            ctx: OnceLock::new(),
            cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
        }
    }

    /// eTimer 节流回调(0.5:管制面快照 + 摊平的状态轮询 + 事件触发的装配显形)。
    fn refresh(&self) {
        let Some(ctx) = self.ctx.get() else {
            return; // 未 init 就吃到 eTimer:忽略,等下轮
        };
        // 0.5 M3/M5:这一拍只做三件事,而且都不问交易数据。
        // 指令辞行搬到命令线程(`spawn_cmd_thread`);四张表吃 eOrder. / eTrade. /
        // ePosition. / eAccount. 推送入账(基线 BaseMonitor 同形)—— 这一拍里再出现
        // 任何「取数类 ask」就是回到 0.4 的病。
        if self.shared.list_poll_due() {
            self.poll_list(ctx);
        }
        self.poll_status(ctx);
        if self.shared.take_pipeline_due() {
            self.poll_pipeline(ctx);
        }
    }

    /// 管制面快照(`svctl.list`):谁在场 + 服务目录 + 邮箱溢出计数。
    ///
    /// 0.5 M3 降到每 `LIST_POLL_TICKS` 拍一次。它给的不是交易数据而是「宿主长什么样」
    /// 的观测面 —— 基线里的对应物是 `get_all_gateway_names` 的同进程直读,不是推送,
    /// 所以它留在白名单里(「界面不周期取数」这条纪律的唯一例外)。
    fn poll_list(&self, ctx: &PluginContext) {
        match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
            Ok(rows) => {
                let mut services = Vec::with_capacity(rows.len());
                for r in &rows {
                    let ids = ctx
                        .services_of(&r.name)
                        .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
                        .unwrap_or_default();
                    services.push((r.name.clone(), ids));
                }
                // 账户在场位回填 + 「实例没了就作废会话读数」。
                // 0.5:「新装配」不再由这里检测 —— ePlugUp. 已订阅,即时且不靠轮询。
                self.shared.sync_accounts_connected(&rows);
                // 0.4.5(P7d):邮箱溢出的当场显形。运维窗那一列一直有 spilled,
                // 但 ek-core 的 stderr 诊断每 128 条才打一条 —— 屏上此前没有任何东西。
                self.shared.note_plugin_spill(&rows);
                self.shared.set_snapshot(BusSnap { rows, services });
            }
            Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e:?}")),
        }
    }

    /// 状态轮询(0.5 M3 摊平):每拍最多问 `STATUS_ASKS_PER_TICK` 个账户,环形推进。
    ///
    /// 只问「已装配且在场」的:实例已被摘除时 ask 会等到超时(2 秒),按账户串行就是
    /// 按账户把整个刷新循环拖住 —— 而那种账户本来也没有会话可问。只读口不写日志
    /// (ek-ctp D-STATUS ④),所以这条轮询不会冲 eLog 环。
    fn poll_status(&self, ctx: &PluginContext) {
        let present: Vec<String> = self
            .shared
            .accounts()
            .iter()
            .filter(|a| a.stage == crate::state::AccountStage::Assembled && a.connected)
            .map(|a| a.gateway.clone())
            .collect();
        for gw in self.shared.status_batch(&present) {
            let svc = format!("{gw}.status");
            match ctx.ask_service::<StatusReq, GatewayStatus>(
                &svc,
                StatusReq,
                Duration::from_secs(2),
            ) {
                Ok(st) => self.shared.note_gateway_status(&gw, &st),
                Err(e) => self.shared.note_status_unreachable(&gw, &format!("{e:?}")),
            }
        }
    }

    /// 装配显形(0.5 M3 改事件触发):`oms.pipeline_status` 只在被标记的那一拍问一次。
    ///
    /// 标记来自启动、`ePlugUp.oms` 与运维窗的刷新钮。它此前是每 4 拍一次的轮询 ——
    /// 装配很少变,而它是 oms worker 上又一次串行 ask;在事件驱动模型里这种轮询是多余的。
    fn poll_pipeline(&self, ctx: &PluginContext) {
        match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
            "oms.pipeline_status",
            PipelineStatusReq,
            Duration::from_secs(2),
        ) {
            Ok(lines) => self.shared.set_pipeline_lines(lines),
            Err(e) => {
                // 不每拍刷日志:oms 不在场是合法态(清单里没选它,ask 走 TargetGone
                // 快错)。只在「从有到无」那一次显形一条,与状态轮询「只读口不冲 eLog 环」
                // 同口径。读数清空 → 运维窗那一节整节不画。
                if !self.shared.pipeline_lines().is_empty() {
                    self.shared
                        .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
                }
                self.shared.set_pipeline_lines(Vec::new());
            }
        }
    }

    /// 启动一次四问(0.5 M4;基线无对应物 —— 插件热插拔是 ek 独有场景)。
    ///
    /// 一次性,不是周期:之后四张表只吃推送。走 `note_*` 而不是 `seed_trade_snap`,
    /// 是为了**合并而不是覆盖** —— 网关可能在 oms 还没消化完之前就已经把委托推给了
    /// 界面,覆盖式播种会把界面已经看见的那一笔抹掉。
    ///
    /// 执行位置 = 命令线程(0.5 执行核查改的):原先它挂在 `on_start` 里,而 `on_start`
    /// 是宿主启动序列的一部分 —— 四口最坏 4 × 2 秒会把后面插件的启动一起拖住;挂在
    /// 入账 worker 上又违反「入账 worker 里没有任何 ask」。所以它跟着命令线程走。
    fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
        let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
            "oms.get_all_orders",
            GetAllOrdersReq,
            Duration::from_secs(2),
        );
        let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
            "oms.get_all_trades",
            GetAllTradesReq,
            Duration::from_secs(2),
        );
        let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
            "oms.get_all_positions",
            GetAllPositionsReq,
            Duration::from_secs(2),
        );
        let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
            "oms.get_all_accounts",
            GetAllAccountsReq,
            Duration::from_secs(2),
        );
        let mut n = 0usize;
        if let Ok(rows) = &orders {
            for o in rows {
                shared.note_order(o);
            }
            n += rows.len();
        }
        if let Ok(rows) = &trades {
            for t in rows {
                shared.note_trade(t);
            }
            n += rows.len();
        }
        if let Ok(rows) = &positions {
            for p in rows {
                shared.note_position(p);
            }
            n += rows.len();
        }
        if let Ok(rows) = &funds {
            for a in rows {
                shared.note_funds(a);
            }
            n += rows.len();
        }
        // 只在真播种到东西时落一条;空清单(刚装配、柜台还没回报)不刷日志。
        if n > 0 {
            let bad = usize::from(orders.is_err())
                + usize::from(trades.is_err())
                + usize::from(positions.is_err())
                + usize::from(funds.is_err());
            shared.log_line(format!(
                "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
                orders.as_ref().map_or(0, |v| v.len()),
                trades.as_ref().map_or(0, |v| v.len()),
                positions.as_ref().map_or(0, |v| v.len()),
                funds.as_ref().map_or(0, |v| v.len()),
                bad
            ));
        }
    }

    /// 命令线程(0.5 M5):点击即发,不等 eTimer 拍,也不占 ui 的入账 worker。
    ///
    /// 基线对应物是「UI 主线程直调 `main_engine.send_order`,立即返回」(engine.py:915-923)。
    /// ek-core 的 ask 是同步的,所以把「等回包」这件事从入账线程挪走:入账线程从此只
    /// 做「收事件 → 写账本」,每事件 O(1),永不被后端拖住。
    /// 这一条线程也是「会等的东西」唯一允许待的地方:启动一次四问在这里跑
    /// (见 `reconcile_trade`),既不占宿主的启动序列,也不占入账 worker。
    fn spawn_cmd_thread(&self) {
        let Some(ctx) = self.ctx.get().cloned() else {
            return; // 未 init:没有管制面可用
        };
        let shared = std::sync::Arc::clone(&self.shared);
        let stop = std::sync::Arc::clone(&self.cmd_stop);
        if std::thread::Builder::new()
            .name("ui-cmd".to_string())
            .spawn(move || {
                loop {
                    if stop.load(std::sync::atomic::Ordering::Acquire) {
                        break;
                    }
                    // 用户指令优先于播种:播种最坏 4 × 2 秒,不能把点击排到它后面。
                    if shared.cmd_pending() {
                        Self::run_cmds(&shared, &ctx);
                        continue;
                    }
                    // 播种档:启动一次 + 之后每次网关 / oms 上线各一次(标记是消费性的)。
                    if shared.take_trade_reseed() {
                        Self::reconcile_trade(&shared, &ctx);
                        continue;
                    }
                    std::thread::sleep(Duration::from_millis(20));
                }
            })
            .is_err()
        {
            // 线程起不来 = 指令只会进队列不会执行。这一条必须上屏,不能静默降级。
            self.shared
                .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
        }
    }

    /// 辞行指令队列(0.5 M5:由命令线程调用,不在 ui worker 上;回执进日志,面板语义可断言)。
    fn run_cmds(shared: &UiShared, ctx: &PluginContext) {
        for cmd in shared.drain_cmds() {
            match cmd {
                crate::state::UiCmd::Start { factory, name } => {
                    match ctx.ask_service::<StartReq, StartRes>(
                        SVC_START,
                        StartReq {
                            factory: factory.clone(),
                            name: name.clone(),
                        },
                        Duration::from_secs(3),
                    ) {
                        Ok(StartRes::Started { name }) => {
                            shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
                        }
                        Ok(StartRes::Rejected { reason }) => {
                            shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
                        }
                    }
                    // 起停回执之后重问一次管制面:运维窗那一行与账户在场位立刻动。
                    shared.request_list_refresh();
                }
                crate::state::UiCmd::Remove { name } => {
                    // 观测面自身的闸(0.4.3 BD-UI-64,审核 P1;0.5 M5 换了理由,闸不换):
                    // 0.4 的理由是「这条指令在 ui 自己的 worker 上执行 → 摘 ui = 自己等自己
                    // (ek-core `close_one` 只能走到 2 秒超时的 detach 分支)」。
                    // 0.5 执行者换成命令线程,「自己等自己」这一半不成立了,但结论一样:
                    // 摘掉 ui 之后没有 eTimer 拍再跑快照与状态轮询、入账 worker 也没了,
                    // 窗口还开着而里面全是上一帧。所以仍然直拒,不发 svctl。
                    // 管制面 `svctl` 的同一道闸长在 ek-station supervisor 侧(0.2.11)。
                    if name == UI_NAME {
                        shared.log_line(
                            "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
                                .to_string(),
                        );
                        continue;
                    }
                    match ctx.ask_service::<RemoveReq, RemoveRes>(
                        SVC_REMOVE,
                        RemoveReq { name: name.clone() },
                        Duration::from_secs(3),
                    ) {
                        Ok(RemoveRes::Removed { name }) => {
                            shared.log_line(format!("[ops] svctl.remove {name} 完成"));
                        }
                        Ok(RemoveRes::NoSuchName { name }) => {
                            shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
                        }
                        // 0.4.3:回执枚举增项(ek-station 0.2.11)—— 名字在场但不可摘,
                        // 或宿主另有错。原先这一档被塞进 NoSuchName 的 name 栏(错误文本
                        // 当名字打),这一栏现在照请求回填,错因只进 reason。
                        Ok(RemoveRes::Rejected { name, reason }) => {
                            shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
                        }
                    }
                    shared.request_list_refresh();
                }
                crate::state::UiCmd::Reload {
                    interval,
                    symbol,
                    bars,
                } => {
                    // 回灌走 dm.reload ask(push=true 逐条广播 → 行情板实时见回执面)。
                    match ctx.ask_service::<ReloadReq, usize>(
                        "dm.reload",
                        ReloadReq {
                            interval,
                            symbol: symbol.clone(),
                            bars: bars.clone(),
                            size: Some(bars.len()),
                            push: true,
                        },
                        Duration::from_secs(5),
                    ) {
                        Ok(n) => {
                            shared.market().note_reload();
                            shared.log_line(format!(
                                "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
                            ));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
                        }
                    }
                }
                crate::state::UiCmd::SendOrder { req, gateway, mode } => {
                    // 0.1.1 BD-UI-30 薄下单者:五段管线全走 oms.submit_order
                    // (拆单/风控/逐腿发送/记账;直发 {gw}.send_order 旁路口删除,
                    // 守卫不可绕过)。parent 聚合逐腿显形进日志。
                    match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
                        "oms.submit_order",
                        SubmitOrderReq {
                            req: req.clone(),
                            gateway_name: gateway.clone(),
                            mode,
                        },
                        Duration::from_secs(10),
                    ) {
                        Ok(reply) => {
                            let tag = match reply.status {
                                SubmitStatus::Sent => "SENT",
                                SubmitStatus::Partial => "PARTIAL",
                                SubmitStatus::Rejected => "REJECTED",
                            };
                            for leg in &reply.legs {
                                match (leg.ok, &leg.en_orderid, &leg.reason) {
                                    (true, Some(id), _) => shared.log_line(format!(
                                        "[ops] submit {} {} {}x{} → {id} (parent {})",
                                        leg.req.symbol,
                                        leg.req.direction.as_str(),
                                        leg.req.offset.as_str(),
                                        leg.req.volume,
                                        reply.parent_id
                                    )),
                                    _ => shared.log_line(format!(
                                        "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
                                        leg.req.symbol,
                                        leg.req.direction.as_str(),
                                        leg.req.offset.as_str(),
                                        leg.req.volume,
                                        leg.reason.as_deref().unwrap_or("?"),
                                        reply.parent_id
                                    )),
                                }
                            }
                            if reply.legs.is_empty() {
                                shared.log_line(format!(
                                    "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
                                    req.symbol,
                                    req.offset.as_str(),
                                    req.volume,
                                    reply.parent_id
                                ));
                            }
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
                        }
                    }
                }
                crate::state::UiCmd::CancelOrder { orderid, gateway } => {
                    // 0.1.1 BD-UI-30:守卫撤单(尝试 ≤4000 在 oms;直发
                    // {gw}.cancel_order 旁路口删除)。en_orderid 组式与
                    // OrderData.en_orderid() 同式。
                    match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
                        "oms.cancel_order",
                        OmsCancelOrderReq {
                            en_orderid: format!("{gateway}.{orderid}"),
                            gateway_name: gateway.clone(),
                        },
                        Duration::from_secs(10),
                    ) {
                        Ok(OmsCancelReply { ok: true, msg }) => {
                            shared.log_line(format!("[ops] {msg}"));
                        }
                        Ok(OmsCancelReply { ok: false, msg }) => {
                            shared.log_line(format!("[ops] 撤单被拒: {msg}"));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
                        }
                    }
                }
                crate::state::UiCmd::CtpConfigure { gateway, form } => {
                    // 定盘口径:界面手工输入 → 插件自带配置口(插件不读文件)。
                    // 多账号面:服务名随网关走(`{gw}.configure`),不再硬编码 ctp。
                    let svc = format!("{gateway}.configure");
                    match ctx.ask_service::<ConfigureReq, ConfigureReply>(
                        svc.as_str(),
                        form.to_configure_req(),
                        Duration::from_secs(5),
                    ) {
                        // `..`:ConfigureReply 0.6.0 起带 product_classes(实际生效值)。
                        // 这里不回填表单勾选 —— 被拒时回带的是「网关当前仍在跑」的那一份,
                        // 用它覆盖用户刚点的勾选会把他的意图吃掉;日志里网关已写明收录产品类。
                        Ok(ConfigureReply { ok: true, msg, .. }) => {
                            shared.log_line(format!("[ops] {svc} 受理:{msg}"));
                        }
                        Ok(ConfigureReply { ok: false, msg, .. }) => {
                            shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
                        }
                    }
                }
                crate::state::UiCmd::Connect { gateway, form } => {
                    // **四态分离的「连接」**(规划裁定「添加 ≠ 连接」):一次做完
                    // `svctl.start(factory=ctp)` + `{gw}.configure`,结果回写账户行
                    // (`Assembling → Assembled / AssembleFailed` + 失败原因留显形)。
                    // 失败**不回滚清册行** —— 行留着给『重试』(旧口径"必须先移除才能重试"作废)。
                    shared.note_connect_sent(&gateway);
                    // 单源纪律在此裁决:非行情源账户的 md 位就地清空,**留档与下发同一份**
                    // (旧口径只清下发不清留档,编辑页回填会带着一根用不上的行情前置)。
                    let form = shared.prepare_connect_form(&gateway, form);
                    let started = match ctx.ask_service::<StartReq, StartRes>(
                        SVC_START,
                        StartReq {
                            factory: "ctp".to_string(),
                            name: gateway.clone(),
                        },
                        Duration::from_secs(3),
                    ) {
                        Ok(StartRes::Started { name }) => {
                            shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
                            true
                        }
                        Ok(StartRes::Rejected { reason }) => {
                            // 防重入(规划 Q7):实例已在场时重复点『连接』/『重试』不是失败,
                            // 重投凭据即可(否则会把正在跑的账户误判成"装配失败")。
                            if shared.account(&gateway).is_some_and(|a| a.connected) {
                                shared
                                    .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
                                true
                            } else {
                                shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
                                shared.note_connect_result(&gateway, Err(reason));
                                continue;
                            }
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
                            shared.note_connect_result(&gateway, Err(format!("{e:?}")));
                            continue;
                        }
                    };
                    if !started {
                        continue;
                    }
                    let svc = format!("{gateway}.configure");
                    match ctx.ask_service::<ConfigureReq, ConfigureReply>(
                        svc.as_str(),
                        form.to_configure_req(),
                        Duration::from_secs(5),
                    ) {
                        // `..`:ConfigureReply 带的 product_classes(实际生效值)不回填勾选
                        // (被拒时回带的是「网关仍在跑」的那一份,覆盖用户刚点的勾选会吃掉意图);
                        // 0.2.18 起把它显示出来(见 state::note_effective_product_classes)。
                        Ok(ConfigureReply {
                            ok: true,
                            msg,
                            product_classes,
                        }) => {
                            shared.log_line(format!("[ops] {svc} 受理:{msg}"));
                            shared.note_effective_product_classes(&gateway, &product_classes);
                            shared.note_connect_result(&gateway, Ok(()));
                        }
                        Ok(ConfigureReply { ok: false, msg, .. }) => {
                            shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
                            shared.note_connect_result(&gateway, Err(msg));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
                            shared.note_connect_result(&gateway, Err(format!("{e:?}")));
                        }
                    }
                }
                crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
                    // 多账号面:一律路由到行情元账号(渲染侧已按此入列);
                    // 纯 TD 账号收到订阅会**静默成功**(do_subscribe 恒 ok:true),
                    // 所以路由正确性靠界面纪律,不靠网关报错(规划 §9.4-21)。
                    let svc = format!("{gateway}.subscribe");
                    match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
                        svc.as_str(),
                        CtpSubscribeReq {
                            symbols: symbols.clone(),
                        },
                        Duration::from_secs(5),
                    ) {
                        Ok(CtpSubscribeReply { ok: true, msg }) => {
                            shared.note_subscriptions(&symbols);
                            shared.log_line(format!(
                                "[ops] {svc} {} 受理:{}",
                                symbols.join(","),
                                msg
                            ));
                        }
                        Ok(CtpSubscribeReply { ok: false, msg }) => {
                            shared.log_line(format!(
                                "[ops] {svc} {} 被拒:{}",
                                symbols.join(","),
                                msg
                            ));
                        }
                        Err(e) => {
                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
                        }
                    }
                }
                crate::state::UiCmd::Disconnect { gateway } => {
                    // 0.2.19 规划裁定 1:『断开』= `{gw}.disconnect`(停会话),
                    // 『移除』= `svctl.remove`(停实例)—— 两条走两条不同的口,不共用。
                    // 发令即把该行标成「已断开」:用户按了钮就要立刻看到,不等下一轮轮询。
                    let svc = format!("{gateway}.disconnect");
                    match ctx.ask_service::<DisconnectReq, DisconnectReply>(
                        svc.as_str(),
                        DisconnectReq,
                        Duration::from_secs(5),
                    ) {
                        Ok(DisconnectReply { ok: true, msg }) => {
                            shared.mark_account_stopped(&gateway);
                            shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
                        }
                        Ok(DisconnectReply { ok: false, msg }) => {
                            shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
                        }
                        Err(e) => {
                            shared.log_line(format!(
                                "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
                            ));
                        }
                    }
                }
            }
        }
    }
}

impl Plugin for UiPlugin {
    fn name(&self) -> &str {
        UI_NAME
    }

    fn deps(&self) -> &[&str] {
        &[] // 纯观测;谁先上都能看到(bus 契约不依赖启动序)
    }

    fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
        vec![
            std::borrow::Cow::Borrowed(EVENT_LOG),
            std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
            std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
            std::borrow::Cow::Borrowed(EVENT_BAR),
            std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
            std::borrow::Cow::Borrowed(EVENT_TICK),
            std::borrow::Cow::Borrowed(EVENT_CONTRACT),
            std::borrow::Cow::Borrowed("eContractFinish"),
            // 0.5 M1 交易四表的推送线(基线 BaseMonitor 各自订阅自己的线)。
            // 不订 eTradeSync.:ek-oms 处理完成交后重推的那一条(plugin.rs:305-308),
            // 与 ek-ctp 的 eTrade. 是同一笔成交的两份 —— 订两条就是重复入账。
            std::borrow::Cow::Borrowed(W_ORDER),
            std::borrow::Cow::Borrowed(W_TRADE),
            std::borrow::Cow::Borrowed(W_POSITION),
            std::borrow::Cow::Borrowed(W_ACCOUNT),
            std::borrow::Cow::Borrowed(EVENT_TIMER),
        ]
    }

    fn mailbox_cap(&self) -> usize {
        UI_MAILBOX_CAP
    }

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

    fn services(&self) -> Vec<ServiceDesc> {
        Vec::new() // 纯视图零服务;管制面在 svctl
    }

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

    fn on_start(&self) {
        self.shared.log_line("[ui] 界面插件上线");
        // 0.5 M5 + M4:命令线程先起(点击即发);启动一次四问在那条线程里跑 ——
        // `on_start` 是 `engine.start()` 启动链的一环,四口最坏 4 × 2 秒不许把后面
        // 插件的启动一起拖住,更不许占入账 worker。
        self.spawn_cmd_thread();
        // 启动一次四问的标记(on_start 是 `engine.start()` 启动链的一环,四口最坏
        // 4 × 2 秒不许把后面插件的启动一起拖住,更不许占入账 worker)。
        self.shared.request_trade_reseed();
        self.shared.request_pipeline_refresh();
    }

    fn on_stop(&self) {
        self.shared.log_line("[ui] 界面插件下线");
        // 命令线程停线:它在 20ms 空转切片自查这个位,不会卡在 sleep 里。
        self.cmd_stop
            .store(true, std::sync::atomic::Ordering::Release);
    }

    fn on_mail(&self, event: &ek_event::Event) {
        let ty = event.event_type.as_str();
        if ty == EVENT_LOG {
            if let Some(d) = event.data.downcast_ref::<LogData>() {
                let tag = if d.is_warning_or_above() {
                    "WARN"
                } else {
                    "INFO"
                };
                self.shared
                    .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
                // 0.5.0(用户定「链进行的状态改为事件日志推送」):`[链] ` 前缀的行除了进日志窗,
                // 还落进账户行的「阶段」栏(ek-ctp 0.9.8 D-QUERY-GATE ⑧)。这一条是纯显示落账,
                // 不参与 `apply_status` 那套在线/断供判据。
                // 前缀这一份是本地常量(ek-ui 现在钉的还是 ek-ctp 0.9.7,那上面没有这个常量);
                // 抬 pin 到 0.9.8 之后改引用 `ek_ctp::driver::CHAIN_STATE_PREFIX`,回到单源。
                if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
                    self.shared.note_chain(&d.gateway_name, note);
                }
            } else {
                self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
            }
            return;
        }
        if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
            if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
                let verb = if ty == EVENT_PLUG_UP {
                    "上线"
                } else {
                    "下线"
                };
                self.shared
                    .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
                // 在场位与运维窗那一行跟着这条线动(不等 4 拍的快照节拍)。
                self.shared.request_list_refresh();
                if ty == EVENT_PLUG_UP {
                    // 0.5 M3:装配显形改吃这条线 —— 不再靠每拍 list 去发现「谁新上线」。
                    if n.name == "oms" {
                        self.shared.request_pipeline_refresh();
                    }
                    // 0.5 执行核查补的第三处:网关 / oms 上线之后重播种一次四张表。
                    // 摘除一个网关账户会作废它名下的委托与成交行,而 oms 那边那些行还在;
                    // 重新添加时不重问一次,界面就永久少一段历史。顺带覆盖「ui 先于 oms
                    // 起来、第一次播种问空了」这一序。
                    let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
                    if n.name == "oms" || is_account {
                        self.shared.request_trade_reseed();
                    }
                    // 0.5 M3:订阅重放同样改吃这条线(与 `sync_accounts_connected` 的
                    // newly 回带是同一件事的两条路,留一条:这条即时、那条要等快照拍)。
                    let md = self.shared.md_source();
                    let subs = self.shared.subscriptions();
                    if !md.is_empty() && n.name == md && !subs.is_empty() {
                        self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
                            gateway: n.name.clone(),
                            symbols: subs.clone(),
                        });
                        self.shared.log_line(format!(
                            "[ops] 行情源 {} 上线,重放订阅 {} 项",
                            n.name,
                            subs.len()
                        ));
                    }
                } else {
                    // 0.5 M1:网关下线 = 它名下的委托 / 成交 / 持仓 / 资金行全部作废。
                    // 推送模型里没人再告诉界面「这一行已经不存在了」,不清就是替一个
                    // 已经不存在的会话说话(与 0.2.19「实例不在场就作废会话读数」同口径)。
                    let removed = self.shared.clear_gateway_rows(&n.name);
                    if removed > 0 {
                        self.shared.log_line(format!(
                            "[ui] 插件下线 {}:作废 {} 行交易数据",
                            n.name, removed
                        ));
                    }
                    // 0.5 执行核查补的第二处:摘掉 oms 之后管线读数不会自己清空。
                    // 0.4 是每 4 拍问一次,4 秒内自己就空了;改事件触发后没人再问,
                    // 运维窗那一节会替一个已经不存在的 oms 说话。标记一次:下一拍问一次
                    // → TargetGone 快错 → `poll_pipeline` 的 Err 分支清空并落一条日志。
                    if n.name == "oms" {
                        self.shared.request_pipeline_refresh();
                    }
                }
            }
            self.shared.bump();
            return;
        }
        if ty == EVENT_BAR {
            if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
                self.shared.market().on_frame(f);
            }
            self.shared.bump();
            return;
        }
        if ty == EVENT_BAR_UPDATE {
            if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
                self.shared.market().on_update(b);
            }
            self.shared.bump();
            return;
        }
        if ty == EVENT_TICK {
            if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
                self.shared.update_tick(t);
            }
            return;
        }
        if ty == EVENT_CONTRACT {
            if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
                self.shared.update_contract(c);
            }
            return;
        }
        if ty == "eContractFinish" {
            let gw = event
                .data
                .downcast_ref::<String>()
                .cloned()
                .unwrap_or_else(|| "?".to_string());
            self.shared
                .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
            return;
        }
        // 0.5 M1 交易四表:后台只把事件推过来,界面自己入账(基线 BaseMonitor.process_event)。
        // 这里不 ask 任何东西,所以入账 worker 永不被后端拖住。
        if ty == W_ORDER {
            match event.data.downcast_ref::<OrderData>() {
                Some(o) => self.shared.note_order(o),
                None => self.shared.note_bad_payload(W_ORDER),
            }
            return;
        }
        if ty == W_TRADE {
            match event.data.downcast_ref::<TradeData>() {
                Some(t) => self.shared.note_trade(t),
                None => self.shared.note_bad_payload(W_TRADE),
            }
            return;
        }
        if ty == W_POSITION {
            match event.data.downcast_ref::<PositionData>() {
                Some(p) => self.shared.note_position(p),
                None => self.shared.note_bad_payload(W_POSITION),
            }
            return;
        }
        if ty == W_ACCOUNT {
            match event.data.downcast_ref::<AccountData>() {
                Some(a) => self.shared.note_funds(a),
                None => self.shared.note_bad_payload(W_ACCOUNT),
            }
            return;
        }
        if ty == EVENT_TIMER {
            self.refresh();
        }
    }
}