ek-ui 0.4.5

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

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

use ek_core::main_engine::MainEngine;
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.4.4,P5-a):每 `PIPELINE_POLL_TICKS` = 4 拍回一次 `true`。
///
/// 为什么钉在 `UiShared` 上而不是真链里:`engine.start()` 会同时起 ek-event 那条
/// 1 秒定时线程,真拍与手动 `put(eTimer)` 交错 —— 「第几拍」在集成套里判不稳
/// (第一版 t3 就是被这条绊倒的:手动三拍之后读数已经非空)。节拍是纯状态判定,
/// 就该在纯状态上判。另一条:它与 `status_poll_due` 是**两个计数器**,共用会被
/// 相位带着走,这条钉也判不出来。
#[test]
fn u3_pipeline_poll_cadence_is_four_ticks() {
    let s = UiShared::new();
    let due: Vec<bool> = (0..9).map(|_| s.pipeline_poll_due()).collect();
    assert_eq!(
        due,
        vec![false, false, false, true, false, false, false, true, false],
        "节拍 = 每 4 拍一次(第 4、第 8 拍)"
    );
    // 与状态轮询互不干扰:status 是 2 拍一档,共用计数器就会在这里露馅。
    let status: Vec<bool> = (0..9).map(|_| s.status_poll_due()).collect();
    assert_eq!(
        status.iter().filter(|b| **b).count(),
        4,
        "状态轮询仍是每 2 拍一次(第 2、4、6、8 拍)"
    );
}

// ---- 集成: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 = 自己等自己:这条指令
/// 正是 ui 自己的 worker 在 eTimer 拍上执行,只能走到 ek-core `close_one` 的 2 秒
/// 超时 detach 分支,窗口还在但快照/日志/指令队列全停)。
/// `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:?}"
        );
    }

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

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

fn spill_row(name: &str, spilled: 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,
    }
}

/// u4(P7a):一张表的读数必须分三档,而且三档各自判得出来。
///
/// 这一钉是「空帧比旧帧坏」那条判断的地面:`Timeout` 保旧(问不到 ≠ 没有),
/// `TargetGone` / `UnknownService` / `InvalidPayload` / `BusStopped` 清空(那是事实)。
/// 把四档宿主错合并进「保旧」那一档,就会让一台已经摘掉的 oms 永远留着一帧委托。
#[test]
fn u4_merge_frame_splits_timeout_from_target_gone() {
    let prev = vec![1u32, 2];
    assert_eq!(
        ek_ui::plugin::merge_frame(Ok(vec![9u32]), prev.clone()),
        vec![9]
    );
    assert_eq!(
        ek_ui::plugin::merge_frame(
            Err(ek_core::error::AskError::Timeout {
                target: "oms".to_string(),
            }),
            prev.clone(),
        ),
        prev,
        "超时保旧帧:原先 unwrap_or_default 把四张表一起清空"
    );
    for gone in [
        ek_core::error::AskError::TargetGone {
            target: "oms".to_string(),
        },
        ek_core::error::AskError::UnknownService {
            service: "oms.get_all_orders".to_string(),
        },
        ek_core::error::AskError::InvalidPayload {
            target: "oms".to_string(),
        },
        ek_core::error::AskError::BusStopped,
    ] {
        assert_eq!(
            ek_ui::plugin::merge_frame::<u32>(Err(gone.clone()), prev.clone()),
            Vec::<u32>::new(),
            "目标不在场类错必须清空(留旧帧才是骗人): {gone}"
        );
    }
}

/// u5(P7a):stale 是**迁移**而不是状态,而且进入那一次必须灌一条告警。
#[test]
fn u5_trade_frame_stale_is_a_transition_and_alerts_on_entry() {
    let s = UiShared::new();
    assert!(!s.trade_frame_stale());
    assert!(s.note_trade_frame_stale(true), "进入 stale 是一次迁移");
    assert_eq!(
        s.pending_alert_count(),
        1,
        "旧帧必须被标明:进入 stale 灌一条告警"
    );
    assert!(
        !s.note_trade_frame_stale(true),
        "持续 stale 不是迁移(拥堵期不每拍刷)"
    );
    assert_eq!(s.pending_alert_count(), 1, "同一状态不重复灌");
    assert!(s.note_trade_frame_stale(false), "回到当前帧也是一次迁移");
    assert_eq!(s.pending_alert_count(), 1, "恢复不灌 toast(日志给)");
}

/// u6(P7d):溢出告警的去重口径 —— 每插件一次,跨过邮箱容量再一次。
#[test]
fn u6_spill_alerts_once_per_plugin_then_again_at_mailbox_cap() {
    let s = UiShared::new();
    s.note_plugin_spill(&[spill_row("ui", 3), spill_row("dm", 0)]);
    assert_eq!(s.pending_alert_count(), 1, "只有真溢出的那一行该响");
    s.note_plugin_spill(&[spill_row("ui", 77)]);
    assert_eq!(
        s.pending_alert_count(),
        1,
        "持续溢出不刷屏(ek-core 侧每 128 条才打一条 stderr)"
    );
    s.note_plugin_spill(&[spill_row("ui", ek_core::plugin::MAILBOX_CAP as u64)]);
    assert_eq!(
        s.pending_alert_count(),
        2,
        "溢满一整轮邮箱是另一件事,必须再弹一次"
    );
    s.note_plugin_spill(&[spill_row("ui", ek_core::plugin::MAILBOX_CAP as u64 + 9)]);
    assert_eq!(s.pending_alert_count(), 2, "容量档也只弹一次");
}

/// u7(P7d):运维窗统计列的颜色判据(纯函数,不看像素)。
#[test]
fn u7_ops_window_stat_color_paints_the_worst_item() {
    assert_eq!(
        ek_ui::render::stat_color(0, 0),
        ek_ui::tokens::semantic::MUTED,
        "0/0 的意思是「没事」,不是数据"
    );
    assert_eq!(
        ek_ui::render::stat_color(1, 0),
        ek_ui::tokens::semantic::WARN
    );
    assert_eq!(
        ek_ui::render::stat_color(ek_core::plugin::MAILBOX_CAP as u64, 0),
        ek_ui::tokens::semantic::ERROR,
        "溢满一整轮 = 丢的是一段时期的事件"
    );
    assert_eq!(
        ek_ui::render::stat_color(0, 1),
        ek_ui::tokens::semantic::ERROR
    );
}

/// t4(P7a 真链):oms 被摘掉之后必须**清空**而不是标成 stale。
///
/// 这一钉钉的是两档宿主错在真链上的分界:摘掉 oms 之后四口走 `UnknownService` /
/// `TargetGone` 快错(不是超时)→ 清空 + 不标 stale。把这一档错并进「保旧」那一档,
/// 界面上就会一直显示一台已经不存在的 oms 的委托。
#[test]
fn t4_removed_oms_clears_the_frame_instead_of_marking_it_stale() {
    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);

    put_timer(&engine);
    wait_until("frame asked", 4, || !shared.pipeline_lines().is_empty());
    assert!(
        !shared.trade_frame_stale(),
        "oms 在场且空闲:四问不该进 stale"
    );

    shared.push_cmd(UiCmd::Remove {
        name: "oms".to_string(),
    });
    put_timer(&engine);
    wait_until("oms removed", 4, || {
        shared.snapshot().rows.iter().all(|r| r.name != "oms")
    });

    put_timer(&engine);
    wait_until("frame cleared", 4, || shared.pipeline_lines().is_empty());
    let ts = shared.trade_snap();
    assert!(
        ts.orders.is_empty()
            && ts.trades.is_empty()
            && ts.positions.is_empty()
            && ts.accounts.is_empty(),
        "oms 不在场 = 事实,四张表必须清空"
    );
    assert!(
        !shared.trade_frame_stale(),
        "目标不在场不是「问不到」,不许标 stale"
    );

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

/// u8(P7b 钉):超时分级必须是**命名常量 + 四口都吃它**,而且另两口留 2 秒不许跟着降。
///
/// 为什么是源码扫描而不是行为测试:1 秒与 2 秒在测试里没法可靠区分 —— 真链上四口
/// 都是毫秒级,两种时限都「过」。所以钉的是形状:分级不许散回字面量,也不许顺手把
/// `svctl.list`(可能正撞 `close_one` 的 2 秒摘除窗)与 `{gw}.status`(可能正撞网关
/// 查询协调点的 900ms 墙钟护栏)一起降掉 —— 那两口是真会慢的。
#[test]
fn u8_oms_ask_tier_is_a_named_constant_and_the_slow_ports_stay_at_two_seconds() {
    let src = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/plugin.rs"))
        .expect("读 src/plugin.rs");
    let window_after = |needle: &str, len: usize| -> String {
        let idx = src
            .find(needle)
            .unwrap_or_else(|| panic!("plugin.rs 里找不到 {needle:?}"));
        src[idx..(idx + len).min(src.len())].to_string()
    };
    assert!(
        src.contains("pub const OMS_ASK_TIMEOUT: Duration = Duration::from_secs(1)"),
        "P7b 的 1 秒必须是命名常量(带理由),不是散在四口里的字面量"
    );
    for svc in [
        "\"oms.get_all_orders\"",
        "\"oms.get_all_trades\"",
        "\"oms.get_all_positions\"",
        "\"oms.get_all_accounts\"",
    ] {
        let w = window_after(svc, 160);
        assert!(
            w.contains("OMS_ASK_TIMEOUT"),
            "{svc} 必须吃分级常量,实测窗口 {w:?}"
        );
        assert!(
            !w.contains("Duration::from_secs(2)"),
            "{svc} 不许留回 2 秒,实测窗口 {w:?}"
        );
    }
    // 慢的那两口留 2 秒。
    let w_list = window_after("ask_service::<ListReq, Vec<PlugRow>>", 120);
    assert!(
        w_list.contains("Duration::from_secs(2)"),
        "svctl.list 留 2 秒(可能正撞 close_one 的摘除窗),实测窗口 {w_list:?}"
    );
    let w_status = window_after("ask_service::<StatusReq, GatewayStatus>", 200);
    assert!(
        w_status.contains("Duration::from_secs(2)"),
        "{{gw}}.status 留 2 秒(可能正撞网关查询协调点的墙钟护栏),实测窗口 {w_status:?}"
    );
}