//! I04 验收测试(ADR 0004 末尾表格):
//!
//! 对 Hub 的端到端测试,避开真实子进程(用手工实现的 stdio server 模拟)。
//! 重点验证:
//! - tools/list 聚合并 namespaced 暴露(§4.5,§12.3 I04-1)
//! - tools/call 创建 DecisionRecord(§12.3 I04-2)
//! - Deny 阻断上游(§12.3 I04-3)
//! - Approve 等待阻塞到 resolution(§12.3 I04-4)
//! - 未登记的 server 不暴露工具(§4.5-6)
//! - F1 ThisSession scope 缓存命中跳过 firewall
//! - F3 Outbox draft → approved → executed 四态
//!
//! **注意**:本文件不启动真实子进程。真实 stdio 测试放到跨平台 I05 / I07 测试环境做。
//! 这里用一个 trait 抽象上游代替 StdioUpstream;但 Hub 当前签名要求 Arc<StdioUpstream>,
//! 为了不引入更多破坏性改动,I04 只测试 "Hub 调 firewall 的路径" —— 即测到 firewall
//! 决策为止,上游执行路径由 `attach_upstream` 返回的错误分支覆盖。

#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::err_expect,
    clippy::panic
)]

use std::sync::Arc;
use std::time::Duration;

use serde_json::{json, Value};
use vigil_audit::Ledger;
use vigil_firewall::{scorer::StaticDescriptorOracle, Firewall, FirewallConfig};
use vigil_mcp::protocol::{JsonRpcError, JsonRpcRequest};
use vigil_mcp::upstream::{McpUpstream, UpstreamError};
use vigil_mcp::{Hub, HubConfig};
use vigil_policy::{defaults::default_ruleset, PolicyEngine};
use vigil_types::{ApprovalScope, ServerProfile, TransportKind, TrustLevel};

fn setup_hub() -> (Arc<Ledger>, Arc<Hub>) {
    let l = Arc::new(Ledger::open_in_memory().unwrap());
    let policy = PolicyEngine::new(default_ruleset());
    let fw = Arc::new(Firewall::new(
        l.clone(),
        policy,
        FirewallConfig {
            project_roots: vec!["/proj".into()],
            ..Default::default()
        },
    ));
    // 测试默认 oracle 返 ApprovedStable —— 通过 StaticDescriptorOracle 包成 trait object。
    let oracle: Arc<dyn vigil_firewall::scorer::DescriptorOracle> = Arc::new(
        StaticDescriptorOracle(vigil_firewall::scorer::DescriptorStatus::ApprovedStable),
    );
    let hub = Arc::new(Hub::new(
        l.clone(),
        fw,
        oracle,
        HubConfig {
            approval_wait: Duration::from_millis(200),
            // 测试的 spawn_attach 上游用 dummy(`cmd`/`sh`,不说 MCP);短 upstream_call_timeout
            // 让其 initialize 握手(非致命)快速超时,避免测试卡默认 30s。
            upstream_call_timeout: Duration::from_millis(500),
            ..Default::default()
        },
        vigil_mcp::SecretAliasMap::default(),
    ));
    (l, hub)
}

fn register_and_approve(l: &Ledger, server_id: &str) {
    let p = ServerProfile {
        server_id: server_id.into(),
        transport: TransportKind::Stdio,
        command: Some(vec!["mock".into()]),
        url: None,
        first_seen_at: 0,
        command_hash: Some("abc".into()),
        descriptor_hash: None,
        trust_level: TrustLevel::Untrusted,
        sandbox_profile_id: None,
    };
    l.register_server(&p).unwrap();
    l.approve_server(server_id, TrustLevel::Limited).unwrap();
}

/// 辅助:建 session
fn init_hub(hub: &Hub) {
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(1)),
        method: "initialize".into(),
        params: Some(json!({})),
    };
    let r = hub.handle_request(req).unwrap().unwrap();
    assert!(r.error.is_none());
}

#[test]
fn initialize_creates_session_and_exposes_server_info() {
    let (_l, hub) = setup_hub();
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(1)),
        method: "initialize".into(),
        params: Some(json!({})),
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    let result = resp.result.as_ref().unwrap();
    assert_eq!(result["serverInfo"]["name"], "vigil-hub");
    assert_eq!(result["protocolVersion"], "2025-06-18");
}

#[test]
fn ping_returns_empty_object() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(2)),
        method: "ping".into(),
        params: None,
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    assert_eq!(resp.result, Some(json!({})));
}

#[test]
fn unknown_method_returns_not_found() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(3)),
        method: "resources/list".into(),
        params: None,
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    assert_eq!(
        resp.error.as_ref().unwrap().code,
        JsonRpcError::METHOD_NOT_FOUND
    );
}

#[test]
fn notifications_produce_no_response() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: None, // notification
        method: "notifications/cancelled".into(),
        params: Some(json!({"requestId": 42})),
    };
    let resp = hub.handle_request(req).unwrap();
    assert!(resp.is_none());
}

#[test]
fn tools_list_is_empty_when_no_servers_registered() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(4)),
        method: "tools/list".into(),
        params: None,
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    assert!(resp.error.is_none());
    let tools = resp.result.as_ref().unwrap().get("tools").unwrap();
    assert!(
        tools.as_array().unwrap().is_empty(),
        "方案 §4.5-6: 未登记 server 不暴露"
    );
}

#[test]
fn unapproved_server_does_not_expose_tools() {
    let (l, hub) = setup_hub();
    init_hub(&hub);
    // 登记但**不**审批
    let p = ServerProfile {
        server_id: "fs".into(),
        transport: TransportKind::Stdio,
        command: Some(vec!["mock".into()]),
        url: None,
        first_seen_at: 0,
        command_hash: Some("abc".into()),
        descriptor_hash: None,
        trust_level: TrustLevel::Untrusted,
        sandbox_profile_id: None,
    };
    l.register_server(&p).unwrap();
    // 不调 approve_server

    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(5)),
        method: "tools/list".into(),
        params: None,
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    let tools = resp.result.as_ref().unwrap().get("tools").unwrap();
    assert!(
        tools.as_array().unwrap().is_empty(),
        "方案 §4.5-6: 未审批 server 的 tool 不暴露"
    );
}

#[test]
fn tools_call_on_unknown_tool_returns_upstream_error() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(6)),
        method: "tools/call".into(),
        params: Some(json!({
            "name": "fs__read_file",
            "arguments": {"path": "/proj/x"},
        })),
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    // router 里没这条,Hub 直接返 upstream_unavailable
    let err = resp.error.as_ref().unwrap();
    assert_eq!(err.code, JsonRpcError::VIGIL_UPSTREAM_UNAVAILABLE);
}

#[test]
fn invalid_params_rejected_with_invalid_params_code() {
    let (_l, hub) = setup_hub();
    init_hub(&hub);
    // 缺 params
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(7)),
        method: "tools/call".into(),
        params: None,
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    assert_eq!(
        resp.error.as_ref().unwrap().code,
        JsonRpcError::INVALID_PARAMS
    );

    // 缺 name 字段
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(8)),
        method: "tools/call".into(),
        params: Some(json!({"arguments": {}})),
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    assert_eq!(
        resp.error.as_ref().unwrap().code,
        JsonRpcError::INVALID_PARAMS
    );
}

/// M3(Codex I04 review 二轮):真正通过 `Hub::handle_request("tools/call")` 验证
/// ThisSession scope 命中时**跳过 firewall**且**不产生新 approval**。
///
/// 构造:
/// - 用 Hub test-only API 注入一条 ToolRoute(避开 tools/list 需要真实上游)
/// - 埋一条同 session + 匹配三元组的 ThisSession approval
/// - 调 tools/call:由于没 attach upstream,最终返 upstream_unavailable;
///   **关键断言**:approvals 总数 == 1(没新建),decision.recorded 有 "session-scope-allow"
#[test]
fn m3_hub_tools_call_short_circuits_firewall_on_session_scope() {
    use sha2::{Digest, Sha256};
    use vigil_types::DecisionRecord;

    let (l, hub) = setup_hub();

    // 手动搭建 Hub 内部状态:session_id + tool route(跳过 initialize 和 tools/list)
    let hub_session = l.start_session("m3_e2e", None).unwrap();
    hub.set_session_id_for_test(hub_session.clone()).unwrap();
    hub.inject_route_for_test("fs", "write_file", "hash_abc")
        .unwrap();

    // 埋一条 ThisSession approval(模拟上一次被批准)
    let args = json!({"path": "/proj/x.md"});
    let args_hash = {
        let b = serde_jcs::to_vec(&args).unwrap();
        let mut h = Sha256::new();
        h.update(&b);
        hex::encode(h.finalize())
    };
    let dec = DecisionRecord {
        decision_id: "d-prev".into(),
        invocation_id: "inv-prev".into(),
        decision: vigil_types::DecisionKind::Approve,
        risk_score: 40,
        reasons: vec![],
        policy_ids: vec![],
        created_at: 0,
    };
    let ctx = vigil_audit::ApprovalTargetContext {
        server_id: Some("fs"),
        tool_name: Some("write_file"),
        args_hash: Some(&args_hash),
    };
    let prev_req = l
        .create_approval(
            &hub_session,
            &dec,
            &vigil_types::EffectVector::default(),
            "t",
            "s",
            600,
            ctx,
        )
        .unwrap();
    l.approve(
        &prev_req.approval_id,
        vigil_types::ApprovalScope::ThisSession,
        Some("u"),
    )
    .unwrap();

    let approvals_before = count_approvals(&l);

    // 真正调 tools/call
    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: Some(json!(99)),
        method: "tools/call".into(),
        params: Some(json!({
            "name": "fs__write_file",
            "arguments": args.clone(),
        })),
    };
    let resp = hub.handle_request(req).unwrap().unwrap();
    // 预期:返 upstream_unavailable(没 attach),因为 scope 快路径让我们走进 invoke_upstream
    assert_eq!(
        resp.error.as_ref().unwrap().code,
        JsonRpcError::VIGIL_UPSTREAM_UNAVAILABLE,
        "scope 快路径应一路走到 invoke_upstream"
    );

    // 关键断言 1:没有新 approval 产生(firewall 被跳过)
    let approvals_after = count_approvals(&l);
    assert_eq!(
        approvals_before, approvals_after,
        "M3: 命中 ThisSession scope 时 firewall 不应再创建 approval"
    );

    // 关键断言 2:产生了 "session-scope-allow" 策略的 decision.recorded 事件
    let events = l.replay_session_verified(&hub_session).unwrap();
    let has_scope_decision = events.iter().any(|e| {
        e.event_type == "decision.recorded"
            && e.payload
                .get("policy_ids")
                .and_then(|v| v.as_array())
                .map(|arr| arr.iter().any(|x| x == "session-scope-allow"))
                .unwrap_or(false)
    });
    assert!(
        has_scope_decision,
        "M3: scope 快路径必须产出 session-scope-allow DecisionRecord"
    );
}

/// I05 §12.3 I05-4:command hash 漂移必须拒绝启动(写 audit)
#[test]
fn i05_command_drift_blocks_spawn_and_audits() {
    let (l, hub) = setup_hub();
    init_hub(&hub);
    register_and_approve(&l, "fs");

    // register_and_approve 默认把 fs 的 command_hash 写为 "abc";
    // 我们传一个不同的 argv 去触发 drift。
    let new_argv = vec!["uvx".into(), "mcp-server-fs".into(), "/new/path".into()];
    let err = hub
        .check_upstream_command_drift("fs", &new_argv)
        .expect_err("drift 必须拒绝");
    match err {
        vigil_mcp::HubError::CommandDrift {
            server_id,
            old_hash,
            ..
        } => {
            assert_eq!(server_id, "fs");
            assert_eq!(old_hash, "abc");
        }
        other => panic!("期望 CommandDrift,得到 {:?}", other),
    }

    // 审计里应有 server.command_drifted 事件(用 FTS 跨 session 搜索)
    let hits = l.fts_search("command_drift").unwrap();
    assert!(
        hits.iter()
            .any(|h| h.event_type == "server.command_drifted"),
        "必须写 server.command_drifted 审计事件"
    );
}

/// I05 §12.3 I05-3:descriptor drift 时 tools/list 不暴露 + 写 audit
#[test]
fn i05_tool_descriptor_drift_hidden_from_tools_list() {
    let (l, _hub) = setup_hub();
    // 这里我们直接验证 Ledger 侧的 drift 行为 —— Hub 的 tools/list 要跑上游
    // 太重;对 drift 的单元覆盖由 drift_state_machine.rs 在 vigil-audit 里完成。
    // 本测试只验证:当 Ledger 已有 drift,list_drifted_tools 能查到。
    l.pin_tool_descriptor("fs", "read", "h1").unwrap();
    l.approve_tool_descriptor("fs", "read").unwrap();
    let pin2 = l.pin_tool_descriptor("fs", "read", "h2").unwrap();
    assert!(matches!(pin2, vigil_audit::PinOutcome::Drifted { .. }));
    let drifted = l.list_drifted_tools().unwrap();
    assert_eq!(drifted.len(), 1);
    assert_eq!(drifted[0].current_hash, "h1");
    assert_eq!(drifted[0].proposed_hash.as_deref(), Some("h2"));
}

fn count_approvals(l: &Ledger) -> i64 {
    // 借 list 近似;更严谨可直接 SELECT COUNT(*) —— 但 Ledger 未暴露此 API,
    // 取 replay 事件里 approval.created 的条数作等价信号(每次 create_approval 必产一条)
    let events = l.replay_session("m3_e2e").unwrap_or_default();
    events
        .iter()
        .filter(|e| e.event_type == "approval.created")
        .count() as i64
}

#[test]
fn f1_session_scope_short_circuits_firewall() {
    let (l, hub) = setup_hub();
    init_hub(&hub);
    register_and_approve(&l, "fs");

    // 手工模拟"上一次调用" —— 产出一条 Approved + ThisSession + 匹配 (server, tool, args_hash) 的 approval。
    let session_id = {
        // session 在 initialize 时已创建;取最新 session
        let events = l.replay_session("mcp_hub").unwrap_or_default();
        // 测试中找不到 mcp_hub session 就用 list_approved_servers 的 session id 推断;
        // 这里直接复用 start_session 另建一个同名 session 的做法会破坏 hub 内部;
        // 所以改为直接用 l 的 current session(通过开新 session)让 F1 scope 与 hub 使用同 session。
        let _ = events;
        // 从 approvals 表直接构造(绕开 firewall):
        let dec = vigil_types::DecisionRecord {
            decision_id: "d1".into(),
            invocation_id: "prev".into(),
            decision: vigil_types::DecisionKind::Approve,
            risk_score: 10,
            reasons: vec![],
            policy_ids: vec![],
            created_at: 0,
        };
        // 构造 args_hash 与 tool_call 里相同的算法
        let args = json!({"path": "/proj/x.md"});
        let hash = {
            use sha2::{Digest, Sha256};
            let b = serde_jcs::to_vec(&args).unwrap();
            let mut h = Sha256::new();
            h.update(&b);
            hex::encode(h.finalize())
        };
        // 借 Hub initialize 产的 session 不易拿到,这里新开一个 session 专测 find_session_scope_allow
        let sid = l.start_session("f1_test", None).unwrap();
        let ctx = vigil_audit::ApprovalTargetContext {
            server_id: Some("fs"),
            tool_name: Some("write_file"),
            args_hash: Some(&hash),
        };
        let req = l
            .create_approval(
                &sid,
                &dec,
                &vigil_types::EffectVector::default(),
                "t",
                "s",
                600,
                ctx,
            )
            .unwrap();
        l.approve(&req.approval_id, ApprovalScope::ThisSession, Some("u"))
            .unwrap();
        sid
    };

    let hit = l
        .find_session_scope_allow(&session_id, "fs", "write_file", &{
            use sha2::{Digest, Sha256};
            let b = serde_jcs::to_vec(&json!({"path": "/proj/x.md"})).unwrap();
            let mut h = Sha256::new();
            h.update(&b);
            hex::encode(h.finalize())
        })
        .unwrap();
    assert!(hit.is_some(), "F1 scope 消费应命中");
    assert_eq!(hit.unwrap().scope, Some(ApprovalScope::ThisSession));
}

// ===================== V1.1 gate-before-spawn(ADR 0007 §I-7.1 / ADR 0005)=====================

/// 注册 + 批准一个 stdio server,command_hash = 真 argv hash(避免 argv drift 先触发)。
fn register_approve_stdio(l: &Ledger, server_id: &str, argv: &[String]) {
    let p = ServerProfile {
        server_id: server_id.into(),
        transport: TransportKind::Stdio,
        command: Some(argv.to_vec()),
        url: None,
        first_seen_at: 0,
        command_hash: Some(vigil_audit::argv_hash(argv)),
        descriptor_hash: None,
        trust_level: TrustLevel::Untrusted,
        sandbox_profile_id: None,
    };
    l.register_server(&p).unwrap();
    l.approve_server(server_id, TrustLevel::Limited).unwrap();
}

/// V1.1:resolved-program drift 必须在 spawn **之前**拦截(Codex R2 实施铁律)。
/// 返回 `ResolvedProgramDrift` —— 因 drift 检查是 spawn_attach 第 3 步、spawn 是第 4 步,
/// 返回该错误即结构性证明"进程未 spawn / upstream 未 attach"。
#[test]
fn v11_resolved_program_drift_blocks_before_spawn() {
    let (l, hub) = setup_hub();
    init_hub(&hub);

    // 跨平台可解析的裸命令
    #[cfg(windows)]
    let bare = "cmd";
    #[cfg(not(windows))]
    let bare = "sh";
    let argv = vec![bare.to_string()];
    register_approve_stdio(&l, "shadow", &argv);

    // 预 pin 一个伪基线 → 真实解析路径必 ≠ 它 → 触发 resolved-program drift
    l.check_server_resolved_program_drift("shadow", "/__vigil_bogus_baseline__")
        .unwrap();

    let err = hub
        .spawn_attach_stdio_upstream("shadow", &argv, &[])
        .expect_err("resolved-program drift 必须在 spawn 前拦截");
    assert!(
        matches!(
            err,
            vigil_mcp::HubError::ResolvedProgramDrift { ref server_id, .. } if server_id == "shadow"
        ),
        "期望 ResolvedProgramDrift(gate-before-spawn),得到 {err:?}"
    );
}

/// V1.1:无法解析的裸命令 → fail-closed `StdioSpawn`(resolve 在 spawn 前,且不退化为笼统错)。
#[test]
fn v11_unresolvable_program_fails_closed_before_spawn() {
    let (l, hub) = setup_hub();
    init_hub(&hub);
    let argv = vec!["__vigil_definitely_not_a_real_cmd_zzz__".to_string()];
    register_approve_stdio(&l, "noexist", &argv);

    let err = hub
        .spawn_attach_stdio_upstream("noexist", &argv, &[])
        .expect_err("无法解析的命令必须 fail-closed");
    assert!(
        matches!(err, vigil_mcp::HubError::StdioSpawn(_)),
        "期望 StdioSpawn(ProgramNotFound),得到 {err:?}"
    );
}

#[derive(Debug)]
struct NoopUpstream(String);
impl McpUpstream for NoopUpstream {
    fn server_id(&self) -> &str {
        &self.0
    }
    fn transport(&self) -> TransportKind {
        TransportKind::Stdio
    }
    fn call(&self, _m: &str, _p: Option<Value>, _t: Duration) -> Result<Value, UpstreamError> {
        Ok(json!({}))
    }
    fn shutdown(&self) {}
}

/// V1.1(Codex code R1 BLOCKER):重复 server_id 必须在 **spawn 之前**短路拒绝。
/// 先用 mock attach 占住 "dup",再 spawn_attach 同名但传一个**无法解析**的命令:
/// 若早 dup 检查生效 → 返 `Duplicate`(根本没走到 resolve);若缺早检查 → 会先 resolve
/// 失败返 `StdioSpawn`。断言 `Duplicate` 即证明"重复不 spawn / 不触碰 resolve"。
#[test]
fn v11_duplicate_attach_short_circuits_before_spawn() {
    let (l, hub) = setup_hub();
    init_hub(&hub);
    let argv = vec!["mock".to_string()];
    register_approve_stdio(&l, "dup", &argv);
    // 先占位(mock,非真进程)
    hub.attach_upstream("dup", &argv, Arc::new(NoopUpstream("dup".into())))
        .unwrap();
    // 重复 attach,故意传无法解析的命令 —— 若走到 resolve 必报 StdioSpawn
    let err = hub
        .spawn_attach_stdio_upstream("dup", &["__nonexistent_cmd_zzz__".to_string()], &[])
        .expect_err("重复 server_id 必须拒绝");
    assert!(
        matches!(
            err,
            vigil_mcp::HubError::Namespace(vigil_mcp::namespace::NamespaceError::Duplicate(_))
        ),
        "期望 Duplicate(早 dup 短路,未走 resolve/spawn),得到 {err:?}"
    );
}

/// V1.1(Codex code R2):**并发**同名 attach 必须由 `attach_lock` 串行 —— 恰一个成功、一个
/// `Duplicate`,且失败者**不 spawn**(串行下它在 dup 检查处即返回)。结论不依赖线程调度时序
/// (attach_lock 互斥保证),故确定性而非 flaky。胜者 spawn 一个真进程(`sh`/`cmd`),Hub drop
/// 时关 stdin → 子进程收 EOF 自退。
#[test]
fn v11_concurrent_duplicate_attach_serialized_exactly_one_wins() {
    use std::sync::Barrier;
    let (l, hub) = setup_hub();
    init_hub(&hub);

    #[cfg(windows)]
    let bare = "cmd";
    #[cfg(not(windows))]
    let bare = "sh";
    let argv = vec![bare.to_string()];
    register_approve_stdio(&l, "racer", &argv);

    let barrier = Arc::new(Barrier::new(2));
    let mut handles = Vec::new();
    for _ in 0..2 {
        let hub_c = Arc::clone(&hub);
        let argv_c = argv.clone();
        let bar_c = Arc::clone(&barrier);
        handles.push(std::thread::spawn(move || {
            bar_c.wait(); // 尽量让两线程同时冲 attach_lock
            hub_c.spawn_attach_stdio_upstream("racer", &argv_c, &[])
        }));
    }
    let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
    let ok = results.iter().filter(|r| r.is_ok()).count();
    let dup = results
        .iter()
        .filter(|r| {
            matches!(
                r,
                Err(vigil_mcp::HubError::Namespace(
                    vigil_mcp::namespace::NamespaceError::Duplicate(_)
                ))
            )
        })
        .count();
    assert_eq!(ok, 1, "恰一个成功,得到 {results:?}");
    assert_eq!(
        dup, 1,
        "恰一个 Duplicate(串行,失败者不 spawn),得到 {results:?}"
    );
}
