Skip to main content

ek_ui/
plugin.rs

1//! `UiPlugin` —— 总线面:`eLog` 入环形日志、`ePlugUp/Down` 记事件 + rev、
2//! `eBar`./`eBarUpdate.` 记行情库(K线窗数据面)、`eTick.` 记 tick 库(v4 行情带
3//! 唯一源,S7 上屏地面)、`eContract.`/`eContractFinish` 记合约表,
4//! **0.5 M1 起 `eOrder.`/`eTrade.`/`ePosition.`/`eAccount.` 四条推送线入交易账本**
5//! (基线 BaseMonitor 同形:后台只 put,界面自己 process_event)。
6//! `eTimer`(1s 宿主自带)那一拍只做三件事:管制面快照(`svctl.list`,每
7//! `LIST_POLL_TICKS` 拍)、**待播种队列**触发的状态播种(`seed_status`:网关刚上线/
8//! 刚重连时一次性问 `{gw}.status`,常态**零轮询** —— 0.7.0 第3刀退役了 0.5 时代的
9//! 摊平状态轮询,登录态读数吃 `eStatus.` 推送)、被标记时才问一次的装配显形
10//! (`oms.pipeline_status`)。**这一拍不再问交易数据,
11//! 也不再辞行操作指令** —— 指令由命令线程执行(0.5 M5,点击即发)。
12//!
13//! 纪律(D1):UI 不周期取数;唯一例外是观测面 `svctl.list`(基线里对应物是同进程
14//! 直读,不是推送)。纪律(D2):后台只 put、不驱动界面 —— 入账 worker 里没有任何 ask。唯一管制面是 ask 契约(svctl / dm.reload /
15//! oms 查询与 submit_order·cancel_order / ctp 配置订阅,全部与
16//! headless 同源;`PluginContext` 只在 worker 可用)。0.1.1 BD-UI-30:
17//! 下单/撤单直发网关的旁路口删除——风控与守卫不可绕过。0.2.10 BD-UI-38:
18//! sim 纸面网关退场(流控查询口随之撤)。
19
20use std::sync::OnceLock;
21use std::time::Duration;
22
23use ek_core::context::PluginContext;
24use ek_core::plugin::Plugin;
25use ek_core::service::{PlugNotice, ServiceDesc};
26use ek_core::{EVENT_PLUG_DOWN, EVENT_PLUG_UP};
27use ek_ctp::plugin::{
28    ConfigureReply, ConfigureReq, DisconnectReply, DisconnectReq, GatewayStatus, StatusReq,
29    SubscribeReply as CtpSubscribeReply, SubscribeReq as CtpSubscribeReq,
30};
31use ek_data::account::AccountData;
32use ek_data::event::{EVENT_BAR, EVENT_CONTRACT, EVENT_LOG, EVENT_TICK};
33use ek_data::log::LogData;
34use ek_data::order::OrderData;
35use ek_data::position::PositionData;
36use ek_data::trade::TradeData;
37use ek_dm::plugin::{EVENT_BAR_UPDATE, ReloadReq};
38use ek_event::EVENT_TIMER;
39use ek_oms::plugin::{
40    GetAllAccountsReq, GetAllOrdersReq, GetAllPositionsReq, GetAllTradesReq, OmsCancelOrderReq,
41    OmsCancelReply, PipelineStatusReq, SubmitOrderReq,
42};
43use ek_oms::{SubmitReply, SubmitStatus};
44use ek_station::wire::{
45    ListReq, PlugRow, RemoveReq, RemoveRes, SVC_LIST, SVC_REMOVE, SVC_START, StartReq, StartRes,
46};
47
48// 配置托底三口的 ask 面(0.6.0 环节8);id 以 ek-cfg `cfg_services()` 单源
49// (逐角取,不本地复定义字面 —— 服务名将来若改,编译/运行端对上就行)。
50// 类型面单源同在 ek-cfg:cfg.save 的 SaveReq/SaveRes ui 现时没有消费方
51// (重存段走 cfg.list 读档发 SaveReq 的是上半执面外的 cmd 线;布局待偿),
52// 0.6.3 clippy 清债撤这两个未用别名 —— 上游契约仍在,用到时再加回。
53//   cfg.list   = CfgListReq/CfgListRes(目录;零值面)
54//   cfg.delete = CfgDeleteReq/CfgDeleteRes(抹键/清段)
55use ek_cfg::wire::{
56    DeleteReq as CfgDeleteReq, DeleteRes as CfgDeleteRes, ListReq as CfgListReq,
57    ListRes as CfgListRes,
58};
59
60/// `cfg.list` / `cfg.delete` 服务 id(单源自 ek-cfg `wire::cfg_services()` 的
61/// 表位;此处取第 2/3 行)。加载期一次,不重复算。
62fn cfg_svc_ids() -> (String, String, String) {
63    let descs = ek_cfg::wire::cfg_services();
64    let id = |i: usize| -> String {
65        descs
66            .get(i)
67            .map_or_else(|| format!("cfg.svc[{i}]"), |d| d.id.clone())
68    };
69    (id(0), id(1), id(2))
70}
71
72/// 配置托底存储根(与 station 工厂 `ek_cfg::CfgPlugin::eco_or_local` 同口径:
73/// eco 成功 = 生态配置目录;失败 = 仓内 `./ekcfg-local`)。命令线程直接落库
74/// 的「重存段」與目录行共根的保证点。
75///
76/// 测试隔离:套件用 `ek_cfg::set_default_root_for_tests(根)` 把缺省根整体钉到
77/// 仓内显式根(单点在 ek-cfg `default_root()`),本函数无需另打缝。
78fn ui_cfg_store() -> ek_cfg::Store {
79    match ek_cfg::Store::eco() {
80        Ok(s) => s,
81        Err(e) => {
82            eprintln!("ek-cfg:运维修复根 eco 失败({e}),回落仓内 ./ekcfg-local(显形口径)");
83            ek_cfg::Store::with_root(std::path::PathBuf::from("ekcfg-local"))
84        }
85    }
86}
87
88/// **主题档的存储键位(0.8.0 配色双档,BD-UI-61)**:`ns="ui"` / `key="theme"`。
89///
90/// 为什么走存储根直用 `Store` 而不是总线:`cfg.list` 回包按契约**只有键名 + 形态标志,
91/// 没有值**(BD-CFG-18 安全视图),契约里没有「读值」的口 → 走总线必然破契约。
92/// 直用 `Store` 是 0.6.0 起的既有口径(BD-CFG-10 soft-centralized,`CfgResave` 同源同根)。
93/// 只有**用户手点过**切换钮才落盘(照参考站「只有点击才存 localStorage」);
94/// 运维窗「清段 ui」= 主题回缺省深色,与既有运维语义自洽。
95pub const THEME_NS: &str = "ui";
96/// 段里的键名。
97pub const THEME_KEY: &str = "theme";
98
99/// 读档(启动时一次,命令线程;选单窗在插件起来之前也用它)。
100/// 读不到(段不存在 / 键不存在 / 脏值 / 读失败)= **缺省深色**(裁定 B:现行为零变化)。
101#[must_use]
102pub fn load_theme() -> crate::tokens::Theme {
103    match ui_cfg_store().load(THEME_NS) {
104        Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map
105            .get_plain(THEME_KEY)
106            .map(crate::tokens::Theme::from_str)
107            .unwrap_or_default(),
108        Ok(ek_cfg::LoadOutcome::Fresh) => crate::tokens::Theme::default(),
109        Err(e) => {
110            eprintln!("ek-cfg:读主题档失败({e}),按缺省深色起界面");
111            crate::tokens::Theme::default()
112        }
113    }
114}
115
116/// 落盘「当前主题档」的**建图面**(载入-改;0.8.0 配色双档,BD-UI-61)。
117///
118/// 步骤 3(升级方案)口径:插件**不自写** —— 写路 = `cfg.save` 总线正路
119/// (`UiPlugin::cfg_bus_save`);载入-改的读面留直读(BD-CFG-18 安全视图
120/// 无读值口,读是装配/命令线程的合法直读面,PLAN 只铲写)。
121fn build_theme_map(store: &ek_cfg::Store, theme: crate::tokens::Theme) -> Result<ek_core::SectionMap, String> {
122    let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
123        ek_cfg::LoadOutcome::Loaded { map, .. } => map,
124        ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
125    };
126    map = map.set(THEME_KEY, ek_core::Field::plain(theme.as_str()));
127    Ok(map)
128}
129
130/// 载入-改「自动连接」勾选(0.9.0;与主题同 `ns="ui"` 一段,载入-改)。
131///
132/// - 主题键(以及段里其余键)一个不动 —— 两拨消费方共用同一档案段,谁也不许把
133///   别人的键抹掉(BD-CFG-5 单写者纪律:ui 插件是这个 ns 的唯一写手);
134/// - `auto_connect.*` 的旧键**先全清再插**:摘除/改名后的行不留魂
135///   (判据:摘除行重存档案后,ui 段里没有它的键);
136/// - `SectionMap::set` 吃 self 返回新值,循环里累计。
137fn build_auto_connect_map(
138    store: &ek_cfg::Store,
139    flags: &[(String, bool)],
140) -> Result<ek_core::SectionMap, String> {
141    let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
142        ek_cfg::LoadOutcome::Loaded { map, .. } => map,
143        ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
144    };
145    let stale: Vec<String> = map
146        .iter()
147        .filter(|(k, _)| k.starts_with("auto_connect."))
148        .map(|(k, _)| (*k).to_string())
149        .collect();
150    for k in stale {
151        map.remove(&k);
152    }
153    for (gw, on) in flags {
154        map = map.set(
155            format!("auto_connect.{gw}"),
156            ek_core::Field::plain(if *on { "1" } else { "0" }),
157        );
158    }
159    Ok(map)
160}
161
162/// 测试/显式根写口(真写面在显式注入的 Store 上发生;ui_restore 钉用)。
163/// 生产写路 = `UiPlugin::cfg_bus_save`(步骤 3 铲插件自写)。
164pub fn save_auto_connect_flags_into(
165    store: &ek_cfg::Store,
166    flags: &[(String, bool)],
167) -> Result<(), String> {
168    let map = build_auto_connect_map(store, flags)?;
169    ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
170}
171
172/// 载入-改「账户别名标签」(0.9.10,用户裁定:仿真/24小时 两套标签要能选、要能留)。
173///
174/// 与自动连接同段同纪律:载入-改,邻居键保全;
175/// `alias.*` 旧键**先全清再插** —— 摘户/清空别名后的行不留魂(空别名 = 不写键)。
176fn build_aliases_map(
177    store: &ek_cfg::Store,
178    aliases: &[(String, String)],
179) -> Result<ek_core::SectionMap, String> {
180    let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
181        ek_cfg::LoadOutcome::Loaded { map, .. } => map,
182        ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
183    };
184    let stale: Vec<String> = map
185        .iter()
186        .filter(|(k, _)| k.starts_with("alias."))
187        .map(|(k, _)| (*k).to_string())
188        .collect();
189    for k in stale {
190        map.remove(&k);
191    }
192    for (gw, alias) in aliases {
193        let alias = alias.trim();
194        if alias.is_empty() {
195            continue;
196        }
197        map = map.set(format!("alias.{gw}"), ek_core::Field::plain(alias));
198    }
199    Ok(map)
200}
201
202/// 测试/显式根写口(同 `save_auto_connect_flags_into`;生产写路 = cfg.save 正路)。
203pub fn save_account_aliases_into(
204    store: &ek_cfg::Store,
205    aliases: &[(String, String)],
206) -> Result<(), String> {
207    let map = build_aliases_map(store, aliases)?;
208    ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
209}
210
211use crate::state::{BusSnap, UiShared};
212
213/// 注册名 = 工厂名 = `"ui"`(BD-U5:两者一致)。
214pub const UI_NAME: &str = "ui";
215
216/// 界面邮箱**软顶**(0.5 M6 首发;0.7.0 背压刀改语义,数值照旧)。
217///
218/// 0.4 口径:界面是链上唯一覆写容量的插件,10 万格把"30 条/秒的行情 34 秒就丢"
219/// 变成"55 分钟"。0.5.0 起邮箱**无界** → 这个数不再是容量(到顶必丢),而是**软顶**
220/// (到顶才丢的那一格):默认一条都不丢,只有积压堆到 10 万帧才说明界面 worker
221/// 已经被彻底甩开,那时保护进程比保住行情更重要 → 开始丢新帧并把 `spilled` 涨起来。
222/// 数值取 10 万的理由不变:一帧 tick ≈ 30 条/秒 ≈ 55 分钟的量,内存上限也就是这个量级。
223/// 越阈显形(黄 6250 / 红 25000,`state::pending_thresholds`)在运维窗与 toast 双路可见。
224pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
225
226/// 交易四表的推送线(基线 `event.py:13-18`;ek-ctp 0.4.0 起发裸键,ek-core 精确匹配)。
227/// 名字单源取 ek-oms::wire —— 同一条线在链上只有一份拼写。
228const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
229const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
230const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
231const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
232
233/// 常驻观测插件(窗口面在 [`crate::runtime`] 持 [`UiShared`] 渲染)。
234pub struct UiPlugin {
235    shared: std::sync::Arc<UiShared>,
236    ctx: OnceLock<PluginContext>,
237    /// 命令线程的停线位(0.5 M5;on_stop 置真,线程自己退)。
238    cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
239}
240
241impl UiPlugin {
242    /// 与窗口面共用的共享状态句柄。
243    pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
244        Self {
245            shared,
246            ctx: OnceLock::new(),
247            cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
248        }
249    }
250
251    /// eTimer 节流回调(0.7.0:管制面快照 + 标记触发的状态播种 + 标记触发的装配显形;
252    /// 0.5 时代的「摊平状态轮询」已被第3刀退役,常态零轮询 = 判据 C13)。
253    fn refresh(&self) {
254        let Some(ctx) = self.ctx.get() else {
255            return; // 未 init 就吃到 eTimer:忽略,等下轮
256        };
257        // 0.5 M3/M5:这一拍只做三件事,而且都不问交易数据。
258        // 指令辞行搬到命令线程(`spawn_cmd_thread`);四张表吃 eOrder. / eTrade. /
259        // ePosition. / eAccount. 推送入账(基线 BaseMonitor 同形)—— 这一拍里再出现
260        // 任何「取数类 ask」就是回到 0.4 的病。
261        if self.shared.list_poll_due() {
262            self.poll_list(ctx);
263        }
264        self.seed_status(ctx);
265        if self.shared.take_pipeline_due() {
266            self.poll_pipeline(ctx);
267        }
268        if self.shared.take_cfg_due() {
269            // 配置托底段目录(0.6.0 环节8):观测面读数,轻 IO 一次(与 poll_pipeline
270            // 同档的「观测面例外」,不进周期节拍 —— 只在钮触/回执后续问触发)。
271            self.poll_cfg(ctx);
272        }
273        self.flush_persist();
274    }
275
276    /// 步骤 3 写口正路:一切落盘走 **`cfg.save` 总线**(插件不自写)。
277    /// cfg 缺席(UnknownService)= 软集中负形直写托底(BD-CFG-10,显形审计行;
278    /// 步骤 2 deps 定死后装配期 cfg 必在位,此形只罩 on_init 前的独立窗口)。
279    fn cfg_bus_save(&self, ns: &str, map: &ek_core::SectionMap) -> Result<(), String> {
280        if let Some(ctx) = self.ctx.get() {
281            match ctx.ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
282                "cfg.save",
283                ek_cfg::wire::SaveReq {
284                    ns: ns.to_string(),
285                    map: map.clone(),
286                },
287                Duration::from_millis(1500),
288            ) {
289                Ok(r) if r.ok => return Ok(()),
290                Ok(r) => return Err(r.msg),
291                Err(ek_core::error::AskError::UnknownService { .. }) => {
292                    self.shared
293                        .log_line(format!("[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"));
294                }
295                Err(e) => return Err(format!("{e:?}")),
296            }
297        }
298        ui_cfg_store().save(ns, map).map_err(|e| e.to_string())
299    }
300
301    /// 步骤 3 写口正路(清段面):走 **`cfg.delete` 总线**(段文件 + 秘库条一起清);
302    /// cfg 缺席 = `purge_account_archive` 直清负形(同 BD-CFG-10 口径)。
303    fn cfg_bus_delete(&self, ns: &str) -> Result<(), String> {
304        if let Some(ctx) = self.ctx.get() {
305            match ctx.ask_service::<ek_cfg::wire::DeleteReq, ek_cfg::wire::DeleteRes>(
306                "cfg.delete",
307                ek_cfg::wire::DeleteReq {
308                    ns: ns.to_string(),
309                    key: String::new(),
310                },
311                Duration::from_millis(1500),
312            ) {
313                Ok(r) if r.ok => return Ok(()),
314                Ok(r) => return Err(r.msg),
315                Err(ek_core::error::AskError::UnknownService { .. }) => {
316                    self.shared
317                        .log_line(format!("[cfg] cfg 缺席,{ns} 直清托底(软集中负形)"));
318                }
319                Err(e) => return Err(format!("{e:?}")),
320            }
321        }
322        crate::appconfig::purge_ns_direct(&ui_cfg_store(), ns)
323    }
324
325    /// **落盘拍(0.9.10 收口,用户实锤「删了账户确认了、重启又显示」的第二条根因)**:
326    /// 四类用户手点落盘(theme / 档案抹除 / 自动连接 / 别名)从 eTimer 拍抽出成
327    /// 单口 —— **eTimer(1s 拍)与 `on_stop`(收场)都吃这一口**。
328    /// 旧口径只在 1s 拍消费:删除后 ≤1s 内关窗 = 待办没消费 = 档案没抹 = 重启复活
329    /// (摘行到关窗常在 1s 内 = race 常态,不是偶发)。收场 flush 封死这个窗口。
330    fn flush_persist(&self) {
331        if self.shared.take_theme_persist_due() {
332            // 0.8.0 配色双档:用户手点切换钮后的那次落盘(渲染线程零文件 IO,写在这拍)。
333            // 步骤 3 正路:载入-改建图 → **cfg.save 总线写**。
334            let theme = self.shared.theme();
335            let wr = build_theme_map(&ui_cfg_store(), theme)
336                .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
337            match wr {
338                Ok(()) => self
339                    .shared
340                    .log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
341                Err(e) => self
342                    .shared
343                    .log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
344            }
345        }
346        let purges = self.shared.take_archive_purges();
347        for gw in purges {
348            // 0.9.10 用户实锤「界面删了的账户,下次启动又显示」:摘行只摘了
349            // 内存清册,档案段 `ctp.{gw}` 还在 = 启动清点扫段 = 原样复活。
350            // 摘行必须连档案一起抹(与 write_account 对称;渲染线程零文件 IO,
351            // 落手在这拍)。步骤 3 正路:cfg.delete 总线清段。
352            match self.cfg_bus_delete(&format!("ctp.{gw}")) {
353                Ok(()) => self.shared.log_line(format!(
354                    "[账户] 档案段 ctp.{gw} 已抹(删除已落盘:下次启动清点不再复活)"
355                )),
356                Err(e) => self.shared.log_line(format!(
357                    "[账户] 档案 ctp.{gw} 抹除失败:{e}(本次删除只在今次运行生效,重启会复活)"
358                )),
359            }
360        }
361        if self.shared.take_auto_connect_persist_due() {
362            // 0.9.0 自动连接勾选:同一支纪律(手点才写,渲染线程零文件 IO)。
363            // ui 段**载入-改-存**:主题键必须保住;摘了户口的行要清键(不留魂)。
364            let flags: Vec<(String, bool)> = self
365                .shared
366                .accounts()
367                .iter()
368                .map(|a| (a.gateway.clone(), a.auto_connect))
369                .collect();
370            // 步骤 3 正路:载入-改建图 → cfg.save 总线写。
371            let wr = build_auto_connect_map(&ui_cfg_store(), &flags)
372                .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
373            match wr {
374                Ok(()) => {
375                    let on: Vec<&str> = flags
376                        .iter()
377                        .filter(|(_, on)| *on)
378                        .map(|(gw, _)| gw.as_str())
379                        .collect();
380                    if on.is_empty() {
381                        self.shared
382                            .log_line("[ui] 自动连接勾选已存(ui 段;当前全部关)".to_string());
383                    } else {
384                        self.shared.log_line(format!(
385                            "[ui] 自动连接勾选已存(ui 段;开:{};守护/桌面启动清点按它起连)",
386                            on.join(", ")
387                        ));
388                    }
389                }
390                Err(e) => self.shared.log_line(format!(
391                    "[ui] 自动连接勾选落盘失败:{e}(本轮勾选只在本次运行内生效)"
392                )),
393            }
394        }
395        if self.shared.take_alias_persist_due() {
396            // 0.9.10 别名标签:保存/改名/摘行触发的落盘(同自动连接纪律)。
397            let aliases: Vec<(String, String)> = self
398                .shared
399                .accounts()
400                .iter()
401                .map(|a| (a.gateway.clone(), a.alias.clone()))
402                .collect();
403            // 步骤 3 正路:载入-改建图 → cfg.save 总线写。
404            let wr = build_aliases_map(&ui_cfg_store(), &aliases)
405                .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
406            match wr {
407                Ok(()) => {
408                    let named: Vec<String> = aliases
409                        .iter()
410                        .filter(|(_, al)| !al.trim().is_empty())
411                        .map(|(gw, al)| format!("{al}({gw})"))
412                        .collect();
413                    if named.is_empty() {
414                        self.shared
415                            .log_line("[ui] 账户别名已存(ui 段;当前全部无别名)".to_string());
416                    } else {
417                        self.shared.log_line(format!("[ui] 账户别名已存(ui 段;{};启动清点按它贴标签)", named.join(", ")));
418                    }
419                }
420                Err(e) => self.shared.log_line(format!(
421                    "[ui] 账户别名落盘失败:{e}(本次改名只在本次运行内生效)"
422                )),
423            }
424        }
425    }
426
427    /// 配置托底段目录(0.6.0;`cfg.list` 零值面,BD-CFG-17/18)。
428    fn poll_cfg(&self, ctx: &PluginContext) {
429        let (_, svc_list, _) = cfg_svc_ids();
430        match ctx.ask_service::<CfgListReq, CfgListRes>(
431            svc_list.as_str(),
432            CfgListReq,
433            Duration::from_secs(2),
434        ) {
435            Ok(res) => {
436                self.shared.set_cfg_briefs(res.entries);
437                if !res.msg.is_empty() {
438                    self.shared
439                        .log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
440                }
441            }
442            Err(e) => self
443                .shared
444                .log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
445        }
446    }
447
448    /// 管制面快照(`svctl.list`):谁在场 + 服务目录 + 邮箱溢出计数。
449    ///
450    /// 0.5 M3 降到每 `LIST_POLL_TICKS` 拍一次。它给的不是交易数据而是「宿主长什么样」
451    /// 的观测面 —— 基线里的对应物是 `get_all_gateway_names` 的同进程直读,不是推送,
452    /// 所以它留在白名单里(「界面不周期取数」这条纪律的唯一例外)。
453    fn poll_list(&self, ctx: &PluginContext) {
454        match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
455            Ok(rows) => {
456                let mut services = Vec::with_capacity(rows.len());
457                for r in &rows {
458                    let ids = ctx
459                        .services_of(&r.name)
460                        .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
461                        .unwrap_or_default();
462                    services.push((r.name.clone(), ids));
463                }
464                // 账户在场位回填 + 「实例没了就作废会话读数」。
465                // 0.5:「新装配」不再由这里检测 —— ePlugUp. 已订阅,即时且不靠轮询。
466                self.shared.sync_accounts_connected(&rows);
467                // 0.4.5(P7d):邮箱溢出的当场显形。运维窗那一列一直有 spilled,
468                // 但 ek-core 的 stderr 诊断每 128 条才打一条 —— 屏上此前没有任何东西。
469                self.shared.note_plugin_spill(&rows);
470                self.shared.set_snapshot(BusSnap { rows, services });
471            }
472            Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
473        }
474    }
475
476    /// 一次性状态播种(0.7.0 第3刀:周期轮询退役)。
477    ///
478    /// 常态**零轮询**:登录态/断供读数全靠 `eStatus.` 推送到货。这里只处理
479    /// 「网关刚上线(或刚重连)」的待播种队列 —— 每网关运行期至多一次 ask,
480    /// 拿初始快照(推送是变化触发,刚上线还没有变化,界面得有一份起点)。
481    /// 只读口不写日志(ek-ctp D-STATUS ④),播种不冲 eLog 环。
482    fn seed_status(&self, ctx: &PluginContext) {
483        for gw in self.shared.take_status_seeds() {
484            let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
485            match ctx.ask_service::<StatusReq, GatewayStatus>(
486                &svc,
487                StatusReq,
488                Duration::from_secs(2),
489            ) {
490                Ok(st) => self.shared.note_gateway_status(&gw, &st),
491                Err(e) => {
492                    // 打 Display(不是 Debug):超时行自带「目标积压 N 帧」(ek-core 0.5.0)。
493                    // **不排回队列重试**:常态零轮询是这一刀的判据(C13),问不到就留在
494                    // 「无证据」态等推送 —— `eStatus.` 一到,读数立刻归位。
495                    self.shared.note_status_unreachable(&gw, &format!("{e}"));
496                }
497            }
498        }
499    }
500
501    /// 装配显形(0.5 M3 改事件触发):`oms.pipeline_status` 只在被标记的那一拍问一次。
502    ///
503    /// 标记来自启动、`ePlugUp.oms` 与运维窗的刷新钮。它此前是每 4 拍一次的轮询 ——
504    /// 装配很少变,而它是 oms worker 上又一次串行 ask;在事件驱动模型里这种轮询是多余的。
505    fn poll_pipeline(&self, ctx: &PluginContext) {
506        match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
507            ek_oms::SVC_PIPELINE_STATUS,
508            PipelineStatusReq,
509            Duration::from_secs(2),
510        ) {
511            Ok(lines) => self.shared.set_pipeline_lines(lines),
512            Err(e) => {
513                // 不每拍刷日志:oms 不在场是合法态(清单里没选它,ask 走 TargetGone
514                // 快错)。只在「从有到无」那一次显形一条,与状态轮询「只读口不冲 eLog 环」
515                // 同口径。读数清空 → 运维窗那一节整节不画。
516                if !self.shared.pipeline_lines().is_empty() {
517                    self.shared
518                        .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
519                }
520                self.shared.set_pipeline_lines(Vec::new());
521            }
522        }
523    }
524
525    /// 启动一次四问(0.5 M4;基线无对应物 —— 插件热插拔是 ek 独有场景)。
526    ///
527    /// 一次性,不是周期:之后四张表只吃推送。走 `note_*` 而不是 `seed_trade_snap`,
528    /// 是为了**合并而不是覆盖** —— 网关可能在 oms 还没消化完之前就已经把委托推给了
529    /// 界面,覆盖式播种会把界面已经看见的那一笔抹掉。
530    ///
531    /// 执行位置 = 命令线程(0.5 执行核查改的):原先它挂在 `on_start` 里,而 `on_start`
532    /// 是宿主启动序列的一部分 —— 四口最坏 4 × 2 秒会把后面插件的启动一起拖住;挂在
533    /// 入账 worker 上又违反「入账 worker 里没有任何 ask」。所以它跟着命令线程走。
534    fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
535        let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
536            "oms.get_all_orders",
537            GetAllOrdersReq,
538            Duration::from_secs(2),
539        );
540        let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
541            "oms.get_all_trades",
542            GetAllTradesReq,
543            Duration::from_secs(2),
544        );
545        let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
546            "oms.get_all_positions",
547            GetAllPositionsReq,
548            Duration::from_secs(2),
549        );
550        let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
551            "oms.get_all_accounts",
552            GetAllAccountsReq,
553            Duration::from_secs(2),
554        );
555        let mut n = 0usize;
556        if let Ok(rows) = &orders {
557            for o in rows {
558                shared.note_order(o);
559            }
560            n += rows.len();
561        }
562        if let Ok(rows) = &trades {
563            for t in rows {
564                shared.note_trade(t);
565            }
566            n += rows.len();
567        }
568        if let Ok(rows) = &positions {
569            for p in rows {
570                shared.note_position(p);
571            }
572            n += rows.len();
573        }
574        if let Ok(rows) = &funds {
575            for a in rows {
576                shared.note_funds(a);
577            }
578            n += rows.len();
579        }
580        // 只在真播种到东西时落一条;空清单(刚装配、柜台还没回报)不刷日志。
581        if n > 0 {
582            let bad = usize::from(orders.is_err())
583                + usize::from(trades.is_err())
584                + usize::from(positions.is_err())
585                + usize::from(funds.is_err());
586            shared.log_line(format!(
587                "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
588                orders.as_ref().map_or(0, |v| v.len()),
589                trades.as_ref().map_or(0, |v| v.len()),
590                positions.as_ref().map_or(0, |v| v.len()),
591                funds.as_ref().map_or(0, |v| v.len()),
592                bad
593            ));
594        }
595    }
596
597    /// 命令线程(0.5 M5):点击即发,不等 eTimer 拍,也不占 ui 的入账 worker。
598    ///
599    /// 基线对应物是「UI 主线程直调 `main_engine.send_order`,立即返回」(engine.py:915-923)。
600    /// ek-core 的 ask 是同步的,所以把「等回包」这件事从入账线程挪走:入账线程从此只
601    /// 做「收事件 → 写账本」,每事件 O(1),永不被后端拖住。
602    /// 这一条线程也是「会等的东西」唯一允许待的地方:启动一次四问在这里跑
603    /// (见 `reconcile_trade`),既不占宿主的启动序列,也不占入账 worker。
604    fn spawn_cmd_thread(&self) {
605        let Some(ctx) = self.ctx.get().cloned() else {
606            return; // 未 init:没有管制面可用
607        };
608        let shared = std::sync::Arc::clone(&self.shared);
609        let stop = std::sync::Arc::clone(&self.cmd_stop);
610        if std::thread::Builder::new()
611            .name("ui-cmd".to_string())
612            .spawn(move || {
613                loop {
614                    if stop.load(std::sync::atomic::Ordering::Acquire) {
615                        break;
616                    }
617                    // 用户指令优先于播种:播种最坏 4 × 2 秒,不能把点击排到它后面。
618                    if shared.cmd_pending() {
619                        Self::run_cmds(&shared, &ctx);
620                        continue;
621                    }
622                    // 播种档:启动一次 + 之后每次网关 / oms 上线各一次(标记是消费性的)。
623                    if shared.take_trade_reseed() {
624                        Self::reconcile_trade(&shared, &ctx);
625                        continue;
626                    }
627                    // K线历史补数(I2 第一刀;开窗入队,这里发 tq 桥)。
628                    if shared.kline_fetch_pending() {
629                        Self::run_kline_fetches(&shared, &ctx);
630                        continue;
631                    }
632                    std::thread::sleep(Duration::from_millis(20));
633                }
634            })
635            .is_err()
636        {
637            // 线程起不来 = 指令只会进队列不会执行。这一条必须上屏,不能静默降级。
638            self.shared
639                .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
640        }
641    }
642
643    /// K线历史补数执行体(I2 第一刀;命令线程):
644    /// 入队单 → 找 tq 派生网关 → `{gw}.klines`(1min×300,Length 写法)→ 回包并格。
645    /// error/truncated/ask 失手 = 日志显形(不静默),cell 只吃 error=None 的数据。
646    fn run_kline_fetches(shared: &UiShared, ctx: &PluginContext) {
647        for f in shared.take_kline_fetches() {
648            let Some(gw) = ctx
649                .plugins()
650                .iter()
651                .map(|p| p.name.clone())
652                .find(|n| n.starts_with("tq"))
653            else {
654                shared.log_line(format!(
655                    "[kline] {} tq 网关不在线,历史补数丢弃(重开 K线窗再试)",
656                    f.symbol
657                ));
658                continue;
659            };
660            let req = ek_tq::KlinesReq {
661                symbol: f.symbol.clone(),
662                raw: false,
663                interval_secs: f.interval_secs,
664                range: ek_tq::RangeSpec::Length { count: 300, end_ns: None },
665                max_rows: Some(300),
666            };
667            match ctx.ask::<ek_tq::KlinesReq, ek_tq::KlinesReply>(
668                &gw,
669                req,
670                Duration::from_secs(45),
671            ) {
672                Ok(reply) => {
673                    if let Some(err) = &reply.error {
674                        shared.log_line(format!(
675                            "[kline] {}/{}s 历史显形: {}",
676                            f.symbol, f.interval_secs, err.message
677                        ));
678                        continue;
679                    }
680                    let n = reply.rows.len();
681                    shared.market().note_history(&f.symbol, f.interval_secs, &reply.rows);
682                    shared.bump();
683                    shared.log_line(format!(
684                        "[kline] {}/{}s 历史补数 +{n} 根(mode={} truncated={})",
685                        f.symbol,
686                        f.interval_secs,
687                        reply.mode.as_str(),
688                        reply.truncated
689                    ));
690                }
691                Err(e) => shared.log_line(format!(
692                    "[kline] {}/{}s 历史 ask 失手: {e:?}",
693                    f.symbol, f.interval_secs
694                )),
695            }
696        }
697    }
698
699    /// 辞行指令队列(0.5 M5:由命令线程调用,不在 ui worker 上;回执进日志,面板语义可断言)。
700    fn run_cmds(shared: &UiShared, ctx: &PluginContext) {        for cmd in shared.drain_cmds() {
701            match cmd {
702                crate::state::UiCmd::Start { factory, name } => {
703                    match ctx.ask_service::<StartReq, StartRes>(
704                        SVC_START,
705                        StartReq {
706                            factory: factory.clone(),
707                            name: name.clone(),
708                        },
709                        Duration::from_secs(3),
710                    ) {
711                        Ok(StartRes::Started { name }) => {
712                            shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
713                        }
714                        Ok(StartRes::Rejected { reason }) => {
715                            shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
716                        }
717                        Err(e) => {
718                            shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
719                        }
720                    }
721                    // 起停回执之后重问一次管制面:运维窗那一行与账户在场位立刻动。
722                    shared.request_list_refresh();
723                }
724                crate::state::UiCmd::Remove { name } => {
725                    // 观测面自身的闸(0.4.3 BD-UI-64,审核 P1;0.5 M5 换了理由,闸不换):
726                    // 0.4 的理由是「这条指令在 ui 自己的 worker 上执行 → 摘 ui = 自己等自己
727                    // (ek-core `close_one` 只能走到 2 秒超时的 detach 分支)」。
728                    // 0.5 执行者换成命令线程,「自己等自己」这一半不成立了,但结论一样:
729                    // 摘掉 ui 之后没有 eTimer 拍再跑快照与状态轮询、入账 worker 也没了,
730                    // 窗口还开着而里面全是上一帧。所以仍然直拒,不发 svctl。
731                    // 管制面 `svctl` 的同一道闸长在 ek-station supervisor 侧(0.2.11)。
732                    if name == UI_NAME {
733                        shared.log_line(
734                            "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
735                                .to_string(),
736                        );
737                        continue;
738                    }
739                    match ctx.ask_service::<RemoveReq, RemoveRes>(
740                        SVC_REMOVE,
741                        RemoveReq { name: name.clone() },
742                        Duration::from_secs(3),
743                    ) {
744                        Ok(RemoveRes::Removed { name }) => {
745                            shared.log_line(format!("[ops] svctl.remove {name} 完成"));
746                        }
747                        Ok(RemoveRes::NoSuchName { name }) => {
748                            shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
749                        }
750                        // 0.4.3:回执枚举增项(ek-station 0.2.11)—— 名字在场但不可摘,
751                        // 或宿主另有错。原先这一档被塞进 NoSuchName 的 name 栏(错误文本
752                        // 当名字打),这一栏现在照请求回填,错因只进 reason。
753                        Ok(RemoveRes::Rejected { name, reason }) => {
754                            shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
755                        }
756                        Err(e) => {
757                            shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
758                        }
759                    }
760                    shared.request_list_refresh();
761                }
762                crate::state::UiCmd::Reload {
763                    interval,
764                    symbol,
765                    bars,
766                } => {
767                    // 回灌走 dm.reload ask(push=true 逐条广播 → 行情板实时见回执面)。
768                    match ctx.ask_service::<ReloadReq, usize>(
769                        "dm.reload",
770                        ReloadReq {
771                            interval,
772                            symbol: symbol.clone(),
773                            bars: bars.clone(),
774                            size: Some(bars.len()),
775                            push: true,
776                        },
777                        Duration::from_secs(5),
778                    ) {
779                        Ok(n) => {
780                            shared.market().note_reload();
781                            shared.log_line(format!(
782                                "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
783                            ));
784                        }
785                        Err(e) => {
786                            shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
787                        }
788                    }
789                }
790                crate::state::UiCmd::SendOrder { req, gateway, mode } => {
791                    // 0.1.1 BD-UI-30 薄下单者:五段管线全走 oms.submit_order
792                    // (拆单/风控/逐腿发送/记账;直发 {gw}.send_order 旁路口删除,
793                    // 守卫不可绕过)。parent 聚合逐腿显形进日志。
794                    match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
795                        "oms.submit_order",
796                        SubmitOrderReq {
797                            req: req.clone(),
798                            gateway_name: gateway.clone(),
799                            mode,
800                        },
801                        Duration::from_secs(10),
802                    ) {
803                        Ok(reply) => {
804                            let tag = match reply.status {
805                                SubmitStatus::Sent => "SENT",
806                                SubmitStatus::Partial => "PARTIAL",
807                                SubmitStatus::Rejected => "REJECTED",
808                            };
809                            for leg in &reply.legs {
810                                match (leg.ok, &leg.en_orderid, &leg.reason) {
811                                    (true, Some(id), _) => shared.log_line(format!(
812                                        "[ops] submit {} {} {}x{} → {id} (parent {})",
813                                        leg.req.symbol,
814                                        leg.req.direction.as_str(),
815                                        leg.req.offset.as_str(),
816                                        leg.req.volume,
817                                        reply.parent_id
818                                    )),
819                                    _ => shared.log_line(format!(
820                                        "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
821                                        leg.req.symbol,
822                                        leg.req.direction.as_str(),
823                                        leg.req.offset.as_str(),
824                                        leg.req.volume,
825                                        leg.reason.as_deref().unwrap_or("?"),
826                                        reply.parent_id
827                                    )),
828                                }
829                            }
830                            if reply.legs.is_empty() {
831                                shared.log_line(format!(
832                                    "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
833                                    req.symbol,
834                                    req.offset.as_str(),
835                                    req.volume,
836                                    reply.parent_id
837                                ));
838                            }
839                        }
840                        Err(e) => {
841                            shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
842                        }
843                    }
844                }
845                crate::state::UiCmd::CancelOrder { orderid, gateway } => {
846                    // 0.1.1 BD-UI-30:守卫撤单(尝试 ≤4000 在 oms;直发
847                    // {gw}.cancel_order 旁路口删除)。en_orderid 组式与
848                    // OrderData.en_orderid() 同式。
849                    match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
850                        "oms.cancel_order",
851                        OmsCancelOrderReq {
852                            en_orderid: format!("{gateway}.{orderid}"),
853                            gateway_name: gateway.clone(),
854                        },
855                        Duration::from_secs(10),
856                    ) {
857                        Ok(OmsCancelReply { ok: true, msg }) => {
858                            shared.log_line(format!("[ops] {msg}"));
859                        }
860                        Ok(OmsCancelReply { ok: false, msg }) => {
861                            // 0.11.0 第6刀 R7:屏幕上给人话,机器码原文照留(取证要吃原文)
862                            shared
863                                .log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
864                        }
865                        Err(e) => {
866                            shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
867                        }
868                    }
869                }
870                crate::state::UiCmd::CtpConfigure { gateway, form } => {
871                    // 定盘口径:界面手工输入 → 插件自带配置口(插件不读文件)。
872                    // 多账号面:服务名随网关走(`{gw}.configure`),不再硬编码 ctp。
873                    let svc = format!("{gateway}.configure");
874                    match ctx.ask_service::<ConfigureReq, ConfigureReply>(
875                        svc.as_str(),
876                        form.to_configure_req(),
877                        Duration::from_secs(5),
878                    ) {
879                        // `..`:ConfigureReply 0.6.0 起带 product_classes(实际生效值)。
880                        // 这里不回填表单勾选 —— 被拒时回带的是「网关当前仍在跑」的那一份,
881                        // 用它覆盖用户刚点的勾选会把他的意图吃掉;日志里网关已写明收录产品类。
882                        Ok(ConfigureReply { ok: true, msg, .. }) => {
883                            shared.log_line(format!("[ops] {svc} 受理:{msg}"));
884                        }
885                        Ok(ConfigureReply { ok: false, msg, .. }) => {
886                            shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
887                        }
888                        Err(e) => {
889                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
890                        }
891                    }
892                }
893                crate::state::UiCmd::Connect { gateway, form } => {
894                    // **四态分离的「连接」**(规划裁定「添加 ≠ 连接」):一次做完
895                    // `svctl.start(factory=ctp)` + `{gw}.configure`,结果回写账户行
896                    // (`Assembling → Assembled / AssembleFailed` + 失败原因留显形)。
897                    // 失败**不回滚清册行** —— 行留着给『重试』(旧口径"必须先移除才能重试"作废)。
898                    shared.note_connect_sent(&gateway);
899                    // 单源纪律在此裁决:非行情源账户的 md 位就地清空,**留档与下发同一份**
900                    // (旧口径只清下发不清留档,编辑页回填会带着一根用不上的行情前置)。
901                    let form = shared.prepare_connect_form(&gateway, form);
902                    let started = match ctx.ask_service::<StartReq, StartRes>(
903                        SVC_START,
904                        StartReq {
905                            factory: "ctp".to_string(),
906                            name: gateway.clone(),
907                        },
908                        Duration::from_secs(3),
909                    ) {
910                        Ok(StartRes::Started { name }) => {
911                            shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
912                            true
913                        }
914                        Ok(StartRes::Rejected { reason }) => {
915                            // 防重入(规划 Q7):实例已在场时重复点『连接』/『重试』不是失败,
916                            // 重投凭据即可(否则会把正在跑的账户误判成"装配失败")。
917                            if shared.account(&gateway).is_some_and(|a| a.connected) {
918                                shared
919                                    .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
920                                true
921                            } else {
922                                shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
923                                shared.note_connect_result(&gateway, Err(reason));
924                                continue;
925                            }
926                        }
927                        Err(e) => {
928                            shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
929                            shared.note_connect_result(&gateway, Err(format!("{e:?}")));
930                            continue;
931                        }
932                    };
933                    if !started {
934                        continue;
935                    }
936                    let svc = format!("{gateway}.configure");
937                    match ctx.ask_service::<ConfigureReq, ConfigureReply>(
938                        svc.as_str(),
939                        form.to_configure_req(),
940                        Duration::from_secs(5),
941                    ) {
942                        // `..`:ConfigureReply 带的 product_classes(实际生效值)不回填勾选
943                        // (被拒时回带的是「网关仍在跑」的那一份,覆盖用户刚点的勾选会吃掉意图);
944                        // 0.2.18 起把它显示出来(见 state::note_effective_product_classes)。
945                        Ok(ConfigureReply {
946                            ok: true,
947                            msg,
948                            product_classes,
949                        }) => {
950                            shared.log_line(format!("[ops] {svc} 受理:{msg}"));
951                            shared.note_effective_product_classes(&gateway, &product_classes);
952                            shared.note_connect_result(&gateway, Ok(()));
953                        }
954                        Ok(ConfigureReply { ok: false, msg, .. }) => {
955                            shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
956                            shared.note_connect_result(&gateway, Err(msg));
957                        }
958                        Err(e) => {
959                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
960                            shared.note_connect_result(&gateway, Err(format!("{e:?}")));
961                        }
962                    }
963                }
964                crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
965                    // 多账号面:一律路由到行情元账号(渲染侧已按此入列);
966                    // 纯 TD 账号收到订阅会**静默成功**(do_subscribe 恒 ok:true),
967                    // 所以路由正确性靠界面纪律,不靠网关报错(规划 §9.4-21)。
968                    let svc = format!("{gateway}.subscribe");
969                    match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
970                        svc.as_str(),
971                        CtpSubscribeReq {
972                            symbols: symbols.clone(),
973                        },
974                        Duration::from_secs(5),
975                    ) {
976                        Ok(CtpSubscribeReply { ok: true, msg }) => {
977                            shared.note_subscriptions(&symbols);
978                            shared.log_line(format!(
979                                "[ops] {svc} {} 受理:{}",
980                                symbols.join(","),
981                                msg
982                            ));
983                        }
984                        Ok(CtpSubscribeReply { ok: false, msg }) => {
985                            shared.log_line(format!(
986                                "[ops] {svc} {} 被拒:{}",
987                                symbols.join(","),
988                                msg
989                            ));
990                        }
991                        Err(e) => {
992                            shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
993                        }
994                    }
995                }
996                crate::state::UiCmd::Disconnect { gateway } => {
997                    // 0.2.19 规划裁定 1:『断开』= `{gw}.disconnect`(停会话),
998                    // 『移除』= `svctl.remove`(停实例)—— 两条走两条不同的口,不共用。
999                    // 发令即把该行标成「已断开」:用户按了钮就要立刻看到,不等下一轮轮询。
1000                    let svc = format!("{gateway}.disconnect");
1001                    match ctx.ask_service::<DisconnectReq, DisconnectReply>(
1002                        svc.as_str(),
1003                        DisconnectReq,
1004                        Duration::from_secs(5),
1005                    ) {
1006                        Ok(DisconnectReply { ok: true, msg }) => {
1007                            shared.mark_account_stopped(&gateway);
1008                            shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
1009                        }
1010                        Ok(DisconnectReply { ok: false, msg }) => {
1011                            shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
1012                        }
1013                        Err(e) => {
1014                            shared.log_line(format!(
1015                                "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
1016                            ));
1017                        }
1018                    }
1019                }
1020                crate::state::UiCmd::CfgList => {
1021                    // 目录刷新(命令线程执行档;与 worker poll_cfg 同一读数面)。
1022                    let (_, svc_list, _) = cfg_svc_ids();
1023                    match ctx.ask_service::<CfgListReq, CfgListRes>(
1024                        svc_list.as_str(),
1025                        CfgListReq,
1026                        Duration::from_secs(2),
1027                    ) {
1028                        Ok(res) => {
1029                            shared.set_cfg_briefs(res.entries);
1030                            if !res.msg.is_empty() {
1031                                shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
1032                            }
1033                        }
1034                        Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
1035                    }
1036                }
1037                crate::state::UiCmd::CfgDelete { ns } => {
1038                    // 清段(BD-CFG-20 运维语义;空 key = 整段;零值面)。
1039                    let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
1040                    let _ = (svc_save, svc_list);
1041                    match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
1042                        svc_delete.as_str(),
1043                        CfgDeleteReq {
1044                            ns: ns.clone(),
1045                            key: String::new(),
1046                        },
1047                        Duration::from_secs(2),
1048                    ) {
1049                        Ok(res) if res.ok => {
1050                            shared.log_line(format!(
1051                                "[cfg] 清段 {ns}:键 {} / 秘条 {}",
1052                                res.removed_keys, res.removed_secrets
1053                            ));
1054                            shared.request_cfg_refresh();
1055                        }
1056                        Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
1057                        Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
1058                    }
1059                    shared.arm_cfg_delete(None);
1060                }
1061                crate::state::UiCmd::CfgResave { ns } => {
1062                    // 重存段(0.6.0):明面按现状重读 → serialize → store。**值不变
1063                    // (BD-CFG-15:不输出 = 不持久,这里连「不重排空行」都不做之外的
1064                    // 改动一律没有)**;rev/格式修复口。兜底根与 cfg 实例同口径
1065                    // (BD-CFG-10 soft-centralized:直调用库,不拉起注册)。
1066                    let store = ui_cfg_store();
1067                    let loaded = store
1068                        .load(&ns)
1069                        .map_err(|e| e.to_string())
1070                        .and_then(|o| match o {
1071                            ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
1072                            map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
1073                        });
1074                    match loaded {
1075                        Ok(ek_cfg::LoadOutcome::Fresh) => {
1076                            unreachable!("Fresh 已被上面的 and_then 挡走");
1077                        }
1078                        Ok(ek_cfg::LoadOutcome::Loaded { map, warnings }) => {
1079                            if !warnings.is_empty() {
1080                                shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
1081                            }
1082                            // 步骤 3 正路:重存写 = cfg.save 总线(run_cmds 处 ctx 在手;
1083                            // cfg 缺席 = 直写负形托底,BD-CFG-10)。
1084                            let wr = match ctx.ask_service::<
1085                                ek_cfg::wire::SaveReq,
1086                                ek_cfg::wire::SaveRes,
1087                            >(
1088                                "cfg.save",
1089                                ek_cfg::wire::SaveReq {
1090                                    ns: ns.clone(),
1091                                    map: map.clone(),
1092                                },
1093                                Duration::from_millis(1500),
1094                            ) {
1095                                Ok(r) if r.ok => Ok(()),
1096                                Ok(r) => Err(r.msg),
1097                                Err(ek_core::error::AskError::UnknownService { .. }) => {
1098                                    shared.log_line(format!(
1099                                        "[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"
1100                                    ));
1101                                    ek_cfg::cfg_save_at(store.root(), &ns, &map).map_err(|e| e.to_string())
1102                                }
1103                                Err(e) => Err(format!("{e:?}")),
1104                            };
1105                            match wr {
1106                                Ok(()) => {
1107                                    shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
1108                                }
1109                                Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
1110                            }
1111                        }
1112                        Err(msg) => shared.log_line(format!("[cfg] {msg}")),
1113                    }
1114                }
1115            }
1116        }
1117    }
1118}
1119
1120impl Plugin for UiPlugin {
1121    fn name(&self) -> &str {
1122        UI_NAME
1123    }
1124
1125    fn deps(&self) -> &[&str] {
1126        // 步骤 2(升级方案,用户裁定):cfg = 核心插件,**全插件必依赖**,
1127        // 宿主 deps 拓扑序保证 cfg 最先激活(激活序定死,不可改)。
1128        &["cfg"]
1129    }
1130
1131    fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
1132        vec![
1133            std::borrow::Cow::Borrowed(EVENT_LOG),
1134            std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
1135            std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
1136            std::borrow::Cow::Borrowed(EVENT_BAR),
1137            std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
1138            std::borrow::Cow::Borrowed(EVENT_TICK),
1139            std::borrow::Cow::Borrowed(EVENT_CONTRACT),
1140            std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
1141            // 0.5 M1 交易四表的推送线(基线 BaseMonitor 各自订阅自己的线)。
1142            // 不订 eTradeSync.:ek-oms 处理完成交后重推的那一条(plugin.rs:305-308),
1143            // 与 ek-ctp 的 eTrade. 是同一笔成交的两份 —— 订两条就是重复入账。
1144            std::borrow::Cow::Borrowed(W_ORDER),
1145            std::borrow::Cow::Borrowed(W_TRADE),
1146            std::borrow::Cow::Borrowed(W_POSITION),
1147            std::borrow::Cow::Borrowed(W_ACCOUNT),
1148            std::borrow::Cow::Borrowed(EVENT_TIMER),
1149            // 0.7.0 第3刀:状态变化推送(裸线,载荷自带网关名)—— 界面零轮询的地基
1150            std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
1151        ]
1152    }
1153
1154    fn mailbox_soft_cap(&self) -> u64 {
1155        UI_MAILBOX_SOFT_CAP
1156    }
1157
1158    fn version(&self) -> std::borrow::Cow<'static, str> {
1159        std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
1160    }
1161
1162    fn services(&self) -> Vec<ServiceDesc> {
1163        Vec::new() // 纯视图零服务;管制面在 svctl
1164    }
1165
1166    fn on_init(&self, ctx: &PluginContext) {
1167        let _ = self.ctx.set(ctx.clone());
1168    }
1169
1170    fn on_start(&self) {
1171        self.shared.log_line("[ui] 界面插件上线");
1172        // 0.5 M5 + M4:命令线程先起(点击即发);启动一次四问在那条线程里跑 ——
1173        // `on_start` 是 `engine.start()` 启动链的一环,四口最坏 4 × 2 秒不许把后面
1174        // 插件的启动一起拖住,更不许占入账 worker。
1175        self.spawn_cmd_thread();
1176        // 启动一次四问的标记(on_start 是 `engine.start()` 启动链的一环,四口最坏
1177        // 4 × 2 秒不许把后面插件的启动一起拖住,更不许占入账 worker)。
1178        self.shared.request_trade_reseed();
1179        self.shared.request_pipeline_refresh();
1180        // 0.8.0 配色双档:启动读一次档(命令线程,一次小文件读),用户没点过钮 = 缺省深色。
1181        let theme = load_theme();
1182        self.shared.set_theme(theme);
1183        if theme == crate::tokens::Theme::Light {
1184            self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
1185        }
1186    }
1187
1188    fn on_stop(&self) {
1189        self.shared.log_line("[ui] 界面插件下线");
1190        // 命令线程停线:它在 20ms 空转切片自查这个位,不会卡在 sleep 里。
1191        self.cmd_stop
1192            .store(true, std::sync::atomic::Ordering::Release);
1193        // **收场 flush(0.9.10,用户实锤「删了账户、确认了,重启又显示」的第二条根因)**:
1194        // 落盘待办过去只在 1s eTimer 拍消费 —— 删除后 ≤1s 内关窗 = 待办没消费 =
1195        // 档案没抹 = 重启复活。收场把四类待办冲干净,封死 race 窗口。
1196        self.flush_persist();
1197    }
1198
1199    fn on_mail(&self, event: &ek_event::Event) {
1200        let ty = event.event_type.as_str();
1201        if ty == EVENT_LOG {
1202            if let Some(d) = event.data.downcast_ref::<LogData>() {
1203                let tag = if d.is_warning_or_above() {
1204                    "WARN"
1205                } else {
1206                    "INFO"
1207                };
1208                self.shared
1209                    .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
1210                // 0.5.0(用户定「链进行的状态改为事件日志推送」):`[链] ` 前缀的行除了进日志窗,
1211                // 还落进账户行的「阶段」栏(ek-ctp 0.9.8 D-QUERY-GATE ⑧)。这一条是纯显示落账,
1212                // 不参与 `apply_status` 那套在线/断供判据。
1213                // 前缀单源:`crate::state::CHAIN_STATE_PREFIX` = ek-ctp 0.9.8 那一份的 re-export
1214                // (state.rs:201;0.6.3 路标兑现,注释同步 —— 无本地字面)。
1215                if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
1216                    self.shared.note_chain(&d.gateway_name, note);
1217                }
1218            } else {
1219                self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
1220            }
1221            return;
1222        }
1223        if ty == ek_ctp::EVENT_STATUS_PREFIX {
1224            // 0.7.0 第3刀:状态变化推送 → 一次到货三件事(文案 / 状态列 / 去重告警)。
1225            if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
1226                self.shared.note_status_event(ev);
1227            } else {
1228                self.shared
1229                    .log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
1230            }
1231            return;
1232        }
1233        if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
1234            if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
1235                let verb = if ty == EVENT_PLUG_UP {
1236                    "上线"
1237                } else {
1238                    "下线"
1239                };
1240                self.shared
1241                    .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
1242                // 在场位与运维窗那一行跟着这条线动(不等 4 拍的快照节拍)。
1243                self.shared.request_list_refresh();
1244                if ty == EVENT_PLUG_UP {
1245                    // 0.5 M3:装配显形改吃这条线 —— 不再靠每拍 list 去发现「谁新上线」。
1246                    if n.name == ek_oms::NAME {
1247                        self.shared.request_pipeline_refresh();
1248                    }
1249                    // 0.5 执行核查补的第三处:网关 / oms 上线之后重播种一次四张表。
1250                    // 摘除一个网关账户会作废它名下的委托与成交行,而 oms 那边那些行还在;
1251                    // 重新添加时不重问一次,界面就永久少一段历史。顺带覆盖「ui 先于 oms
1252                    // 起来、第一次播种问空了」这一序。
1253                    let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
1254                    if n.name == ek_oms::NAME || is_account {
1255                        self.shared.request_trade_reseed();
1256                    }
1257                    // 0.7.0 第3刀:网关上线 → 播种一次初始快照(运行期至多这一发 ask;
1258                    // 之后登录态/断供全靠 `eStatus.` 推送,界面零轮询)。
1259                    if is_account {
1260                        self.shared.request_status_seed(&n.name);
1261                    }
1262                    // 0.5 M3:订阅重放同样改吃这条线(与 `sync_accounts_connected` 的
1263                    // newly 回带是同一件事的两条路,留一条:这条即时、那条要等快照拍)。
1264                    let md = self.shared.md_source();
1265                    let subs = self.shared.subscriptions();
1266                    if !md.is_empty() && n.name == md && !subs.is_empty() {
1267                        self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
1268                            gateway: n.name.clone(),
1269                            symbols: subs.clone(),
1270                        });
1271                        self.shared.log_line(format!(
1272                            "[ops] 行情源 {} 上线,重放订阅 {} 项",
1273                            n.name,
1274                            subs.len()
1275                        ));
1276                    }
1277                } else {
1278                    // 0.5 M1:网关下线 = 它名下的委托 / 成交 / 持仓 / 资金行全部作废。
1279                    // 推送模型里没人再告诉界面「这一行已经不存在了」,不清就是替一个
1280                    // 已经不存在的会话说话(与 0.2.19「实例不在场就作废会话读数」同口径)。
1281                    let removed = self.shared.clear_gateway_rows(&n.name);
1282                    if removed > 0 {
1283                        self.shared.log_line(format!(
1284                            "[ui] 插件下线 {}:作废 {} 行交易数据",
1285                            n.name, removed
1286                        ));
1287                    }
1288                    // 0.5 执行核查补的第二处:摘掉 oms 之后管线读数不会自己清空。
1289                    // 0.4 是每 4 拍问一次,4 秒内自己就空了;改事件触发后没人再问,
1290                    // 运维窗那一节会替一个已经不存在的 oms 说话。标记一次:下一拍问一次
1291                    // → TargetGone 快错 → `poll_pipeline` 的 Err 分支清空并落一条日志。
1292                    if n.name == ek_oms::NAME {
1293                        self.shared.request_pipeline_refresh();
1294                    }
1295                }
1296            }
1297            self.shared.bump();
1298            return;
1299        }
1300        if ty == EVENT_BAR {
1301            if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
1302                self.shared.market().on_frame(f);
1303            }
1304            self.shared.bump();
1305            return;
1306        }
1307        if ty == EVENT_BAR_UPDATE {
1308            if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
1309                self.shared.market().on_update(b);
1310            }
1311            self.shared.bump();
1312            return;
1313        }
1314        if ty == EVENT_TICK {
1315            if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
1316                self.shared.update_tick(t);
1317            }
1318            return;
1319        }
1320        if ty == EVENT_CONTRACT {
1321            // 0.11.0 第4刀(ek-ctp R5):合约流分批到货,一帧 ≤2000 行 → 逐行上屏
1322            // (合约窗照旧一段段长出来,只是每一拍长一批);单条形继续受理(兼容)。
1323            if let Some(v) = event
1324                .data
1325                .downcast_ref::<Vec<ek_data::contract::ContractData>>()
1326            {
1327                for c in v {
1328                    self.shared.update_contract(c);
1329                }
1330            } else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
1331                self.shared.update_contract(c);
1332            }
1333            return;
1334        }
1335        if ty == ek_ctp::EVENT_CONTRACT_FINISH {
1336            let gw = event
1337                .data
1338                .downcast_ref::<String>()
1339                .cloned()
1340                .unwrap_or_else(|| "?".to_string());
1341            self.shared
1342                .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
1343            return;
1344        }
1345        // 0.5 M1 交易四表:后台只把事件推过来,界面自己入账(基线 BaseMonitor.process_event)。
1346        // 这里不 ask 任何东西,所以入账 worker 永不被后端拖住。
1347        if ty == W_ORDER {
1348            match event.data.downcast_ref::<OrderData>() {
1349                Some(o) => self.shared.note_order(o),
1350                None => self.shared.note_bad_payload(W_ORDER),
1351            }
1352            return;
1353        }
1354        if ty == W_TRADE {
1355            match event.data.downcast_ref::<TradeData>() {
1356                Some(t) => self.shared.note_trade(t),
1357                None => self.shared.note_bad_payload(W_TRADE),
1358            }
1359            return;
1360        }
1361        if ty == W_POSITION {
1362            match event.data.downcast_ref::<PositionData>() {
1363                Some(p) => self.shared.note_position(p),
1364                None => self.shared.note_bad_payload(W_POSITION),
1365            }
1366            return;
1367        }
1368        if ty == W_ACCOUNT {
1369            match event.data.downcast_ref::<AccountData>() {
1370                Some(a) => self.shared.note_funds(a),
1371                None => self.shared.note_bad_payload(W_ACCOUNT),
1372            }
1373            return;
1374        }
1375        if ty == EVENT_TIMER {
1376            self.refresh();
1377        }
1378    }
1379}