ek-ui 0.9.3

//! S2 验收:svctl 操作闭环真链(ui 窗口按钮 → UiCmd 队列 → eTimer 辞行 →
//! svctl ask → 日志/快照)+ 状态机单元(指令 / 二次确认 / 隔离判定)。

use std::sync::Arc;
use std::time::{Duration, Instant};

use ek_core::main_engine::MainEngine;
use ek_core::service::PlugNotice;
use ek_event::EVENT_TIMER;
use ek_event::Event;
use ek_station::{StationConfig, boot};
use ek_ui::state::{ConfirmStep, UiCmd, UiShared, is_quarantined};

fn wait_until(tag: &str, secs: u64, f: impl Fn() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(secs);
    while Instant::now() < deadline {
        if f() {
            return;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    panic!("wait_until timeout: {tag}");
}

// ---- L1:状态机 ----

#[test]
fn u0_cmd_queue_roundtrip() {
    let s = UiShared::new();
    s.push_cmd(UiCmd::Start {
        factory: "dm".to_string(),
        name: String::new(),
    });
    s.push_cmd(UiCmd::Remove {
        name: "dicy".to_string(),
    });
    assert_eq!(
        s.drain_cmds(),
        vec![
            UiCmd::Start {
                factory: "dm".to_string(),
                name: String::new()
            },
            UiCmd::Remove {
                name: "dicy".to_string()
            },
        ]
    );
    assert!(s.drain_cmds().is_empty(), "drain 是消费性取走");
}

#[test]
fn u1_confirm_steps_and_cancel() {
    let s = UiShared::new();
    assert_eq!(s.confirm_step("a"), ConfirmStep::Armed);
    assert_eq!(s.pending_confirm().as_deref(), Some("a"));
    assert_eq!(s.confirm_step("a"), ConfirmStep::Fired);
    assert_eq!(s.pending_confirm(), None);
    assert_eq!(s.confirm_step("b"), ConfirmStep::Armed);
    assert_eq!(s.confirm_step("c"), ConfirmStep::Switched);
    assert_eq!(s.pending_confirm().as_deref(), Some("c"));
    s.confirm_cancel();
    assert_eq!(s.pending_confirm(), None);
    assert_eq!(s.confirm_step("b"), ConfirmStep::Armed); // 取消后重新武装
}

#[test]
fn u2_quarantined_detection() {
    assert!(is_quarantined("Quarantined"));
    assert!(is_quarantined("State::Quarantined(panic x3)"));
    assert!(!is_quarantined("Running"));
    assert!(!is_quarantined("Ready"));
}

// ==== 0.6.0(BD-CFG 环节8):配置托底三口 ====

#[test]
fn u5_cfg_readouts_and_delete_arm() {
    let s = UiShared::new();
    // 目录:`None` = 从未问过(渲染为「未问过」句,不是空表)。
    assert_eq!(s.cfg_briefs(), None);
    // 待问标记单次消费。
    s.request_cfg_refresh();
    assert!(s.take_cfg_due());
    assert!(!s.take_cfg_due(), "consume = take once is cleared");
    // 零值面读Data line(NsBrief 只有 key/形态;BD-CFG-17/18)。
    s.set_cfg_briefs(vec![ek_cfg::wire::NsBrief {
        ns: "oms".to_owned(),
        rev: 12,
        keys: vec![ek_cfg::wire::KeyBrief {
            key: "rate_limit".to_owned(),
            secret: false,
        }],
    }]);
    let b = s.cfg_briefs().unwrap();
    assert_eq!(b.len(), 1);
    assert_eq!(b[0].rev, 12);
    assert!(!b[0].keys[0].secret);
    // 清段二次确认臂(独立臂,不与 svctl 的 ConfirmStep 互扰)。
    assert_eq!(s.cfg_delete_armed(), None);
    s.arm_cfg_delete(Some("oms".to_owned()));
    assert_eq!(s.cfg_delete_armed().as_deref(), Some("oms"));
    s.arm_cfg_delete(None);
    assert_eq!(s.cfg_delete_armed(), None);
}

#[test]
fn t6_cfg_list_real_chain() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec![
                "cfg".to_string(),
                "dm".to_string(),
                "oms".to_string(),
                "ui".to_string(),
            ],
        },
    )
    .unwrap();
    engine.start().unwrap();
    // cfg 收件壳在场(ui_kinds 的 default_kinds 首行;0.6.0 起带上)。
    wait_until("cfg row", 4, || {
        shared.snapshot().rows.iter().any(|r| r.name == "cfg")
    });
    // 运维窗点名 → 目录回写(真链 ask;零值面,不再写明文)。
    shared.push_cmd(UiCmd::CfgList);
    put_timer(&engine);
    wait_until("cfg briefs", 4, || shared.cfg_briefs().is_some());
    engine.close_all(Duration::from_secs(2));
}

/// 0.5 M3 三处降档的节拍钉(纯状态,不靠真链拍序 —— 真链里 ek-event 的 1 秒定时线程
/// 与手动 `put(eTimer)` 会交错,「第几拍」在集成套里判不稳)。
///
/// - `svctl.list`:每 `LIST_POLL_TICKS` = 4 拍一次(观测面,不是交易数据)。
/// - `{gw}.status`:不再是「每 2 拍全问一遍」,而是「每拍最多问 2 个、环形推进」——
///   8 个账户 4 拍走完一轮,单拍 ask 上限不随账户数线性长。
/// - `oms.pipeline_status`:不再是节拍,是标记(消费性,两次标记合并成一次问)。
#[test]
fn u3_list_cadence_status_flattening_and_pipeline_flag() {
    let s = UiShared::new();
    let due: Vec<bool> = (0..9).map(|_| s.list_poll_due()).collect();
    assert_eq!(
        due,
        vec![true, false, false, true, false, false, false, true, false],
        "管制面快照 = 第一拍必问(界面不该空着等 4 秒),之后每 4 拍(第 4、第 8 拍)"
    );

    // 标记档:插件上/下线、svctl 起停之后,下一拍必须重问 —— 0.4 是每拍一次,
    // 降到 4 拍如果不补这一档,「点了移除、行 4 秒不动」就成了常态。
    s.request_list_refresh();
    assert!(s.list_poll_due(), "被标记的那一拍立刻重问");
    assert!(!s.list_poll_due(), "标记是消费性的,不是一档状态");

    let present: Vec<String> = (0..8).map(|i| format!("gw{i}")).collect();
    let mut asked: Vec<String> = Vec::new();
    for _ in 0..8 {
        let batch = s.status_batch(&present);
        assert!(
            batch.len() <= ek_ui::state::STATUS_ASKS_PER_TICK,
            "单拍问的账户数不许超过 STATUS_ASKS_PER_TICK,实测 {}",
            batch.len()
        );
        asked.extend(batch);
    }
    for gw in &present {
        assert!(
            asked.iter().filter(|a| *a == gw).count() >= 2,
            "8 拍里每个账户至少被问到一轮: {gw} 实测 {asked:?}"
        );
    }
    assert!(s.status_batch(&[]).is_empty(), "没有账户就不问");

    // 去重(0.5 执行核查补的洞):账户清册允许两行共用同一个网关(第二行 start 回
    // 「已在场」而行仍在清册里),而 `{gw}.status` 是按网关寻址的同一个服务口。
    let dup = vec!["gwA".to_string(), "gwA".to_string(), "gwB".to_string()];
    let b = s.status_batch(&dup);
    assert!(
        b.len() <= ek_ui::state::STATUS_ASKS_PER_TICK,
        "去重不放宽单拍上限,实测 {b:?}"
    );
    assert_eq!(
        b.iter().filter(|g| *g == "gwA").count(),
        1,
        "同一拍里同一个网关只许问一次,实测 {b:?}"
    );

    assert!(!s.take_pipeline_due(), "没标记就不问");
    s.request_pipeline_refresh();
    assert!(s.take_pipeline_due(), "标记一次就问一次");
    assert!(!s.take_pipeline_due(), "标记是消费性的,不是一档状态");

    // 播种档(0.5 执行核查补的第三处):同一形状的标记,消费方是命令线程。
    assert!(!s.take_trade_reseed(), "没标记就不播种");
    s.request_trade_reseed();
    assert!(s.take_trade_reseed(), "标记一次就播种一次");
    assert!(!s.take_trade_reseed(), "标记是消费性的,不是一档状态");
    s.request_pipeline_refresh();
    s.request_pipeline_refresh();
    assert!(s.take_pipeline_due());
    assert!(!s.take_pipeline_due(), "两次标记不变成两次问(合并成一次)");
}

// ---- 集成:cmd → eTimer 辞行 → svctl 真链 ----

fn put_timer(engine: &MainEngine) {
    engine.event_engine().put(Event::new(EVENT_TIMER)).unwrap();
}

#[test]
fn t0_remove_then_restart_loop() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec!["dm".to_string(), "oms".to_string(), "ui".to_string()],
        },
    )
    .unwrap();
    engine.start().unwrap();

    // 快照先落(等 eTimer 真实一拍)。
    wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
    assert!(shared.snapshot().rows.iter().any(|r| r.name == "dm"));

    // 摘除 dm:push → 人工拍 eTimer → 行消失 + 业务回执日志。
    shared.push_cmd(UiCmd::Remove {
        name: "dm".to_string(),
    });
    put_timer(&engine);
    wait_until("dm removed", 4, || {
        shared.snapshot().rows.iter().all(|r| r.name != "dm")
    });
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("[ops] svctl.remove dm 完成")),
        "remove log missing"
    );

    // 重注入 dm:单步直发 → 行回来。
    shared.push_cmd(UiCmd::Start {
        factory: "dm".to_string(),
        name: String::new(),
    });
    put_timer(&engine);
    wait_until("dm back", 4, || {
        shared
            .snapshot()
            .rows
            .iter()
            .any(|r| r.name == "dm" && r.state == "Running")
    });
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("svctl.start dm → dm 上线")),
        "start log missing"
    );

    engine.close_all(Duration::from_secs(2));
}

#[test]
fn t1_unknown_factory_rejected_logged() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec!["dm".to_string(), "oms".to_string(), "ui".to_string()],
        },
    )
    .unwrap();
    engine.start().unwrap();

    wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
    shared.push_cmd(UiCmd::Start {
        factory: "ghost".to_string(),
        name: String::new(),
    });
    put_timer(&engine);
    wait_until("rejected log", 4, || {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("start ghost 被拒"))
    });

    engine.close_all(Duration::from_secs(2));
}

/// 自身不可摘(0.4.3 BD-UI-64,审核 P1)—— 真链钉,两条闸各在两侧。
///
/// `ui`:本 crate 的 `plugin.rs` 直拒,连 ask 都不发(摘掉 ui 之后没有 eTimer 拍再跑
/// 快照与状态轮询、入账 worker 也没了,窗口还在但里面全是上一帧。0.4 的理由是「指令
/// 在 ui 自己的 worker 上执行 = 自己等自己」;0.5 M5 执行者换成命令线程,理由换了、闸不换)。
/// `svctl`:闸在 ek-station 0.2.11 的 `supervisor.rs::remove_by`,回带走
/// `RemoveRes::Rejected{name, reason}`,本 crate 打「被拒」。
/// 判据 = 摘令入列 + 等到回执之后:那一行仍在场,日志给的是拒绝不是「完成」。
#[test]
fn t2_self_removal_refused_on_both_faces() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec!["dm".to_string(), "oms".to_string(), "ui".to_string()],
        },
    )
    .unwrap();
    engine.start().unwrap();
    wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
    assert!(
        shared.snapshot().rows.iter().any(|r| r.name == "svctl"),
        "svctl 应在场(它是 svctl.* 的提供方)"
    );

    // ① ui:本 crate 侧直拒 —— 不发 ask,行仍在,日志给理由。
    shared.push_cmd(UiCmd::Remove {
        name: "ui".to_string(),
    });
    put_timer(&engine);
    wait_until("ui refusal logged", 4, || {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("拒绝移除 ui"))
    });
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .all(|l| !l.contains("svctl.remove ui 完成")),
        "不得真摘 ui"
    );

    // ② svctl:闸在 ek-station 侧 —— 回执 Rejected,本 crate 打「被拒」。
    shared.push_cmd(UiCmd::Remove {
        name: "svctl".to_string(),
    });
    put_timer(&engine);
    wait_until("svctl refusal logged", 4, || {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("remove svctl 被拒"))
    });

    // ③ 两拍之后快照仍在:两条自身行都还在场,启停面与观测面都活着。
    put_timer(&engine);
    wait_until("both faces still listed", 4, || {
        let rows = &shared.snapshot().rows;
        rows.iter().any(|r| r.name == "svctl") && rows.iter().any(|r| r.name == "ui")
    });
    assert!(
        shared
            .snapshot()
            .rows
            .iter()
            .all(|r| r.name != "dm" || r.state == "Running"),
        "dm 不该被顺手摘掉"
    );

    engine.close_all(Duration::from_secs(2));
}

/// 装配显形真链(0.4.4,P5-a):`oms.pipeline_status` 从总线进界面状态。
///
/// 这条服务在 0.4.3 之前是全链零消费方(ek-oms 0.3.x 加它时是「可观测性补偿」,
/// 界面没有出口,只在 on_start 日志里显形一次)。这一钉钉的是**内容**:
/// 工厂注入的正是 `ek_station::default_kinds()` 里 `standard_gates(&RiskLimits::default())`
/// 那四道闸 —— 界面现在必须能看见它们,而不是只能翻启动日志。
/// (节拍本身在 u3 钉,不在这里:见下面的注释。)
#[test]
fn t3_pipeline_status_reaches_ui_state() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec!["dm".to_string(), "oms".to_string(), "ui".to_string()],
        },
    )
    .unwrap();
    engine.start().unwrap();

    wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);
    // 节拍不在这里判:`engine.start()` 同时起了 ek-event 那条 1 秒定时线程,
    // 真拍与我手动 put 的拍交错,「第几拍」在这里判不稳(第一版就是被这条绊倒的)。
    // 节拍本身由下面 u3 在 UiShared 上钉死(确定性、无引擎、无竞态)。
    put_timer(&engine);
    wait_until("pipeline lines", 6, || !shared.pipeline_lines().is_empty());
    let lines = shared.pipeline_lines();
    assert!(
        lines
            .iter()
            .any(|l| l == "stage:validate|split|risk|send|book"),
        "五段管线没显形: {lines:?}"
    );
    assert!(
        lines.iter().any(|l| l.starts_with("floor:field|rate(")),
        "限速/撤单守卫没显形: {lines:?}"
    );
    assert!(
        lines.iter().any(|l| l == "splitter:ek-converter"),
        "拆单器没显形: {lines:?}"
    );
    for gate in [
        "gate:ek-risk:contract",
        "gate:ek-risk:volume",
        "gate:ek-risk:position-limit",
        "gate:ek-risk:price-band",
    ] {
        assert!(
            lines.iter().any(|l| l == gate),
            "工厂注入的风控闸 {gate} 没显形: {lines:?}"
        );
    }

    // 摘掉 oms 之后管线读数必须自己清空(0.5 执行核查补的第二处):问口改事件触发之后
    // 没人再周期问它,不补这一条,运维窗那一节会一直显示旧读数 —— 替一个已经不存在的
    // oms 说话。判据走真链:摘令 → 行没了 → 标记的那一拍问一次 → TargetGone → 清空。
    shared.push_cmd(UiCmd::Remove {
        name: "oms".to_string(),
    });
    put_timer(&engine);
    wait_until("oms gone from rows", 6, || {
        !shared.snapshot().rows.iter().any(|r| r.name == "oms")
    });
    put_timer(&engine);
    wait_until("pipeline lines cleared", 6, || {
        shared.pipeline_lines().is_empty()
    });
    assert!(
        shared.pipeline_lines().is_empty(),
        "摘掉 oms 之后管线读数必须清空,实测 {:?}",
        shared.pipeline_lines()
    );

    engine.close_all(Duration::from_secs(2));
}

// ================ 0.4.5(P7a / P7b / P7d)的判据 ================

fn spill_row(name: &str, spilled: u64) -> ek_station::wire::PlugRow {
    water_row(name, spilled, 0, 0, ek_core::plugin::MAILBOX_SOFT_CAP)
}

/// 0.7.0 背压刀:带水位的行(积压/峰值/该插件自己的软顶随快照带来)。
fn water_row(
    name: &str,
    spilled: u64,
    pending: u64,
    peak: u64,
    soft_cap: u64,
) -> ek_station::wire::PlugRow {
    ek_station::wire::PlugRow {
        name: name.to_string(),
        version: String::new(),
        state: "Running".to_string(),
        deps: Vec::new(),
        spilled,
        panics: 0,
        pending,
        peak,
        soft_cap,
    }
}

// ---- 0.5 M1/M2:四本 keyed 账的语义(基线 BaseMonitor 的列可变性) ----

fn order_row(
    gw: &str,
    id: &str,
    status: ek_data::constant::Status,
    volume: f64,
    price: f64,
) -> ek_data::order::OrderData {
    ek_data::order::OrderData {
        gateway_name: gw.to_string(),
        symbol: "rb2605".to_string(),
        exchange: ek_data::constant::Exchange::SHFE,
        orderid: id.to_string(),
        direction: Some(ek_data::constant::Direction::LONG),
        offset: ek_data::constant::Offset::OPEN,
        price,
        volume,
        status,
        name: "螺纹钢2605".to_string(),
        ..ek_data::order::OrderData::default()
    }
}

fn trade_row(id: &str) -> ek_data::trade::TradeData {
    ek_data::trade::TradeData {
        gateway_name: "ctp".to_string(),
        symbol: "rb2605".to_string(),
        exchange: ek_data::constant::Exchange::SHFE,
        orderid: "A1".to_string(),
        tradeid: id.to_string(),
        direction: Some(ek_data::constant::Direction::LONG),
        offset: ek_data::constant::Offset::OPEN,
        price: 102.5,
        volume: 1.0,
        ..ek_data::trade::TradeData::default()
    }
}

fn position_row(gw: &str, symbol: &str, volume: f64) -> ek_data::position::PositionData {
    ek_data::position::PositionData {
        gateway_name: gw.to_string(),
        symbol: symbol.to_string(),
        exchange: ek_data::constant::Exchange::SHFE,
        direction: ek_data::constant::Direction::LONG,
        volume,
        can_use_volume: volume,
        ..ek_data::position::PositionData::default()
    }
}

fn account_row(gw: &str, acct: &str, balance: f64) -> ek_data::account::AccountData {
    ek_data::account::AccountData {
        gateway_name: gw.to_string(),
        accountid: acct.to_string(),
        balance,
        ..ek_data::account::AccountData::default()
    }
}

/// u4(0.5 M2):委托行按 `en_orderid` 归并,而且**只有可变列**随回报走。
///
/// 基线 `BaseCell.__init__` 先置 `is_update = True`(插入时全列都写),再按表头把
/// `update: False` 的列改回 False;`set_content` 里 `if self.is_update` 才 `setText`
/// (`widget.py:89-98 / 112-113`)。所以 CTP 后续回报里给空的 `volume / price` 从来
/// 没把已经显示过的「总量」抹成 0。整行覆盖就是这个病的 Rust 版本 —— 这一钉挡的就是它。
#[test]
fn u4_order_book_merges_by_en_orderid_and_keeps_immutable_columns() {
    let s = UiShared::new();
    s.note_order(&order_row(
        "ctp",
        "A1",
        ek_data::constant::Status::SUBMITTING,
        10.0,
        102.5,
    ));
    // 后续回报:状态 / 已成 / 成交均价走;不可变列故意给 0(CTP 部分回报就是这个形)。
    let mut later = order_row("ctp", "A1", ek_data::constant::Status::PARTTRADED, 0.0, 0.0);
    later.traded = 4.0;
    later.traded_price = 102.5;
    s.note_order(&later);

    let ts = s.trade_snap();
    assert_eq!(ts.orders.len(), 1, "同 en_orderid 是一条行,不是两条");
    let o = &ts.orders[0];
    assert_eq!(
        o.status,
        ek_data::constant::Status::PARTTRADED,
        "状态列可变"
    );
    assert_eq!(o.traded, 4.0, "已成列可变");
    assert_eq!(o.traded_price, 102.5, "成交均价列可变");
    assert_eq!(o.volume, 10.0, "总量列不可变:回报给 0 也不许写进表");
    assert_eq!(o.price, 102.5, "委托价列不可变");
    assert_eq!(
        o.direction,
        Some(ek_data::constant::Direction::LONG),
        "方向列不可变"
    );
    assert_eq!(o.name, "螺纹钢2605", "名称列可变(基线 update True)");
}

/// u4b(0.5 M2):行序 = 新事件在前(基线 `insert_new_row` 是 `insertRow(0)`);
/// 而**播种**不是「一批新事件」——`seed_trade_snap` 喂进来的顺序原样保留。
#[test]
fn u4b_newest_event_is_the_first_row_but_seeding_keeps_order() {
    let s = UiShared::new();
    s.note_order(&order_row(
        "ctp",
        "A1",
        ek_data::constant::Status::SUBMITTING,
        1.0,
        1.0,
    ));
    s.note_order(&order_row(
        "ctp",
        "A2",
        ek_data::constant::Status::SUBMITTING,
        2.0,
        2.0,
    ));
    let ts = s.trade_snap();
    assert_eq!(ts.orders.len(), 2);
    assert_eq!(ts.orders[0].orderid, "A2", "后到的排在前(基线顶部插入)");

    s.seed_trade_snap(ek_ui::state::TradeSnap {
        orders: vec![
            order_row("ctp", "B1", ek_data::constant::Status::SUBMITTING, 1.0, 1.0),
            order_row("ctp", "B2", ek_data::constant::Status::SUBMITTING, 2.0, 2.0),
        ],
        trades: Vec::new(),
        positions: Vec::new(),
        accounts: Vec::new(),
    });
    let ts = s.trade_snap();
    assert_eq!(
        ts.orders
            .iter()
            .map(|o| o.orderid.clone())
            .collect::<Vec<_>>(),
        vec!["B1".to_string(), "B2".to_string()],
        "播种保序:那是「已有的一份账」,不是一批新事件"
    );
}

/// u4c(0.5 M1):按网关清账只清那一个网关。
#[test]
fn u4c_clear_gateway_rows_voids_only_that_gateway() {
    let s = UiShared::new();
    s.note_order(&order_row(
        "gwA",
        "A1",
        ek_data::constant::Status::SUBMITTING,
        1.0,
        1.0,
    ));
    s.note_order(&order_row(
        "gwB",
        "B1",
        ek_data::constant::Status::SUBMITTING,
        1.0,
        1.0,
    ));
    s.note_position(&position_row("gwA", "rb2605", 3.0));
    s.note_funds(&account_row("gwA", "220380", 100.0));
    assert_eq!(s.clear_gateway_rows("gwA"), 3, "委托 + 持仓 + 资金三行");
    let ts = s.trade_snap();
    assert_eq!(ts.orders.len(), 1);
    assert_eq!(ts.orders[0].gateway_name, "gwB");
    assert!(ts.positions.is_empty() && ts.accounts.is_empty());
    assert_eq!(s.clear_gateway_rows("gwZ"), 0, "没这个名字就什么都不动");
}

/// u4d(0.5 偏离基线③):成交账本触顶丢最旧,而且只在第一次弹一次告警。
#[test]
fn u4d_trade_book_caps_at_trade_cap_and_alerts_once() {
    let s = UiShared::new();
    for i in 0..(ek_ui::state::TRADE_CAP + 5) {
        s.note_trade(&trade_row(&format!("T{i}")));
    }
    let ts = s.trade_snap();
    assert_eq!(ts.trades.len(), ek_ui::state::TRADE_CAP, "触顶就丢最旧");
    assert_eq!(
        ts.trades[0].tradeid,
        format!("T{}", ek_ui::state::TRADE_CAP + 4),
        "最新的一笔仍在最前"
    );
    assert_eq!(s.trade_dropped(), 5);
    assert_eq!(s.pending_alert_count(), 1, "触顶只弹一次,不每笔弹");
}

/// u4e(0.5 M2):持仓的 `volume` 是可变列 —— 平仓写 0 留行,不删行(基线 hideRow)。
#[test]
fn u4e_position_volume_is_mutable_and_the_row_survives_flattening() {
    let s = UiShared::new();
    s.note_position(&position_row("ctp", "rb2605", 5.0));
    s.note_position(&position_row("ctp", "rb2605", 0.0));
    let ts = s.trade_snap();
    assert_eq!(ts.positions.len(), 1, "同 en_positionid 归并成一条");
    assert_eq!(ts.positions[0].volume, 0.0, "volume 可变:平仓写 0,不删行");
}

/// u4f(0.5 M1):载荷类型不对不静默丢 —— 计数,并在第一次显形一条日志。
#[test]
fn u4f_bad_payload_is_counted_not_silently_dropped() {
    let s = UiShared::new();
    s.note_bad_payload("eOrder.");
    s.note_bad_payload("eOrder.");
    assert_eq!(s.bad_payload_count("eOrder."), 2);
    assert_eq!(s.bad_payload_count("eTrade."), 0, "逐线各记各的");
    assert!(
        s.logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("eOrder. 载荷类型异常")),
        "第一次必须显形(静默丢 = 界面少了一路数据而没人知道)"
    );
}

/// u4g(0.5 裁定):成交按 `en_tradeid` 去重,推送与播种同一条规则。
///
/// 口径来自基线引擎层 `BaseTradeEngine.trades_filter_dict`(`engine.py:1072-1080`,注释写明
/// 防的是账户重连时柜台重推同一笔成交),与 `ek-oms` 的 `trades_filter` 同键同口径。
/// 实连演示里量出来的洞:摘账户再重加之后,元账号 3 行成交变 4 行、纯 TD 账号 3 行变 6 行。
#[test]
fn u4g_trade_dedups_by_en_tradeid_across_push_and_seed() {
    let s = UiShared::new();
    // 同一笔推两次只落一行(去重集在账本之前把关)。
    s.note_trade(&trade_row("T1"));
    s.note_trade(&trade_row("T1"));
    assert_eq!(s.trade_snap().trades.len(), 1, "同一笔成交只算一次");

    // 播种(启动 / 上线重问)与推送走同一条规则:已见过的笔不再落一行。
    s.note_trade(&trade_row("T1"));
    assert_eq!(
        s.trade_snap().trades.len(),
        1,
        "播种与推送同规则,实测 {:?}",
        s.trade_snap()
            .trades
            .iter()
            .map(|t| t.tradeid.clone())
            .collect::<Vec<_>>()
    );

    // 没见过的笔照常入账,新在前。
    s.note_trade(&trade_row("T2"));
    let ids: Vec<_> = s
        .trade_snap()
        .trades
        .iter()
        .map(|t| t.tradeid.clone())
        .collect();
    assert_eq!(ids, vec!["T2", "T1"], "新成交在前");

    // 键带网关前缀(`en_tradeid` = `{网关}.{成交号}`):同号成交跨网关是两笔,不许互相吃掉。
    let mut other = trade_row("T1");
    other.gateway_name = "ctp_2".to_string();
    s.note_trade(&other);
    assert_eq!(s.trade_snap().trades.len(), 3, "跨网关同号成交是两笔");

    // 摘网关:去重集跟着账本清,那个网关之后再推同一笔还能上屏。
    s.clear_gateway_rows("ctp");
    assert_eq!(s.trade_snap().trades.len(), 1, "只清 ctp 名下");
    s.note_trade(&trade_row("T1"));
    assert_eq!(s.trade_snap().trades.len(), 2, "清掉的笔要能重新入账");
}

/// u4h:去重集不许比账本大 —— 触顶被挤掉的那一行,它的键同时出集。
#[test]
fn u4h_trade_cap_evicts_the_dedup_key_too() {
    let s = UiShared::new();
    for i in 0..=ek_ui::state::TRADE_CAP {
        s.note_trade(&trade_row(&format!("T{i}")));
    }
    assert_eq!(
        s.trade_snap().trades.len(),
        ek_ui::state::TRADE_CAP,
        "账本封顶"
    );
    assert_eq!(s.trade_dropped(), 1, "最旧那笔被挤掉");
    // 被挤掉的正是最先喂进来的 T0:它现在不在去重集里,再推一次还能上屏。
    s.note_trade(&trade_row("T0"));
    let snap = s.trade_snap();
    assert_eq!(snap.trades.len(), ek_ui::state::TRADE_CAP, "封顶不变");
    assert_eq!(snap.trades[0].tradeid, "T0", "挤掉的笔再推还能上屏");
}

/// u4i(0.5.0,用户定「链进行的状态改为事件日志推送」):`[链] ` 状态行落进账户行的「阶段」栏。
///
/// 三条判据:① 只有带前缀的行动这一栏,而且它**只是显示** —— `apply_status` 那套在线/断供判据
/// 不许读它;② 没有对应账户行(网关还没进清册)时不炸也不记;③ 前缀与列形状是单源/形状钉。
#[test]
fn u4i_chain_note_lands_only_from_chain_lines() {
    let s = UiShared::new();
    let form = ek_ui::state::CtpForm {
        broker_id: "9999".to_string(),
        td_user: "278153".to_string(),
        td_addr: "tcp://127.0.0.1:30001".to_string(),
        ..ek_ui::state::CtpForm::default()
    };
    let acct = ek_ui::state::AccountRow::derive(&form).expect("夹具账户必须可派生");
    let gw = acct.gateway.clone();
    s.add_account(acct).expect("夹具账户必须入列");
    assert_eq!(
        s.account(&gw).unwrap().chain_note,
        None,
        "没推过状态行 = None(显示「—」)"
    );

    // 网关还没进清册:状态行先到(装配与推送的先后本来不固定)不炸、不记。
    s.note_chain("gw_absent", "前置已连接");
    assert!(s.account(&gw).unwrap().chain_note.is_none());

    s.note_chain(&gw, "前置已连接");
    assert_eq!(
        s.account(&gw).unwrap().chain_note.as_deref(),
        Some("前置已连接")
    );
    s.note_chain(&gw, "合约未到手,本轮查询已停止");
    assert_eq!(
        s.account(&gw).unwrap().chain_note.as_deref(),
        Some("合约未到手,本轮查询已停止"),
        "「阶段」栏跟着最近一条状态行走"
    );

    // 生命周期两处(自查补的洞):① 新一轮『连接』入列那一刻,上一轮的链状态不再是这一轮的事实
    // (与登录态 / 状态细节同一条口径);② 实例不在场(被摘 / 宿主崩了)一并作废 —— 留着就是替
    // 一个不存在的会话说话。
    s.note_connect_sent(&gw);
    assert!(
        s.account(&gw).unwrap().chain_note.is_none(),
        "『连接』入列必须清掉上一轮的链状态"
    );
    s.note_chain(&gw, "资金/持仓轮询中");
    s.sync_accounts_connected(&[]);
    assert!(
        s.account(&gw).unwrap().chain_note.is_none(),
        "实例不在场还留着「资金/持仓轮询中」= 替一个不存在的会话说话"
    );
    // 实例回来(重新装配)后推送照落 —— 作废不是单向锁死
    s.sync_accounts_connected(&[spill_row(&gw, 0)]);
    s.note_chain(&gw, "结算单已确认");
    assert_eq!(
        s.account(&gw).unwrap().chain_note.as_deref(),
        Some("结算单已确认"),
        "重新装配后链状态照落"
    );

    // 前缀形状钉:界面这一份必须与 ek-ctp 0.9.8 的 `CHAIN_STATE_PREFIX` 同字面量
    // (抬 pin 到 0.9.8 后改引用网关那一份,回到单源 —— 挂账见 DESIGN.md)。
    assert_eq!(ek_ui::state::CHAIN_STATE_PREFIX, "[链] ");

    let state_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/state.rs"))
        .expect("读 src/state.rs")
        .replace("\r\n", "\n");
    let i = state_src
        .find("pub fn apply_status(")
        .expect("apply_status 必须在 state.rs");
    let j = state_src[i..]
        .find("\n    }\n")
        .expect("apply_status 函数体收尾");
    let apply = &state_src[i..i + j];
    assert!(
        !apply.contains("chain_note"),
        "链状态只是显示,不得参与在线/断供判据,实测 {}",
        apply.chars().take(120).collect::<String>()
    );

    let plugin_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/plugin.rs"))
        .expect("读 src/plugin.rs")
        .replace("\r\n", "\n");
    assert!(
        plugin_src.contains("strip_prefix(crate::state::CHAIN_STATE_PREFIX)"),
        "eLog 分支筛状态行必须走常量,不是散在分支里的字面量"
    );
    assert!(plugin_src.contains("note_chain(&d.gateway_name"));

    let render_src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/render.rs"))
        .expect("读 src/render.rs")
        .replace("\r\n", "\n");
    assert!(
        render_src.contains("const ACCOUNT_HEADERS: [&str; 8]"),
        "账户表 8 列(0.5.0 加「阶段」)"
    );
    assert!(
        render_src.contains(".num_columns(8)"),
        "account_grid 的列数必须跟着表头一起改(7 列表头 + 8 列 Grid 就是错位)"
    );
}

/// u5(0.5 M2):快照是账本的**缓存视图** —— 脏了才重建,一帧最多一次。
///
/// 回报潮里一百条事件如果每条都重建,入账就从 O(1) 变成 O(表长)。
#[test]
fn u5_trade_snap_is_rebuilt_only_when_the_books_are_dirty() {
    let s = UiShared::new();
    s.note_order(&order_row(
        "ctp",
        "A1",
        ek_data::constant::Status::SUBMITTING,
        1.0,
        1.0,
    ));
    let a = s.trade_snap();
    let b = s.trade_snap();
    assert!(
        Arc::ptr_eq(&a, &b),
        "没新事件就不重建(同一份 Arc,不是每次重新组帧)"
    );
    s.note_order(&order_row(
        "ctp",
        "A2",
        ek_data::constant::Status::SUBMITTING,
        2.0,
        2.0,
    ));
    s.note_order(&order_row(
        "ctp",
        "A3",
        ek_data::constant::Status::SUBMITTING,
        3.0,
        3.0,
    ));
    let c = s.trade_snap();
    assert!(!Arc::ptr_eq(&a, &c), "脏了必须重建");
    assert_eq!(c.orders.len(), 3);
    let d = s.trade_snap();
    assert!(Arc::ptr_eq(&c, &d), "三条事件只重建一次");
}

/// u6(0.7.0 背压刀改口径):水位与触顶的三段告警 —— 每级每插件一次,阈值吃行内软顶。
///
/// 0.5 M6 时代界面自带一张容量登记表(`mailbox_cap_of`);ek-station 0.4.0 把
/// `soft_cap` 随快照带来,登记表退役(u6c 钉住"界面不再自备容量"这条)。
/// 新语义链:无界默认不丢 → `spilled>0` = **已触顶**(红),`pending` 越阈 = 滞后显形。
#[test]
fn u6_water_and_soft_cap_alerts_dedupe_per_level() {
    let s = UiShared::new();
    // u6c:阈值口径 = 比例(1/16 黄、1/4 红)+ 旧定数下限;0 软顶不许炸
    let (w1, c1) = ek_ui::state::pending_thresholds(262_144);
    assert_eq!((w1, c1), (16_384, 65_536), "缺省软顶行的阈 = 1/16 与 1/4");
    let (w2, c2) = ek_ui::state::pending_thresholds(100_000);
    assert_eq!((w2, c2), (6_250, 32_768), "ui 行(10 万软顶)的阈各算各的");
    let (w3, c3) = ek_ui::state::pending_thresholds(1_000);
    assert!(
        w3 >= 1 && c3 >= w3 && c3 <= 1_000,
        "小软顶收口不炸:({w3},{c3})"
    );
    assert_eq!(
        ek_ui::state::pending_thresholds(0),
        (1, 1),
        "软顶 0 的异常快照:阈值收口到 1,不炸也不静默"
    );

    // 全 0 = 没事:不响
    s.note_plugin_spill(&[water_row("dm", 0, 0, 0, 262_144)]);
    assert_eq!(s.pending_alert_count(), 0, "零水位零丢弃不该响");

    // 黄阈(16384)→ 一条 Warning;持续积压不刷屏
    s.note_plugin_spill(&[water_row("dm", 0, 16_384, 16_384, 262_144)]);
    assert_eq!(s.pending_alert_count(), 1, "越黄阈弹一条");
    s.note_plugin_spill(&[water_row("dm", 0, 20_000, 20_000, 262_144)]);
    assert_eq!(s.pending_alert_count(), 1, "同一级不重复弹");

    // 红阈(65536)→ 再一条(「再涨就要丢新帧」是另一件事)
    s.note_plugin_spill(&[water_row("dm", 0, 65_536, 65_536, 262_144)]);
    assert_eq!(s.pending_alert_count(), 2, "越红阈是另一件事,必须再弹");

    // 触顶丢弃(spilled>0)→ 第三条(已丢 = 红色大事)
    s.note_plugin_spill(&[water_row("dm", 1, 65_536, 65_536, 262_144)]);
    assert_eq!(s.pending_alert_count(), 3, "已触顶丢新帧必须显形");
    s.note_plugin_spill(&[water_row("dm", 99, 65_536, 65_536, 262_144)]);
    assert_eq!(s.pending_alert_count(), 3, "同一级只弹一次(常显交给运维窗)");

    // 同一数字在不同软顶的行上不是同一件事(ui 行 10 万:4 万越红阈)
    let t = UiShared::new();
    t.note_plugin_spill(&[water_row("ui", 0, 40_000, 40_000, 100_000)]);
    assert_eq!(
        t.pending_alert_count(),
        2,
        "ui 行 4 万积压:黄阈(6250)与红阈(32768)都越过"
    );
}

/// u7(0.7.0 背压刀改口径):运维窗统计列的颜色判据(纯函数,不看像素)—— 主角是积压。
#[test]
fn u7_ops_window_stat_color_paints_the_worst_item() {
    // 0.8.0 双档:这一条钉的是水位阈值 → 颜色的决策口径,与档无关,按深档钉。
    const DARK: ek_ui::tokens::Theme = ek_ui::tokens::Theme::Dark;
    let def = ek_core::plugin::MAILBOX_SOFT_CAP; // 262144:黄 16384 / 红 65536
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 0, 0, def),
        ek_ui::tokens::semantic::MUTED,
        "0/0/0 的意思是「没事」,不是数据"
    );
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 0, 16_384, def),
        ek_ui::tokens::semantic::WARN,
        "积压越黄阈 → 黄"
    );
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 0, 65_536, def),
        ek_ui::tokens::semantic::ERROR,
        "积压越红阈(软顶 1/4)→ 红:再涨就丢新帧"
    );
    assert_eq!(
        ek_ui::render::stat_color(DARK, 1, 0, 0, def),
        ek_ui::tokens::semantic::ERROR,
        "新口径里 spilled>0 = 已触顶丢新帧 = 红色大事"
    );
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 1, 0, def),
        ek_ui::tokens::semantic::ERROR,
        "panic 照旧红"
    );
    // 同一数字在两行软顶各自判:4 万在缺顶行(红阈 65536)是黄,在 ui 行(红阈 32768)是红
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 0, 40_000, def),
        ek_ui::tokens::semantic::WARN,
        "缺省软顶行:4 万没到它的 1/4"
    );
    assert_eq!(
        ek_ui::render::stat_color(DARK, 0, 0, 40_000, ek_ui::plugin::UI_MAILBOX_SOFT_CAP),
        ek_ui::tokens::semantic::ERROR,
        "ui 行(10 万软顶):4 万越过它的 1/4"
    );
}

/// t4(0.5 M1/M2 真链):委托行是「推来的」而不是「问来的」。
///
/// 判据 = **不拍任何 eTimer**:直发一条 `eOrder.` 事件,行就该进账本、进渲染快照。
/// 后半段钉的是「作废归谁」:摘掉 oms 不动这些行(它们属于网关 gwA,不属于 oms);
/// `ePlugDown.<网关>` 才作废它们 —— 推送模型里没人再告诉界面「这一行已经不存在了」,
/// 不清就是替一个已经不存在的会话说话。
#[test]
fn t4_order_rows_arrive_by_push_not_by_polling() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    boot(
        &engine,
        &ek_ui::ui_kinds(&shared, true),
        &StationConfig {
            plugins: vec!["dm".to_string(), "oms".to_string(), "ui".to_string()],
        },
    )
    .unwrap();
    engine.start().unwrap();
    wait_until("rows fill", 4, || shared.snapshot().rows.len() >= 4);

    engine
        .event_engine()
        .put(Event::with_data(
            "eOrder.",
            Arc::new(order_row(
                "gwA",
                "A1",
                ek_data::constant::Status::SUBMITTING,
                3.0,
                7.0,
            )),
        ))
        .unwrap();
    wait_until("pushed order lands", 4, || {
        shared.trade_snap().orders.iter().any(|o| o.orderid == "A1")
    });

    shared.push_cmd(UiCmd::Remove {
        name: "oms".to_string(),
    });
    wait_until("oms removed", 4, || {
        shared.snapshot().rows.iter().all(|r| r.name != "oms")
    });
    assert!(
        shared.trade_snap().orders.iter().any(|o| o.orderid == "A1"),
        "摘掉 oms 不清交易表:那些行属于网关 gwA,不属于 oms(0.4 的「oms 不在场就清空」随四问一起退役)"
    );

    engine
        .event_engine()
        .put(Event::with_data(
            ek_core::EVENT_PLUG_DOWN,
            Arc::new(PlugNotice {
                name: "gwA".to_string(),
                version: "0".to_string(),
            }),
        ))
        .unwrap();
    wait_until("gateway rows voided", 4, || {
        shared.trade_snap().orders.is_empty()
    });

    engine.close_all(Duration::from_secs(2));
}

/// u8(0.5 M3/M5/M6 形状钉):`refresh` 那一拍里不许再出现「取数类 ask」。
///
/// 为什么是源码扫描而不是行为测试:轮询与推送在测试里都能「过」——真链上四口都是
/// 毫秒级,问与不问看不出差别。所以钉形状:refresh 只许调三个 `poll_*`;四口只许出现在
/// `reconcile_trade`(启动一次)里;邮箱容量必须是命名常量;四条推送线必须在 routes 里。
#[test]
fn u8_refresh_asks_no_trading_data_and_the_push_lines_are_routed() {
    let src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/plugin.rs"))
        .expect("读 src/plugin.rs")
        .replace("\r\n", "\n");
    let body = |needle: &str, stop: &str| -> String {
        let i = src
            .find(needle)
            .unwrap_or_else(|| panic!("plugin.rs 里找不到 {needle:?}"));
        let j = src[i..]
            .find(stop)
            .unwrap_or_else(|| panic!("{needle:?} 之后找不到 {stop:?}"));
        src[i..i + j + stop.len()].to_string()
    };

    let refresh = body("    fn refresh(&self) {", "\n    }\n");
    assert!(
        !refresh.contains("ask_service"),
        "refresh 那一拍不许直接 ask(0.5 M3),实测 {refresh:?}"
    );
    assert!(
        !refresh.contains("run_cmds"),
        "指令辞行不在 eTimer 拍上(0.5 M5),实测 {refresh:?}"
    );

    let rec = body(
        "    fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {",
        "\n    }\n",
    );
    for svc in [
        "\"oms.get_all_orders\"",
        "\"oms.get_all_trades\"",
        "\"oms.get_all_positions\"",
        "\"oms.get_all_accounts\"",
    ] {
        assert_eq!(
            src.matches(svc).count(),
            1,
            "{svc} 只许出现一次(启动一次四问),不许回到周期轮询"
        );
        assert!(rec.contains(svc), "{svc} 必须在 reconcile_trade 里");
    }

    let start = body("    fn on_start(&self) {", "\n    }\n");
    assert!(
        start.contains("spawn_cmd_thread"),
        "启动 = 起命令线程(会等的东西都归它),实测 {start:?}"
    );
    let spawn = body("    fn spawn_cmd_thread(&self) {", "\n    }\n");
    assert!(
        spawn.contains("reconcile_trade"),
        "启动一次四问挂在命令线程里:`on_start` 是宿主启动链的一环,四口最坏 8 秒不许把后面插件的启动一起拖住"
    );
    assert!(
        start.contains("request_trade_reseed"),
        "启动播种走标记(消费方 = 命令线程),不在 on_start 里直接问,实测 {start:?}"
    );
    let mail = body(
        "    fn on_mail(&self, event: &ek_event::Event) {",
        "\n    }\n",
    );
    assert!(
        mail.contains("request_trade_reseed") && mail.contains("request_pipeline_refresh"),
        "网关 / oms 上线必须重播种一次并重问装配:摘除时作废的行要能回来,实测 {:?}",
        mail.chars().take(160).collect::<String>()
    );

    assert!(
        src.contains("pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000"),
        "邮箱软顶必须是命名常量(带理由),不是散在 mailbox_soft_cap 里的字面量(0.7.0 口径)"
    );
    let mb = body("    fn mailbox_soft_cap(&self) -> u64 {", "\n    }\n");
    assert!(
        mb.contains("UI_MAILBOX_SOFT_CAP") && !mb.contains("1024"),
        "mailbox_soft_cap 吃常量,实测 {mb:?}"
    );
    assert!(
        !src.contains("fn mailbox_cap("),
        "0.4 的容量覆写 `mailbox_cap` 必须从链上消失(邮箱已无界,它不再是容量)"
    );

    let routes = body("    fn routes(&self)", "\n    }\n");
    for line in ["W_ORDER", "W_TRADE", "W_POSITION", "W_ACCOUNT"] {
        assert!(routes.contains(line), "routes 里缺 {line},实测 {routes:?}");
    }
    assert!(
        !routes.contains("Cow::Borrowed(\"eTradeSync\")") && !routes.contains("EVENT_TRADE_SYNC"),
        "eTradeSync. 是 oms 处理完成交后的重推,订两条就是同一笔成交入账两次(注释里提到它可以,订阅不行)"
    );
    assert!(
        !src.contains("OMS_ASK_TIMEOUT") && !src.contains("merge_frame"),
        "0.4.5 的超时分级与读数裁决随四问一起退役,不许留成死代码"
    );
}