ek-station 0.4.13

//! 双插件装配与互动(跨 crate 集成套;0.1.5 波次从 ek-converter 迁来本 crate)。
//!
//! 宿主 = ek-event 总线之上的 ek-core MainEngine;`oms` 与夹具(注册名
//! `offset`)在同一宿主内,依赖由 `Plugin::deps()` 声明(`offset → ["oms"]`)。
//! 经总线推送四类数据行(eContract./ePosition./eOrder./eTrade.,不推 eTick.),
//! 验证:
//!
//! - **A 装配**:a1 拓扑序(ePlugUp 到达序 = deps 拓扑序);a2 缺依赖
//!   (`UnknownDep`);a3 重名(`Duplicate`)。
//! - **B 缓存经总线可查**:b1 合约命中/miss;b2 持仓与活跃单;b4 三计量
//!   (trade_info_count / cancel_count / order_cancel_count)。
//! - **C 互动(核心)**:夹具点对点 ask oms 拿合约 —— c1 动态注入后首见回查
//!   (ask 计数 = 1,顺带证明动态 add 上线);c2 已收 eContract. 的本地缓存
//!   直用(ask 计数 = 0);c3 双方皆无该合约 → 库层闸门,不建账本不 panic;
//!   c4 账本服务快照(6 持仓量 + 6 冻结量)。
//! - **D 拆单(只走服务路径)**:d1 SHFE 今昨拆(今先);d2 可平不足 →
//!   空 Vec = 拒单;不重复本 crate 的库测(golden/matrix/semantics 已覆盖)。
//! - **E 契约面**:e1 错服务名 → `UnknownService`;e2 close_all 终态
//!   (close 不广播逐插件 ePlugDown,断言只看 plugin_names/get_state)。
//! - **F 相位门**:f1 `ePositionDone.` 后夹具不再把持仓摆入账本(镜像原稿
//!   持仓两相语义);oms 自身的三秒稳定门经总线不可快测(时钟不外流),
//!   oms 侧保持其既有黄金/语义覆盖,此处不断言其门。
//!
//! 结构保证(跨 crate,且比原来更强):夹具与 ek-oms、ek-converter **都不同 crate**——
//! 不是"纪律上不读",而是【结构上拿不到】:宿主内跑着的 Oms 状态是 `OmsPlugin` 的私有
//! 字段,跨 crate 无路可及;夹具只见其公开服务契约(请求类型 + 总线)。
//!
//! 为什么搬(用户裁定 2026-10-23,拆环):原来这套住在 ek-converter,靠三个 dev-dep
//! (`ek-oms` / `ek-core` / `ek-event`)跑,于是注册表里出现 `converter → oms → converter` 的环
//! (`ek-converter 0.1.4` 记着 `ek-oms ^0.3.13 [dev]`,`ek-oms 0.3.13` 记着 `ek-converter ^0.1.1
//! [normal]`),ek-oms 只能把自己的 pin 冻在 "0.1.1" 绕开发布死锁。搬到 ek-station 后:
//! ek-converter 变纯叶子(只依赖 ek-data),`ek-oms → ek-converter` 恢复成单向边,冻结 pin 解除。
//! 搬到这里而不是 ek-oms,正是为了保住上面那条「夹具与 oms 不同 crate」的结构保证。
//!
//! 与同 crate 既有套的分工(不重复):`station.rs` 测的是 **svctl 监管面**(未知 kind /
//! 重名 kind 被监管挡下、start→remove 业务流、close 反拓扑序);本文件测的是 **宿主 add 层**
//! 的 deps 拓扑与 `UnknownDep` / `Duplicate`,以及 oms 与拆单夹具隔着总线的服务契约。
//! `multi_account.rs` 测多网关实例;本文件只有一台 oms + 一个拆单夹具。

use ek_converter::{ConvertMode, OffsetConverter, PositionHolding};
use ek_core::lifecycle::EVENT_PLUG_UP;
use ek_core::service::PlugNotice;
use ek_core::{
    AskError, Envelope, MainEngine, Plugin, PluginContext, PluginError, PluginState, ReplyErr,
    ServiceDesc, e_req_type,
};
use ek_data::OrderData;
use ek_data::OrderRequest;
use ek_data::constant::{Asset, Direction, Exchange, Offset, OrderType, Status};
use ek_data::contract::ContractData;
use ek_data::position::PositionData;
use ek_data::trade::TradeData;
use ek_event::{Event, EventEngine};
use ek_oms::plugin::{
    CancelCountReq, GetAllActiveOrdersReq, GetAllOrdersReq, GetContractReq, GetPositionReq,
    OmsPlugin, OrderCancelCountReq, TradeInfoCountReq,
};
use std::any::Any;
use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

// ==== 夹具(线值常量:复制同源,不持 oms 的 wire 模块;测试内私有) ====

const W_ORDER: &str = "eOrder.";
const W_TRADE: &str = "eTrade.";
const W_POSITION: &str = "ePosition.";
const W_CONTRACT: &str = "eContract.";
const W_POS_DONE: &str = "ePositionDone.";

/// 夹具注册名(测试内私有,不发布;生态服务命名照 `<名>.<动作>` 惯例)。
const FIXTURE_NAME: &str = "offset";

/// 夹具服务契约:取账本快照(整只 [`PositionHolding`],含冻结量)。
#[derive(Debug, Clone)]
struct GetHoldingReq {
    gateway_name: String,
    en_symbol: String,
}

/// 夹具服务契约:下单请求转换(镜像原稿 `lock_mode` 四态交由全权)。
#[derive(Debug, Clone)]
struct ConvertReq {
    req: OrderRequest,
    gateway_name: String,
    lock_mode: LockMode,
}

/// 原稿 `lock_mode` 四态(engine.py:1798):auto/lock/net/none。
///
/// 测试只触达 Auto/None 两态(Lock/Net 的判定语义由 converter 库测覆盖,R2);
/// 变体保留以镜像原稿四态面。
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LockMode {
    Auto,
    Lock,
    Net,
    None,
}

/// 平仓拆单夹具:包住 [`OffsetConverter`],镜像原稿 OmsEngine.process_* 中
/// converter 的接入位(合约懒建 converter;订单/成交常入;持仓只入初始化相)。
/// 洞察仅展活动现场(应答时算,不留壳内账)。
struct OffsetFixture {
    /// gateway → converter(原稿 `offset_converters`;首张合约/首笔回查命中时懒建)。
    conv: Mutex<HashMap<String, OffsetConverter>>,
    /// 两级合约缓存:总线已收的直用(原稿本地缓存面)。
    contracts: Mutex<HashMap<String, ContractData>>,
    /// 相位标志:收到 `ePositionDone.` 的 gateway 视为初始化完成(原稿
    /// `_position_initialized` 由 ePositionDone. 触发的镜像;无钟依赖)。
    started: Mutex<HashSet<String>>,
    /// 向 oms 发起的 ask 次数(互动实证计数:c1 应为 1,c2 应为 0)。
    asks: Mutex<u64>,
    ctx: OnceLock<PluginContext>,
}

impl OffsetFixture {
    fn new() -> Self {
        Self {
            conv: Mutex::new(HashMap::new()),
            contracts: Mutex::new(HashMap::new()),
            started: Mutex::new(HashSet::new()),
            asks: Mutex::new(0),
            ctx: OnceLock::new(),
        }
    }

    fn asks_made(&self) -> u64 {
        *self.asks.lock().unwrap()
    }

    /// 合约解析(两级):总线已收的用本地;miss → 点对点 ask oms
    /// (500ms,超时重试一次)。命中并记下(两级缓存自填,原稿本地缓存的
    /// 镜像),同步懒建该 gateway 的 converter。
    fn resolve(&self, en_symbol: &str) -> Option<ContractData> {
        if let Some(c) = self.contracts.lock().unwrap().get(en_symbol) {
            return Some(c.clone());
        }
        let ctx = self.ctx.get()?;
        *self.asks.lock().unwrap() += 1;
        for _ in 0..2 {
            match ctx.ask::<GetContractReq, Option<ContractData>>(
                "oms",
                GetContractReq {
                    en_symbol: en_symbol.to_string(),
                },
                Duration::from_millis(500),
            ) {
                Ok(hit) => {
                    if let Some(c) = &hit {
                        self.contracts
                            .lock()
                            .unwrap()
                            .insert(en_symbol.to_string(), c.clone());
                        self.ensure_converter(c);
                    }
                    return hit;
                }
                Err(AskError::Timeout { .. }) => continue,
                Err(_) => return None,
            }
        }
        None
    }

    /// 该 gateway 首张合约 → 懒建 converter(原稿 1484-1486)。
    fn ensure_converter(&self, c: &ContractData) {
        self.conv
            .lock()
            .unwrap()
            .entry(c.gateway_name.clone())
            .or_default();
    }

    fn on_contract(&self, c: &ContractData) {
        self.contracts
            .lock()
            .unwrap()
            .insert(c.en_symbol(), c.clone());
        self.ensure_converter(c);
    }

    fn on_order(&self, o: &OrderData) {
        let contract = self.resolve(&o.en_symbol());
        if let Some(conv) = self.conv.lock().unwrap().get_mut(o.gateway_name.as_str()) {
            conv.update_order(o, contract.as_ref());
        }
    }

    fn on_trade(&self, t: &TradeData) {
        let contract = self.resolve(&t.en_symbol());
        if let Some(conv) = self.conv.lock().unwrap().get_mut(t.gateway_name.as_str()) {
            conv.update_trade(t, contract.as_ref());
        }
    }

    fn on_position(&self, p: &PositionData) {
        // 相位门(镜像原稿初始化相才入 converter;运行相持仓不覆盖成交账本)。
        if self
            .started
            .lock()
            .unwrap()
            .contains(p.gateway_name.as_str())
        {
            return;
        }
        let contract = self.resolve(&p.en_symbol());
        if let Some(conv) = self.conv.lock().unwrap().get_mut(p.gateway_name.as_str()) {
            conv.update_position(p, contract.as_ref());
        }
    }

    fn on_pos_done(&self, gateway: &str) {
        self.started.lock().unwrap().insert(gateway.to_string());
    }

    /// 账本快照(服务面; Convert 快照同形状原则整库对象直出)。
    fn holding(&self, gateway: &str, en_symbol: &str) -> Option<PositionHolding> {
        self.conv
            .lock()
            .unwrap()
            .get(gateway)
            .and_then(|c| c.holding(en_symbol).cloned())
    }

    /// 下单转换(镜像原稿 convert_order_request 的四态分派与锁仓判定)。
    fn convert(&self, req: OrderRequest, gateway: &str, m: LockMode) -> Vec<OrderRequest> {
        if m == LockMode::None {
            return vec![req]; // 原稿"none"短路(engine.py:1802)。
        }
        let contract = self.resolve(&req.en_symbol());
        let mut convs = self.conv.lock().unwrap();
        let Some(conv) = convs.get_mut(gateway) else {
            return vec![req]; // converter 未主(该 gateway 无合约史)→ 原样。
        };
        match m {
            LockMode::None => unreachable!("short-circuited above"),
            LockMode::Lock => {
                conv.convert_order_request(&req, ConvertMode::Lock, contract.as_ref())
            }
            LockMode::Net => conv.convert_order_request(&req, ConvertMode::Net, contract.as_ref()),
            LockMode::Auto => {
                if req.offset != Offset::CLOSE {
                    return conv.convert_order_request(&req, ConvertMode::Auto, contract.as_ref());
                }
                // 原稿 1810-1856:CLOSE + auto → 锁仓品种判定(SETTINGS
                // lock_setting.lock_products + _check_auto_lock)。夹具不配
                // 锁仓品种表(未配置的品种恒走非锁仓路径),因此恒 Auto:
                // SHFE/INE 今昨拆,它家原样。真锁仓品种表随将来配置注入,
                // 判定语义已在 converter 的 lock/net 库测里覆盖(R2:不重复)。
                let Some(c) = contract else { return vec![req] };
                conv.convert_order_request(&req, ConvertMode::Auto, Some(&c))
            }
        }
    }

    /// e1 用:错服务名探测(ask_service 的零成本显错)。
    fn unknown_service_probe(&self) -> Result<Vec<OrderData>, AskError> {
        self.ctx
            .get()
            .unwrap()
            .ask_service::<GetAllOrdersReq, Vec<OrderData>>(
                "offset.nothing",
                GetAllOrdersReq,
                Duration::from_millis(200),
            )
    }
}

impl Plugin for OffsetFixture {
    fn name(&self) -> &str {
        FIXTURE_NAME
    }

    fn deps(&self) -> &[&str] {
        &["oms"]
    }

    fn routes(&self) -> Vec<Cow<'static, str>> {
        vec![
            Cow::Borrowed(W_ORDER),
            Cow::Borrowed(W_TRADE),
            Cow::Borrowed(W_POSITION),
            Cow::Borrowed(W_CONTRACT),
            Cow::Borrowed(W_POS_DONE),
            Cow::Owned(e_req_type(self.name())),
        ]
    }

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

    fn services(&self) -> Vec<ServiceDesc> {
        vec![
            ServiceDesc::describe::<GetHoldingReq, Option<PositionHolding>>(
                "offset.get_holding",
                "",
            ),
            ServiceDesc::describe::<ConvertReq, Vec<OrderRequest>>("offset.convert", ""),
        ]
    }

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

    fn on_mail(&self, event: &Event) {
        match event.event_type.as_str() {
            W_ORDER => {
                if let Some(o) = event.data.downcast_ref::<OrderData>() {
                    self.on_order(o);
                }
            }
            W_TRADE => {
                if let Some(t) = event.data.downcast_ref::<TradeData>() {
                    self.on_trade(t);
                }
            }
            W_POSITION => {
                if let Some(p) = event.data.downcast_ref::<PositionData>() {
                    self.on_position(p);
                }
            }
            W_CONTRACT => {
                if let Some(c) = event.data.downcast_ref::<ContractData>() {
                    self.on_contract(c);
                }
            }
            W_POS_DONE => {
                if let Some(g) = event.data.downcast_ref::<String>() {
                    self.on_pos_done(g);
                }
            }
            t if t == e_req_type(self.name()) => self.on_request(event),
            _ => {}
        }
    }
}

impl OffsetFixture {
    fn on_request(&self, event: &Event) {
        let Some(env) = event.data.downcast_ref::<Envelope>() else {
            return;
        };
        macro_rules! arm {
            ($ty:ty, $res:ty, $body:expr) => {
                if let Some(req) = env.payload.downcast_ref::<$ty>() {
                    let res: $res = $body(req);
                    let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
                    return;
                }
            };
        }
        arm!(
            GetHoldingReq,
            Option<PositionHolding>,
            |r: &GetHoldingReq| self.holding(r.gateway_name.as_str(), r.en_symbol.as_str())

        );
        arm!(ConvertReq, Vec<OrderRequest>, |r: &ConvertReq| self

            .convert(r.req.clone(), r.gateway_name.as_str(), r.lock_mode));
        let _ = env
            .reply
            .send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
    }
}

// ==== 数据构造与总线辅助 ====

/// 浮点等值(审计纪律:容差 1e-9)。
fn feq(a: f64, b: f64) -> bool {
    (a - b).abs() < 1e-9
}

fn contract(gateway: &str, symbol: &str, ex: Exchange, product: &str) -> ContractData {
    ContractData {
        gateway_name: gateway.to_string(),
        symbol: symbol.to_string(),
        exchange: ex,
        name: symbol.to_string(),
        asset: Asset::FUTURES,
        product_id: Some(product.to_string()),
        ..ContractData::default()
    }
}

#[allow(clippy::too_many_arguments)]
fn order(
    oid: &str,
    status: Status,
    gateway: &str,
    symbol: &str,
    ex: Exchange,
    dir: Direction,
    offset: Offset,
    vol: f64,
    traded: f64,
) -> OrderData {
    OrderData {
        gateway_name: gateway.to_string(),
        symbol: symbol.to_string(),
        exchange: ex,
        orderid: oid.to_string(),
        order_type: OrderType::LIMIT,
        direction: Some(dir),
        offset,
        price: 3500.0,
        volume: vol,
        traded,
        status,
        ..OrderData::default()
    }
}

fn trade(
    gateway: &str,
    symbol: &str,
    ex: Exchange,
    dir: Direction,
    offset: Offset,
    vol: f64,
) -> TradeData {
    TradeData {
        gateway_name: gateway.to_string(),
        symbol: symbol.to_string(),
        exchange: ex,
        orderid: "1".to_string(),
        tradeid: "t1".to_string(),
        direction: Some(dir),
        offset,
        price: 3500.0,
        volume: vol,
        ..TradeData::default()
    }
}

fn position(
    gateway: &str,
    symbol: &str,
    ex: Exchange,
    dir: Direction,
    vol: f64,
    yd: f64,
) -> PositionData {
    PositionData {
        gateway_name: gateway.to_string(),
        symbol: symbol.to_string(),
        exchange: ex,
        direction: dir,
        volume: vol,
        yd_volume: yd,
        ..PositionData::default()
    }
}

fn req(symbol: &str, ex: Exchange, dir: Direction, offset: Offset, vol: f64) -> OrderRequest {
    OrderRequest {
        symbol: symbol.to_string(),
        exchange: ex,
        direction: dir,
        order_type: OrderType::LIMIT,
        volume: vol,
        price: 3500.0,
        offset,
        reference: String::new(),
        old_order_id: String::new(),
    }
}

/// 总线级 ask(真实投递 eReq.<target> + oneshot 收包;2s 自动超时的心思同宿主 deliver)。
fn bus_ask<T: Any + Send + Sync, R: Any + Send + Sync>(
    ev: &EventEngine,
    target: &str,
    req: T,
) -> Result<R, ReplyErr> {
    let (env, rx) = Envelope::new(Arc::new(req));
    ev.put(Event::with_data(e_req_type(target), Arc::new(env)))
        .expect("bus alive");
    let boxed = rx
        .recv_timeout(Duration::from_secs(2))
        .expect("reply within 2s (serial mailbox makes put→ask deterministic)");
    match boxed.downcast::<Result<R, ReplyErr>>() {
        Ok(inner) => *inner,
        Err(_) => panic!("reply contract mismatch"),
    }
}

fn put<T: Any + Send + Sync>(ev: &EventEngine, ty: &str, payload: T) {
    ev.put(Event::with_data(ty.to_string(), Arc::new(payload)))
        .expect("bus alive");
}

/// 轮询至条件成立或超时(唯一需要的异步断言:ePlugUp 巡检在总线线程侧)。
fn wait_for(ms: u64, mut f: impl FnMut() -> bool) -> bool {
    let t0 = Instant::now();
    loop {
        if f() {
            return true;
        }
        if t0.elapsed() >= Duration::from_millis(ms) {
            return f();
        }
        std::thread::sleep(Duration::from_millis(2));
    }
}

struct Rig {
    host: MainEngine,
}

/// 装配一台宿主(oms 总在,夹具遵循相同依赖面;start 由调用方发起)。
fn rig() -> (Rig, Arc<EventEngine>, Arc<OmsPlugin>, Arc<OffsetFixture>) {
    let ev = Arc::new(EventEngine::new());
    let host = MainEngine::new(Some(ev.clone()));
    // 步骤 2 口径:OmsPlugin deps=["cfg"] = 宿主先装配 cfg 收件壳(显式 temp 根)。
    host.add_plugin(Arc::new(ek_cfg::CfgPlugin::new(
        ek_cfg::Store::with_root(std::path::PathBuf::from("target/asm-cfg-rig")),
        false,
    )))
    .expect("cfg add");
    let oms = Arc::new(OmsPlugin::new());
    host.add_plugin(oms.clone()).expect("oms add");
    let fx = Arc::new(OffsetFixture::new());
    (Rig { host }, ev, oms, fx)
}

fn close_all(host: &MainEngine) {
    host.close_all(Duration::from_secs(2));
}

// ==== A 组:装配 ====

// —— a1:拓扑序装配(ePlugUp 到达序 = deps 拓扑序:cfg 先、oms 次、offset 后;状态 Running) ——
#[test]
fn a1_topo_assembly() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    // 订阅取证:固定键 ePlugUp(载荷 PlugNotice)按到达序收集。
    let seen: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let cap = seen.clone();
    ev.register_general(Arc::new(move |e: &Event| {
        if e.event_type == EVENT_PLUG_UP {
            if let Some(n) = e.data.downcast_ref::<PlugNotice>() {
                cap.lock().unwrap().push(n.name.clone());
            }
        }
    }));
    rig.host.start().expect("start");
    assert!(
        wait_for(2000, || *seen.lock().unwrap()

            == vec![
                "cfg".to_string(),
                "oms".to_string(),
                FIXTURE_NAME.to_string()
            ]),
        "ePlugUp order must be cfg → oms → offset (deps topo,步骤 2 口径)"
    );
    assert_eq!(rig.host.get_state("cfg"), Some(PluginState::Running));
    assert_eq!(rig.host.get_state("oms"), Some(PluginState::Running));
    assert_eq!(rig.host.get_state(FIXTURE_NAME), Some(PluginState::Running));
    assert_eq!(
        rig.host.plugin_names(),
        vec!["cfg".to_string(), "oms".to_string(), "offset".to_string()]
    );
    close_all(&rig.host);
}

fn host_add(rig: &Rig, fx: &Arc<OffsetFixture>) {
    rig.host.add_plugin(fx.clone()).expect("fixture add");
}

// —— a2:缺依赖:start 拓扑校验 → UnknownDep(oms 不在册) ——
#[test]
fn a2_missing_dep() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let ev = Arc::new(EventEngine::new());
    let host = MainEngine::new(Some(ev));
    host.add_plugin(Arc::new(OffsetFixture::new()))
        .expect("fixture add");
    let err = host.start().expect_err("dangling deps must fail at start");
    assert!(matches!(err, PluginError::UnknownDep(_)), "got: {err:?}");
}

// —— a3:重名 → Duplicate ——
#[test]
fn a3_duplicate_name() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let ev = Arc::new(EventEngine::new());
    let host = MainEngine::new(Some(ev));
    host.add_plugin(Arc::new(OmsPlugin::new())).expect("first");
    let err = host
        .add_plugin(Arc::new(OmsPlugin::new()))
        .expect_err("dup");
    assert!(matches!(err, PluginError::Duplicate(_)), "got: {err:?}");
}

// ==== B 组:缓存经总线可查 ====

// —— b1:合约命中/miss(miss = Ok(None),业务缺失走载荷) ——
#[test]
fn b1_contract_over_bus() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, _fx) = rig();
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "rb2601", Exchange::SHFE, "rb"),
    );
    let hit: Option<ContractData> = bus_ask(
        &ev,
        "oms",
        GetContractReq {
            en_symbol: "SHFE.rb2601".into(),
        },
    )
    .expect("ok");
    assert_eq!(hit.expect("cached").symbol, "rb2601");
    let miss: Option<ContractData> = bus_ask(
        &ev,
        "oms",
        GetContractReq {
            en_symbol: "SHFE.nope".into(),
        },
    )
    .expect("ok");
    assert!(miss.is_none());
    close_all(&rig.host);
}

// —— b2:持仓与活跃单经总线可查 ——
#[test]
fn b2_positions_and_active_orders() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, _fx) = rig();
    rig.host.start().expect("start");
    let p = position("gx", "IF2506", Exchange::CFFEX, Direction::LONG, 10.0, 4.0);
    put(&ev, W_POSITION, p.clone());
    put(
        &ev,
        W_ORDER,
        order(
            "7",
            Status::NOTTRADED,
            "gx",
            "IF2506",
            Exchange::CFFEX,
            Direction::SHORT,
            Offset::CLOSE,
            4.0,
            0.0,
        ),
    );
    let got: Option<PositionData> = bus_ask(
        &ev,
        "oms",
        GetPositionReq {
            en_positionid: p.en_positionid(),
        },
    )
    .expect("ok");
    let got = got.expect("cached");
    assert!(feq(got.volume, 10.0) && feq(got.yd_volume, 4.0));
    let active: Vec<OrderData> = bus_ask(
        &ev,
        "oms",
        GetAllActiveOrdersReq {
            en_symbol: Some("CFFEX.IF2506".into()),
        },
    )
    .expect("ok");
    assert_eq!(active.len(), 1);
    assert!(feq(active[0].volume, 4.0));
    close_all(&rig.host);
}

// —— b4:三计量经总线可查(0.2.0 新服务面) ——
#[test]
fn b4_counters_over_bus() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, _fx) = rig();
    rig.host.start().expect("start");
    put(
        &ev,
        W_ORDER,
        order(
            "9",
            Status::NOTTRADED,
            "gx",
            "IF2506",
            Exchange::CFFEX,
            Direction::SHORT,
            Offset::CLOSE,
            5.0,
            0.0,
        ),
    );
    put(
        &ev,
        W_ORDER,
        order(
            "9",
            Status::CANCELLED,
            "gx",
            "IF2506",
            Exchange::CFFEX,
            Direction::SHORT,
            Offset::CLOSE,
            5.0,
            0.0,
        ),
    );
    let contract_key = "CFFEX.IF2506";
    let ti: Option<u64> = bus_ask(
        &ev,
        "oms",
        TradeInfoCountReq {
            gateway_name: "gx".into(),
            en_symbol: contract_key.into(),
        },
    )
    .expect("ok");
    assert_eq!(ti, Some(2), "首见 +1、撤单 +1");
    let cc: Option<u64> = bus_ask(
        &ev,
        "oms",
        CancelCountReq {
            gateway_name: "gx".into(),
            en_symbol: contract_key.into(),
        },
    )
    .expect("ok");
    assert_eq!(cc, Some(1));
    let oc: Option<u64> = bus_ask(
        &ev,
        "oms",
        OrderCancelCountReq {
            en_orderid: "gx.9".into(),
        },
    )
    .expect("ok");
    assert_eq!(oc, Some(1));
    close_all(&rig.host);
}

// ==== C 组:互动(核心) ====

// —— c1:动态注入 + 首见回查:合约先到(当时夹具未在场,错过广播)→ 夹具上线
//       后收成交 → 本地 miss → ask oms 命中 → 账本出现;ask 计数 = 1(顺带证明
//       动态 add 即插即上) ——
fn c1_cfg(host: &MainEngine) {
    // 步骤 2 口径:全插件 deps 必含 cfg = 先装配 cfg 收件壳(显式 temp 根)。
    host.add_plugin(Arc::new(ek_cfg::CfgPlugin::new(
        ek_cfg::Store::with_root(std::path::PathBuf::from("target/asm-cfg-rig")),
        false,
    )))
    .expect("cfg add");
}

#[test]
fn c1_dynamic_add_backquery() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let ev = Arc::new(EventEngine::new());
    let host = MainEngine::new(Some(ev.clone()));
    c1_cfg(&host);
    let oms = Arc::new(OmsPlugin::new());
    host.add_plugin(oms).expect("oms add");
    host.start().expect("start");
    // 合约在夹具到场前推送:先确认 oms 已看过它(ask 应答本身就意味着
    // 该事件已被总线派发完 —— put 只保证入队,注册与派发之间存在真窗口,
    // 必须把「夹具错过广播」从假设变成实证),再动态注入。
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "rb2601", Exchange::SHFE, "rb"),
    );
    let cached: Option<ContractData> = bus_ask(
        &ev,
        "oms",
        GetContractReq {
            en_symbol: "SHFE.rb2601".into(),
        },
    )
    .expect("ok");
    assert!(
        cached.is_some(),
        "oms must have cached the contract before fixture joins"
    );
    let fx = Arc::new(OffsetFixture::new());
    host.add_plugin(fx.clone()).expect("dynamic add launches");
    assert_eq!(host.get_state(FIXTURE_NAME), Some(PluginState::Running));
    let t1 = trade(
        "gx",
        "rb2601",
        Exchange::SHFE,
        Direction::LONG,
        Offset::OPEN,
        2.0,
    );
    put(&ev, W_TRADE, t1);
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "SHFE.rb2601".into(),
        },
    )
    .expect("ok");
    let h = h.expect("backquery must build the ledger");
    assert_eq!(h.en_symbol, "SHFE.rb2601");
    assert!(feq(h.long_td, 2.0), "trade ingested after backquery");
    assert_eq!(fx.asks_made(), 1, "exactly one backquery");
    close_all(&host);
}

// —— c2:已收 eContract. → 本地缓存直用(ask 计数 = 0) ——
#[test]
fn c2_local_cache_path() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "IF2506", Exchange::CFFEX, "IF"),
    );
    let t1 = trade(
        "gx",
        "IF2506",
        Exchange::CFFEX,
        Direction::LONG,
        Offset::OPEN,
        3.0,
    );
    put(&ev, W_TRADE, t1);
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "CFFEX.IF2506".into(),
        },
    )
    .expect("ok");
    assert!(feq(h.unwrap().long_td, 3.0));
    assert_eq!(fx.asks_made(), 0, "contract was broadcast; zero asks");
    close_all(&rig.host);
}

// —— c3:双方皆无该合约 → 库层闸门 → 不建账本、不 panic ——
#[test]
fn c3_no_contract_gate() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_TRADE,
        trade(
            "gx",
            "cu2601",
            Exchange::SHFE,
            Direction::LONG,
            Offset::OPEN,
            1.0,
        ),
    );
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "SHFE.cu2601".into(),
        },
    )
    .expect("ok");
    assert!(h.is_none(), "gate must keep the empty ledger absent");
    assert_eq!(fx.asks_made(), 1, "one backquery came up empty");
    close_all(&rig.host);
}

// —— c4:账本服务快照:6 持仓量 + 活跃平仓单夹取(静态冻结四分量联动) ——
#[test]
fn c4_holding_service() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "IF2506", Exchange::CFFEX, "IF"),
    );
    put(
        &ev,
        W_POSITION,
        position("gx", "IF2506", Exchange::CFFEX, Direction::LONG, 10.0, 4.0),
    );
    put(
        &ev,
        W_ORDER,
        order(
            "7",
            Status::NOTTRADED,
            "gx",
            "IF2506",
            Exchange::CFFEX,
            Direction::SHORT,
            Offset::CLOSE,
            4.0,
            0.0,
        ),
    );
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "CFFEX.IF2506".into(),
        },
    )
    .expect("ok");
    let h = h.expect("ledger present");
    assert_eq!(h.en_symbol, "CFFEX.IF2506");
    assert_eq!(h.exchange, Exchange::CFFEX);
    assert!(feq(h.long_pos, 10.0) && feq(h.long_yd, 4.0) && feq(h.long_td, 6.0));
    assert!(feq(h.long_td_frozen, 4.0) && feq(h.long_yd_frozen, 0.0));
    assert!(feq(h.long_pos_frozen, 4.0));
    assert!(feq(h.short_pos, 0.0) && feq(h.short_pos_frozen, 0.0));
    close_all(&rig.host);
}

// ==== D 组:拆单(只走服务路径) ====

// —— d1:SHFE auto 平仓 → 今先拆(今 3 + 昨 2) ——
#[test]
fn d1_shfe_split_via_service() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "rb2601", Exchange::SHFE, "rb"),
    );
    put(
        &ev,
        W_POSITION,
        position("gx", "rb2601", Exchange::SHFE, Direction::LONG, 5.0, 2.0),
    );
    let out: Vec<OrderRequest> = bus_ask(
        &ev,
        FIXTURE_NAME,
        ConvertReq {
            req: req(
                "rb2601",
                Exchange::SHFE,
                Direction::SHORT,
                Offset::CLOSE,
                5.0,
            ),
            gateway_name: "gx".into(),
            lock_mode: LockMode::Auto,
        },
    )
    .expect("ok");
    assert_eq!(out.len(), 2, "td first(yd leg then follows");
    assert_eq!(out[0].offset, Offset::CLOSETODAY);
    assert!(feq(out[0].volume, 3.0));
    assert_eq!(out[1].offset, Offset::CLOSEYESTERDAY);
    assert!(feq(out[1].volume, 2.0));
    close_all(&rig.host);
}

// —— d2:可平不足(6 > 5)→ 空 Vec = 拒单 ——
#[test]
fn d2_reject_insufficient() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "rb2601", Exchange::SHFE, "rb"),
    );
    put(
        &ev,
        W_POSITION,
        position("gx", "rb2601", Exchange::SHFE, Direction::LONG, 5.0, 2.0),
    );
    let out: Vec<OrderRequest> = bus_ask(
        &ev,
        FIXTURE_NAME,
        ConvertReq {
            req: req(
                "rb2601",
                Exchange::SHFE,
                Direction::SHORT,
                Offset::CLOSE,
                6.0,
            ),
            gateway_name: "gx".into(),
            lock_mode: LockMode::Auto,
        },
    )
    .expect("ok");
    assert!(out.is_empty(), "insufficient closeable must reject");
    close_all(&rig.host);
}

// ==== E 组:契约面与收尾 ====

// —— e1:错服务名 → UnknownService ——
#[test]
fn e1_unknown_service() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, _ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    let err = fx.unknown_service_probe().expect_err("no provider");
    assert!(
        matches!(err, AskError::UnknownService { .. }),
        "got: {err:?}"
    );
    close_all(&rig.host);
}

// —— e2:close_all 终态(反序五步;不广播逐插件 ePlugDown → 断言只看宿主侧) ——
#[test]
fn e2_close_terminal_state() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    close_all(&rig.host);
    assert!(rig.host.plugin_names().is_empty(), "registry cleared");
    assert_eq!(rig.host.get_state("oms"), None);
    assert!(!ev.is_active(), "bus stopped by close_all");
}

// ==== F 组:相位门 ====

// —— f1:ePositionDone. 后,夹具不再把持仓摆入账本(镜像原稿两相);oms 走自己的门 ——
#[test]
fn f1_phase_gate() {
    ek_cfg::set_default_root_for_tests(std::path::PathBuf::from("C:/dev/fuke/ek-station/target/cfg-test-root"));
    let (rig, ev, _oms, fx) = rig();
    host_add(&rig, &fx);
    rig.host.start().expect("start");
    put(
        &ev,
        W_CONTRACT,
        contract("gx", "IF2506", Exchange::CFFEX, "IF"),
    );
    put(
        &ev,
        W_POSITION,
        position("gx", "IF2506", Exchange::CFFEX, Direction::LONG, 2.0, 0.0),
    );
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "CFFEX.IF2506".into(),
        },
    )
    .expect("ok");
    assert!(feq(h.unwrap().long_pos, 2.0), "init-phase ingests");
    put(&ev, W_POS_DONE, "gx".to_string());
    put(
        &ev,
        W_POSITION,
        position("gx", "IF2506", Exchange::CFFEX, Direction::LONG, 7.0, 0.0),
    );
    let h: Option<PositionHolding> = bus_ask(
        &ev,
        FIXTURE_NAME,
        GetHoldingReq {
            gateway_name: "gx".into(),
            en_symbol: "CFFEX.IF2506".into(),
        },
    )
    .expect("ok");
    assert!(
        feq(h.unwrap().long_pos, 2.0),
        "run-phase must NOT overwrite the ledger"
    );
    // oms 自身:无成交历史 → 稳定门未开 → 持仓缓存照原稿初始化相处理(7 的快照进字典)。
    let p = position("gx", "IF2506", Exchange::CFFEX, Direction::LONG, 7.0, 0.0);
    let got: Option<PositionData> = bus_ask(
        &ev,
        "oms",
        GetPositionReq {
            en_positionid: p.en_positionid(),
        },
    )
    .expect("ok");
    assert!(feq(got.expect("cached").volume, 7.0));
    close_all(&rig.host);
}