dev-manager-tauri/src-tauri/src/mcp_server.rs

use axum::{
    extract::{Path, Query, State},
    http::{header::AUTHORIZATION, HeaderMap, StatusCode},
    response::{
        sse::{Event, KeepAlive, Sse},
        IntoResponse, Response,
    },
    routing::{get, post},
    Json, Router,
};
use serde::Deserialize;
use serde_json::{json, Value};
use std::{
    collections::{HashMap, HashSet},
    convert::Infallible,
    net::SocketAddr,
    sync::{Arc, OnceLock, RwLock},
};
use tokio::sync::{mpsc, Mutex};
use tokio_stream::{wrappers::ReceiverStream, StreamExt as _};

use crate::db;

struct SessionEntry {
    sender: mpsc::Sender<(String, String)>,
    project_scope: Option<String>,
}

/// SSE 会话表：session_id → 发送端与连接作用域
type Sessions = Arc<Mutex<HashMap<String, SessionEntry>>>;

const MCP_LOOPBACK_HOST: &str = "127.0.0.1";
const DEV_VARIABLE_READ_TOOL: &str = "read_project_dev_variable";
const TEST_ACCOUNT_READ_TOOL: &str = "read_project_test_account";
const MCP_SERVER_INSTRUCTIONS: &str = "先读取 get_diagnostics、verify_onboarding 或项目状态，再执行后续操作。项目级工具必须使用当前项目的 canonical project_id，不得跨项目猜测或改写身份。测试需要登录时先查询 list_project_dev_variables，已有需求就复用；未登记时可登记不含秘密的开发需求元数据，账号与密码只由用户在项目 UI 填写。其他写操作仅在用户明确确认目标与影响后调用；状态 unavailable/unknown 不得当作成功。";
static INITIALIZED_PROJECT_CLIENTS: OnceLock<RwLock<HashSet<String>>> = OnceLock::new();

fn initialized_project_clients() -> &'static RwLock<HashSet<String>> {
    INITIALIZED_PROJECT_CLIENTS.get_or_init(|| RwLock::new(HashSet::new()))
}

fn record_project_client_initialize(req: &RpcRequest, project_scope: Option<&str>) {
    let Some(project_id) = project_scope else {
        return;
    };
    let client_name = req
        .params
        .as_ref()
        .and_then(|params| params.get("clientInfo"))
        .and_then(|info| info.get("name"))
        .and_then(Value::as_str)
        .unwrap_or_default();
    if client_name == "lian-jing-readiness" {
        return;
    }
    if let Ok(mut clients) = initialized_project_clients().write() {
        clients.insert(project_id.to_string());
    }
}

pub(crate) fn has_initialized_project_client(project_id: &str) -> bool {
    initialized_project_clients()
        .read()
        .map(|clients| clients.contains(project_id))
        .unwrap_or(false)
}

/// MCP 仅监听 IPv4 loopback，避免把本机 agent 控制面暴露到局域网。
fn mcp_bind_addr(port: u16) -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], port))
}

/// 所有 MCP 对外 URL 的单一基址事实。
fn mcp_base_url(port: u16) -> String {
    format!("http://{MCP_LOOPBACK_HOST}:{port}")
}

pub(crate) fn mcp_health_url(port: u16) -> String {
    format!("{}/health", mcp_base_url(port))
}

fn sse_messages_url(port: u16, session_id: &str) -> String {
    format!(
        "{}/mcp/messages?session_id={session_id}",
        mcp_base_url(port)
    )
}

pub(crate) fn scoped_mcp_url(port: u16, project_id: &str) -> String {
    format!(
        "{}/mcp/projects/{}",
        mcp_base_url(port),
        urlencoding::encode(project_id)
    )
}

fn scoped_sse_messages_url(port: u16, project_id: &str, session_id: &str) -> String {
    format!(
        "{}/messages?session_id={session_id}",
        scoped_mcp_url(port, project_id)
    )
}

/// 从 settings 表读取 MCP 端口，默认 27190
pub(crate) fn read_port_from_conn(conn: &rusqlite::Connection) -> u16 {
    conn.query_row(
        "SELECT value FROM settings WHERE key = 'mcp_port'",
        [],
        |row| row.get::<_, String>(0),
    )
    .ok()
    .and_then(|value| value.parse().ok())
    .unwrap_or(27190)
}

pub fn read_port() -> u16 {
    db::pool()
        .get()
        .ok()
        .map(|conn| read_port_from_conn(&conn))
        .unwrap_or(27190)
}

/// 启动 MCP HTTP 服务（同时支持 Streamable HTTP 和 SSE 两种传输）
pub async fn start(port: u16) {
    let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
    let app = mcp_router(sessions);

    let addr = mcp_bind_addr(port);
    match tokio::net::TcpListener::bind(addr).await {
        Ok(listener) => {
            db::log_event(
                "mcp_server",
                "info",
                &format!("MCP Server 已启动，端口 {port}（HTTP + SSE）"),
                None,
            );
            if let Err(e) = axum::serve(listener, app).await {
                db::log_event(
                    "mcp_server",
                    "error",
                    &format!("Server 异常退出: {e}"),
                    None,
                );
            }
        }
        Err(e) => db::log_event(
            "mcp_server",
            "error",
            &format!("绑定端口 {port} 失败: {e}"),
            None,
        ),
    }
}

fn mcp_router(sessions: Sessions) -> Router {
    Router::new()
        .route("/health", get(health))
        // 单一全局服务器路由（lian-jing）：寻址不依赖 group_id，工具显式传 project_id
        .route("/mcp", post(handle_mcp).get(handle_sse_connect))
        .route(
            "/mcp/projects/:project_id",
            post(handle_scoped_mcp).get(handle_scoped_sse_connect),
        )
        // SSE 消息端点：客户端通过此 URL 发送 JSON-RPC，响应经 SSE 推回
        .route("/mcp/messages", post(handle_sse_message))
        .route(
            "/mcp/projects/:project_id/messages",
            post(handle_scoped_sse_message),
        )
        // Gitea webhook：接收 push/pull_request 事件写入飞轮
        .route(
            "/webhook/gitea/:project_id",
            post(crate::mcp_webhook::handle_gitea_webhook),
        )
        .with_state(sessions)
}

async fn health() -> &'static str {
    "炼境 MCP Server running"
}

// ── JSON-RPC 类型 ─────────────────────────────────────────────────────────────

#[derive(Deserialize)]
struct RpcRequest {
    #[allow(dead_code)]
    jsonrpc: String,
    method: String,
    id: Option<Value>,
    #[serde(default)]
    params: Option<Value>,
}

#[derive(Deserialize)]
struct SessionQuery {
    session_id: String,
}

/// 仅敏感单项读取要求项目级认证；旧的非敏感 MCP 工具保持原有契约。
fn authorize_sensitive_request(
    req: &RpcRequest,
    project_scope: Option<&str>,
    headers: &HeaderMap,
    store: &impl crate::credential_store::CredentialStore,
) -> Result<(), &'static str> {
    let requested_tool = req
        .params
        .as_ref()
        .and_then(|params| params.get("name"))
        .and_then(Value::as_str);
    if req.method != "tools/call"
        || !matches!(
            requested_tool,
            Some(DEV_VARIABLE_READ_TOOL | TEST_ACCOUNT_READ_TOOL)
        )
    {
        return Ok(());
    }
    let project_id = project_scope.ok_or("敏感工具仅允许项目级 MCP 连接")?;
    let requested_project = req
        .params
        .as_ref()
        .and_then(|params| params.get("arguments"))
        .and_then(|args| args.get("project_id"))
        .and_then(Value::as_str);
    if requested_project.is_some_and(|requested| requested != project_id) {
        return Err("项目作用域不匹配");
    }
    let mut auth_values = headers.get_all(AUTHORIZATION).iter();
    let header = auth_values
        .next()
        .and_then(|header| header.to_str().ok())
        .ok_or("缺少 MCP 客户端认证")?;
    if auth_values.next().is_some() {
        return Err("MCP 客户端认证格式无效");
    }
    let provided = header
        .strip_prefix("Bearer ")
        .ok_or("MCP 客户端认证格式无效")?;
    let digest =
        crate::credential_store::mcp_auth_digest(project_id).map_err(|_| "MCP 项目认证不可用")?;
    let expected = crate::credential_store::read_mcp_auth_token_for_digest(store, &digest)
        .map_err(|_| "MCP 项目认证不可用")?;
    let equal = provided.len() == expected.len()
        && provided
            .bytes()
            .zip(expected.bytes())
            .fold(0u8, |difference, (left, right)| difference | (left ^ right))
            == 0;
    if !equal {
        return Err("MCP 客户端认证失败");
    }
    Ok(())
}

// ── 共享 RPC 处理逻辑 ─────────────────────────────────────────────────────────

async fn process_rpc(req: &RpcRequest, project_scope: Option<&str>) -> Result<Value, String> {
    if req.method == "initialize" {
        record_project_client_initialize(req, project_scope);
    }
    match req.method.as_str() {
        "initialize" => Ok(json!({
            "protocolVersion": "2024-11-05",
            "capabilities": { "tools": {} },
            "serverInfo": { "name": "炼境", "version": env!("CARGO_PKG_VERSION") },
            "instructions": MCP_SERVER_INSTRUCTIONS,
            "lianJing": match project_scope {
                Some(project_id) => json!({
                    "scope_mode": "project",
                    "project_id": project_id
                }),
                None => json!({
                    "scope_mode": "legacy",
                    "upgrade_hint": "重跑 inject_mcp，将项目配置升级到 /mcp/projects/:project_id"
                })
            }
        })),
        "tools/list" => Ok(match project_scope {
            Some(project_id) => crate::mcp_tools::tools_list_result_for_project(project_id),
            None => crate::mcp_tools::tools_list_result(),
        }),
        "tools/call" => Ok(match project_scope {
            Some(project_id) => {
                crate::mcp_tools::tools_call_for_project(req.params.as_ref(), project_id).await
            }
            None => crate::mcp_tools::tools_call(req.params.as_ref()).await,
        }),
        _ => Err(format!("Method not found: {}", req.method)),
    }
}

// ── Streamable HTTP（POST）─────────────────────────────────────────────────────

async fn handle_mcp(headers: HeaderMap, Json(req): Json<RpcRequest>) -> Response {
    if authorize_sensitive_request(
        &req,
        None,
        &headers,
        &crate::credential_store::WindowsCredentialStore,
    )
    .is_err()
    {
        return StatusCode::FORBIDDEN.into_response();
    }
    if req.id.is_none() {
        return StatusCode::ACCEPTED.into_response();
    }
    let id = req.id.clone();
    match process_rpc(&req, None).await {
        Ok(result) => Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response(),
        Err(msg) => Json(json!({
            "jsonrpc": "2.0", "id": id,
            "error": { "code": -32601, "message": msg }
        }))
        .into_response(),
    }
}

async fn handle_scoped_mcp(
    Path(project_id): Path<String>,
    headers: HeaderMap,
    Json(req): Json<RpcRequest>,
) -> Response {
    if authorize_sensitive_request(
        &req,
        Some(&project_id),
        &headers,
        &crate::credential_store::WindowsCredentialStore,
    )
    .is_err()
    {
        return StatusCode::FORBIDDEN.into_response();
    }
    if req.id.is_none() {
        return StatusCode::ACCEPTED.into_response();
    }
    let id = req.id.clone();
    match process_rpc(&req, Some(&project_id)).await {
        Ok(result) => Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response(),
        Err(msg) => Json(json!({
            "jsonrpc": "2.0", "id": id,
            "error": { "code": -32601, "message": msg }
        }))
        .into_response(),
    }
}

// ── SSE 传输（GET + POST /messages）─────────────────────────────────────────

/// GET /mcp — Claude Code SSE 握手，首条事件告知后续 POST URL
async fn handle_sse_connect(
    State(sessions): State<Sessions>,
) -> Sse<impl futures_core::Stream<Item = Result<Event, Infallible>>> {
    let session_id = uuid::Uuid::new_v4().to_string();
    let (tx, rx) = mpsc::channel::<(String, String)>(32);

    let port = read_port();
    let endpoint_uri = sse_messages_url(port, &session_id);
    let _ = tx.send(("endpoint".to_string(), endpoint_uri)).await;
    sessions.lock().await.insert(
        session_id,
        SessionEntry {
            sender: tx,
            project_scope: None,
        },
    );

    let stream = ReceiverStream::new(rx).map(|(evt_name, data)| {
        Ok::<Event, Infallible>(Event::default().event(evt_name).data(data))
    });

    Sse::new(stream).keep_alive(KeepAlive::default())
}

async fn handle_scoped_sse_connect(
    Path(project_id): Path<String>,
    State(sessions): State<Sessions>,
) -> Sse<impl futures_core::Stream<Item = Result<Event, Infallible>>> {
    let session_id = uuid::Uuid::new_v4().to_string();
    let (tx, rx) = mpsc::channel::<(String, String)>(32);
    let endpoint_uri = scoped_sse_messages_url(read_port(), &project_id, &session_id);
    let _ = tx.send(("endpoint".to_string(), endpoint_uri)).await;
    sessions.lock().await.insert(
        session_id,
        SessionEntry {
            sender: tx,
            project_scope: Some(project_id),
        },
    );
    let stream = ReceiverStream::new(rx).map(|(evt_name, data)| {
        Ok::<Event, Infallible>(Event::default().event(evt_name).data(data))
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::credential_store::{get_or_create_mcp_auth_token, MemoryCredentialStore};

    fn rpc(method: &str) -> RpcRequest {
        RpcRequest {
            jsonrpc: "2.0".to_string(),
            method: method.to_string(),
            id: Some(json!(1)),
            params: None,
        }
    }

    #[test]
    fn sensitive_read_requires_matching_project_bearer_on_both_transports() {
        let store = MemoryCredentialStore::new();
        let p1 = get_or_create_mcp_auth_token(&store, "p1").unwrap();
        let _p2 = get_or_create_mcp_auth_token(&store, "p2").unwrap();
        let req = RpcRequest {
            jsonrpc: "2.0".into(),
            method: "tools/call".into(),
            id: Some(json!(1)),
            params: Some(json!({
                "name": DEV_VARIABLE_READ_TOOL,
                "arguments": {"project_id": "p1", "name": "E2E_PASSWORD"}
            })),
        };
        let mut headers = HeaderMap::new();
        assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_err());
        headers.insert(AUTHORIZATION, "Bearer wrong".parse().unwrap());
        assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_err());
        headers.insert(AUTHORIZATION, format!("Bearer {p1}").parse().unwrap());
        assert!(authorize_sensitive_request(&req, Some("p1"), &headers, &store).is_ok());
        assert!(authorize_sensitive_request(&req, Some("p2"), &headers, &store).is_err());
        assert!(authorize_sensitive_request(&req, None, &headers, &store).is_err());

        let test_account = RpcRequest {
            params: Some(json!({
                "name": TEST_ACCOUNT_READ_TOOL,
                "arguments": {"project_id": "p1", "name": "E2E_LOGIN"}
            })),
            ..rpc("tools/call")
        };
        assert!(
            authorize_sensitive_request(&test_account, Some("p1"), &HeaderMap::new(), &store)
                .is_err()
        );
        assert!(authorize_sensitive_request(&test_account, Some("p1"), &headers, &store).is_ok());
        assert!(authorize_sensitive_request(&test_account, Some("p2"), &headers, &store).is_err());
        assert!(authorize_sensitive_request(&test_account, None, &headers, &store).is_err());

        let mut implicit_project = req;
        implicit_project.params.as_mut().unwrap()["arguments"] = json!({"name": "E2E_PASSWORD"});
        assert!(
            authorize_sensitive_request(&implicit_project, Some("p1"), &headers, &store).is_ok()
        );
        assert!(authorize_sensitive_request(
            &rpc("initialize"),
            Some("p1"),
            &HeaderMap::new(),
            &store
        )
        .is_ok());
        let normal = RpcRequest {
            method: "tools/call".into(),
            params: Some(json!({"name": "get_diagnostics", "arguments": {}})),
            ..rpc("tools/call")
        };
        assert!(
            authorize_sensitive_request(&normal, Some("p1"), &HeaderMap::new(), &store).is_ok()
        );
    }

    /// 单一 lian-jing 服务器的握手路径：initialize / tools/list 必须可用
    #[tokio::test]
    async fn process_rpc_handshake() {
        let init = process_rpc(&rpc("initialize"), None).await.unwrap();
        assert_eq!(init["serverInfo"]["name"], "炼境");
        let instructions = init["instructions"].as_str().unwrap();
        assert!(instructions.len() <= 512);
        assert!(instructions.contains("先读取"));
        assert!(instructions.contains("项目"));
        assert!(instructions.contains("用户明确确认"));
        assert_eq!(init["lianJing"]["scope_mode"], "legacy");
        assert!(init["lianJing"]["upgrade_hint"].is_string());

        let tools = process_rpc(&rpc("tools/list"), None).await.unwrap();
        let list = tools["tools"].as_array().unwrap();
        assert!(
            list.len() >= 20,
            "工具清单不应因去组化而缺失，实际 {}",
            list.len()
        );
        assert!(
            list.iter()
                .any(|tool| tool["name"] == "get_delivery_status"),
            "MCP 握手后的工具清单应暴露 get_delivery_status"
        );
    }

    #[tokio::test]
    async fn scoped_handshake_binds_project_and_filters_global_tools() {
        let init = process_rpc(&rpc("initialize"), Some("p1")).await.unwrap();
        assert_eq!(init["lianJing"]["scope_mode"], "project");
        assert_eq!(init["lianJing"]["project_id"], "p1");

        let tools = process_rpc(&rpc("tools/list"), Some("p1")).await.unwrap();
        let list = tools["tools"].as_array().unwrap();
        assert!(list
            .iter()
            .any(|tool| tool["name"] == "get_delivery_status"));
        assert!(!list.iter().any(|tool| tool["name"] == "unregister_project"));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn scoped_enterprise_fixture_reads_non_secret_resource_status_over_mcp() {
        db::init_test_pool();
        let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        {
            let conn = db::pool().get().unwrap();
            conn.execute("INSERT OR IGNORE INTO project_profiles (id, name) VALUES ('enterprise-fixture-profile', 'Enterprise fixture')", []).unwrap();
            conn.execute("INSERT OR IGNORE INTO project_workspaces (id, profile_id) VALUES ('enterprise-fixture', 'enterprise-fixture-profile')", []).unwrap();
            conn.execute(
                "UPDATE settings SET value=?1 WHERE key='mcp_port'",
                [port.to_string()],
            )
            .unwrap();
        }
        let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
        let server = tokio::spawn(async move {
            axum::serve(listener, mcp_router(sessions)).await.unwrap();
        });
        let url = format!("http://127.0.0.1:{port}/mcp/projects/enterprise-fixture");
        let call = |method: &str, params: Value| -> Value {
            ureq::post(&url)
                .send_json(json!({"jsonrpc":"2.0","id":1,"method":method,"params":params}))
                .unwrap()
                .into_json()
                .unwrap()
        };
        let init = call("initialize", json!({}));
        assert_eq!(
            init["result"]["lianJing"]["project_id"],
            "enterprise-fixture"
        );
        let catalog = call("tools/list", json!({}));
        assert!(catalog["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .any(|tool| tool["name"] == "get_project_resource_status"));
        let result = call(
            "tools/call",
            json!({"name":"get_project_resource_status","arguments":{}}),
        );
        assert_eq!(
            result["result"]["structuredContent"]["project_id"],
            "enterprise-fixture"
        );
        assert_eq!(
            result["result"]["structuredContent"]["gitea_pr"]["state"],
            "not_configured"
        );
        assert!(!result.to_string().contains("access_token"));
        let denied = call(
            "tools/call",
            json!({"name":"get_project_resource_status","arguments":{"project_id":"other-project"}}),
        );
        assert_eq!(
            denied["result"]["structuredContent"]["code"],
            "scope_violation"
        );

        use std::io::BufRead;
        let sse = ureq::get(&url).call().unwrap();
        assert!(sse
            .header("Content-Type")
            .unwrap_or("")
            .contains("text/event-stream"));
        let mut line = String::new();
        std::io::BufReader::new(sse.into_reader())
            .read_line(&mut line)
            .unwrap();
        assert!(line.contains("endpoint"));
        server.abort();
    }

    #[cfg(windows)]
    #[tokio::test(flavor = "multi_thread")]
    async fn scoped_dev_variable_http_pilot_denies_missing_wrong_cross_and_global_auth() {
        struct Cleanup {
            ids: Vec<(String, String)>,
        }
        impl Drop for Cleanup {
            fn drop(&mut self) {
                let store = crate::credential_store::WindowsCredentialStore;
                for (project_id, profile_id) in &self.ids {
                    if let Ok(conn) = db::pool().get() {
                        let _ = crate::commands::dev_variables::revoke(
                            &conn,
                            &store,
                            project_id,
                            "E2E_PASSWORD",
                        );
                        let _ = crate::commands::dev_variables::revoke(
                            &conn,
                            &store,
                            project_id,
                            "E2E_LOGIN",
                        );
                        let _ = conn
                            .execute("DELETE FROM project_workspaces WHERE id=?1", [project_id]);
                        let _ =
                            conn.execute("DELETE FROM project_profiles WHERE id=?1", [profile_id]);
                    }
                    let _ = crate::credential_store::delete_mcp_auth_token_for_project(
                        &store, project_id,
                    );
                }
            }
        }

        db::init_test_pool();
        let p1 = format!("qa-only-dev-mcp-{}", uuid::Uuid::new_v4());
        let p2 = format!("qa-only-dev-mcp-{}", uuid::Uuid::new_v4());
        let profile1 = format!("{p1}-profile");
        let profile2 = format!("{p2}-profile");
        let _cleanup = Cleanup {
            ids: vec![
                (p1.clone(), profile1.clone()),
                (p2.clone(), profile2.clone()),
            ],
        };
        {
            let conn = db::pool().get().unwrap();
            for (project_id, profile_id) in [(&p1, &profile1), (&p2, &profile2)] {
                conn.execute(
                    "INSERT INTO project_profiles (id,name) VALUES (?1,'QA')",
                    [profile_id],
                )
                .unwrap();
                conn.execute(
                    "INSERT INTO project_workspaces (id,profile_id) VALUES (?1,?2)",
                    rusqlite::params![project_id, profile_id],
                )
                .unwrap();
            }
        }
        let store = crate::credential_store::WindowsCredentialStore;
        let token = crate::credential_store::get_or_create_mcp_auth_token(&store, &p1).unwrap();
        let _other_token =
            crate::credential_store::get_or_create_mcp_auth_token(&store, &p2).unwrap();
        let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
        let server = tokio::spawn(async move {
            axum::serve(listener, mcp_router(sessions)).await.unwrap();
        });
        let p1_url = format!("http://127.0.0.1:{port}/mcp/projects/{p1}");
        let p2_url = format!("http://127.0.0.1:{port}/mcp/projects/{p2}");
        let global_url = format!("http://127.0.0.1:{port}/mcp");
        let call_body = |name: &str, arguments: Value| {
            json!({
                "jsonrpc":"2.0","id":1,"method":"tools/call",
                "params":{"name":name,"arguments":arguments}
            })
        };
        let scoped_catalog = ureq::post(&p1_url)
            .send_json(json!({"jsonrpc":"2.0","id":1,"method":"tools/list"}))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        let tool_names: Vec<&str> = scoped_catalog["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(|tool| tool["name"].as_str())
            .collect();
        assert!(tool_names.contains(&"read_project_dev_variable"));
        assert!(tool_names.contains(&"register_project_test_account"));
        assert!(tool_names.contains(&"read_project_test_account"));
        assert!(!tool_names
            .iter()
            .any(|name| name.contains("get_secret") || name.contains("export_env")));
        let global_catalog = ureq::post(&global_url)
            .send_json(json!({"jsonrpc":"2.0","id":1,"method":"tools/list"}))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert!(!global_catalog
            .to_string()
            .contains("read_project_dev_variable"));
        assert!(!global_catalog
            .to_string()
            .contains("read_project_test_account"));
        let register = ureq::post(&p1_url)
            .send_json(call_body(
                "register_project_dev_variable",
                json!({"name":"E2E_PASSWORD","purpose":"合成验收"}),
            ))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert!(register["result"]["isError"].is_null());
        let pending = ureq::post(&p1_url)
            .send_json(call_body("list_project_dev_variables", json!({})))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert!(pending["result"]["content"][0]["text"]
            .as_str()
            .unwrap()
            .contains("\"configured\":false"));
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::save(
                &conn,
                &store,
                &p1,
                "E2E_PASSWORD",
                "qa-only-http-value",
            )
            .unwrap();
        }
        let read_call = call_body("read_project_dev_variable", json!({"name":"E2E_PASSWORD"}));
        assert!(matches!(
            ureq::post(&p1_url).send_json(read_call.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&p1_url)
                .set("Authorization", "Bearer wrong")
                .send_json(read_call.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&p2_url)
                .set("Authorization", &format!("Bearer {token}"))
                .send_json(read_call.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        let cross_argument = call_body(
            "read_project_dev_variable",
            json!({"project_id":p2,"name":"E2E_PASSWORD"}),
        );
        assert!(matches!(
            ureq::post(&p1_url)
                .set("Authorization", &format!("Bearer {token}"))
                .send_json(cross_argument),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&global_url)
                .set("Authorization", &format!("Bearer {token}"))
                .send_json(read_call.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        let read = ureq::post(&p1_url)
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(read_call.clone())
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert_eq!(read["result"]["content"][0]["text"], "qa-only-http-value");
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::revoke(&conn, &store, &p1, "E2E_PASSWORD").unwrap();
        }
        let after_revoke = ureq::post(&p1_url)
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(read_call)
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert_eq!(after_revoke["result"]["isError"], true);
        assert!(!after_revoke.to_string().contains("qa-only-http-value"));

        let account_register = ureq::post(&p1_url)
            .send_json(call_body(
                "register_project_test_account",
                json!({"name":"E2E_LOGIN","service":"enterprise-system","environment":"test","purpose":"合成登录验收"}),
            ))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert!(account_register["result"]["isError"].is_null());
        let account_pending = ureq::post(&p1_url)
            .send_json(call_body("list_project_dev_variables", json!({})))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        let pending_text = account_pending["result"]["content"][0]["text"]
            .as_str()
            .unwrap();
        assert!(pending_text.contains("\"kind\":\"test_account\""));
        assert!(pending_text.contains("\"configured\":false"));
        assert!(!pending_text.contains("qa-only-account-user"));
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::save_test_account(
                &conn,
                &store,
                &p1,
                "E2E_LOGIN",
                "qa-only-account-user",
                "qa-only-account-password",
            )
            .unwrap();
        }
        let account_registered_again = ureq::post(&p1_url)
            .send_json(call_body(
                "register_project_test_account",
                json!({"name":"E2E_LOGIN","service":"enterprise-system","environment":"test","purpose":"合成登录验收"}),
            ))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert!(account_registered_again["result"]["isError"].is_null());
        let account_ready = ureq::post(&p1_url)
            .send_json(call_body("list_project_dev_variables", json!({})))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        let ready_text = account_ready["result"]["content"][0]["text"]
            .as_str()
            .unwrap();
        assert_eq!(ready_text.matches("\"name\":\"E2E_LOGIN\"").count(), 1);
        assert!(ready_text.contains("\"configured\":true"));
        assert!(!ready_text.contains("qa-only-account-user"));
        assert!(!ready_text.contains("qa-only-account-password"));
        let account_read = call_body("read_project_test_account", json!({"name":"E2E_LOGIN"}));
        assert!(matches!(
            ureq::post(&p1_url).send_json(account_read.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&p1_url)
                .set("Authorization", "Bearer wrong")
                .send_json(account_read.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&p2_url)
                .set("Authorization", &format!("Bearer {token}"))
                .send_json(account_read.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        assert!(matches!(
            ureq::post(&global_url)
                .set("Authorization", &format!("Bearer {token}"))
                .send_json(account_read.clone()),
            Err(ureq::Error::Status(403, _))
        ));
        let account = ureq::post(&p1_url)
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(account_read.clone())
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        let value: Value =
            serde_json::from_str(account["result"]["content"][0]["text"].as_str().unwrap())
                .unwrap();
        assert_eq!(value["username"], "qa-only-account-user");
        assert_eq!(value["password"], "qa-only-account-password");
        let scalar_read = ureq::post(&p1_url)
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(call_body(
                "read_project_dev_variable",
                json!({"name":"E2E_LOGIN"}),
            ))
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert_eq!(scalar_read["result"]["isError"], true);
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::revoke(&conn, &store, &p1, "E2E_LOGIN").unwrap();
        }
        let account_after_revoke = ureq::post(&p1_url)
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(account_read)
            .unwrap()
            .into_json::<Value>()
            .unwrap();
        assert_eq!(account_after_revoke["result"]["isError"], true);
        assert!(!account_after_revoke
            .to_string()
            .contains("qa-only-account-password"));
        server.abort();
    }

    #[cfg(windows)]
    #[tokio::test(flavor = "multi_thread")]
    async fn scoped_mcp_create_pull_request_uses_synthetic_credential() {
        use std::io::{BufRead, BufReader, Read, Write};

        struct SyntheticInstanceGuard(Option<String>);
        impl Drop for SyntheticInstanceGuard {
            fn drop(&mut self) {
                if let Some(instance_id) = self.0.take() {
                    let _ = crate::commands::gitea::gitea_delete_instance(instance_id);
                }
            }
        }

        db::init_test_pool();
        let project_id = format!("synthetic-broker-{}", uuid::Uuid::new_v4());
        let profile_id = format!("{project_id}-profile");
        let token = "qa-only-synthetic-mcp-token";
        let remote = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let remote_address = remote.local_addr().unwrap();
        let instance = crate::commands::gitea::gitea_save_instance(
            crate::commands::gitea::SaveGiteaInstanceInput {
                name: "synthetic MCP PR".into(),
                url: format!("http://{remote_address}"),
                access_token: token.into(),
                is_default: Some(false),
            },
        )
        .unwrap();
        let mut instance_guard = SyntheticInstanceGuard(Some(instance.id.clone()));
        let reference = {
            let conn = db::pool().get().unwrap();
            conn.query_row(
                "SELECT access_token_ref FROM gitea_instances WHERE id = ?1",
                [&instance.id],
                |row| row.get::<_, String>(0),
            )
            .unwrap()
        };
        {
            let conn = db::pool().get().unwrap();
            conn.execute(
                "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic MCP PR')",
                [&profile_id],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)",
                rusqlite::params![project_id, profile_id],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES (?1, ?2, ?3, 'owner', 'repo')",
                rusqlite::params![uuid::Uuid::new_v4().to_string(), instance.id, project_id],
            )
            .unwrap();
        }

        let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
        let server = tokio::spawn(async move {
            axum::serve(listener, mcp_router(sessions)).await.unwrap();
        });
        let url = format!("http://127.0.0.1:{port}/mcp/projects/{project_id}");
        let call = |method: &str, params: Value| -> Value {
            ureq::post(&url)
                .send_json(json!({"jsonrpc":"2.0","id":1,"method":method,"params":params}))
                .unwrap()
                .into_json()
                .unwrap()
        };

        let init = call("initialize", json!({}));
        assert_eq!(init["result"]["lianJing"]["project_id"], project_id);
        let catalog = call("tools/list", json!({}));
        assert!(catalog["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .any(|tool| tool["name"] == "create_pull_request"));
        let readiness = call(
            "tools/call",
            json!({"name":"get_project_resource_status","arguments":{}}),
        );
        assert_eq!(
            readiness["result"]["structuredContent"]["gitea_pr"]["state"],
            "configured"
        );
        assert!(!readiness.to_string().contains(token));
        let denied = call(
            "tools/call",
            json!({"name":"create_pull_request","arguments":{"project_id":"other-project","head":"feature","base":"main","title":"synthetic"}}),
        );
        assert_eq!(
            denied["result"]["structuredContent"]["code"],
            "scope_violation"
        );

        remote.set_nonblocking(true).unwrap();
        let mock = std::thread::spawn(move || {
            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
            let (mut stream, _) = loop {
                match remote.accept() {
                    Ok(accepted) => break accepted,
                    Err(error)
                        if error.kind() == std::io::ErrorKind::WouldBlock
                            && std::time::Instant::now() < deadline =>
                    {
                        std::thread::sleep(std::time::Duration::from_millis(10));
                    }
                    Err(error) => panic!("模拟 Gitea 未收到请求: {error}"),
                }
            };
            stream
                .set_read_timeout(Some(std::time::Duration::from_secs(3)))
                .unwrap();
            let mut reader = BufReader::new(stream.try_clone().unwrap());
            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            assert!(line.starts_with("POST /api/v1/repos/owner/repo/pulls "));
            let mut authorized = false;
            let mut content_length = 0;
            loop {
                line.clear();
                reader.read_line(&mut line).unwrap();
                if line == "\r\n" || line.is_empty() {
                    break;
                }
                if line.eq_ignore_ascii_case("authorization: token qa-only-synthetic-mcp-token\r\n")
                {
                    authorized = true;
                }
                if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
                    content_length = value.trim().parse().unwrap();
                }
            }
            assert!(authorized, "模拟 Gitea 未收到受保护存储中的合成 token");
            let mut body = vec![0; content_length];
            reader.read_exact(&mut body).unwrap();
            let request: Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(request["head"], "feature");
            let response =
                r#"{"number":42,"html_url":"https://example.invalid/pr/42","title":"synthetic"}"#;
            write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap();
        });
        let created = call(
            "tools/call",
            json!({"name":"create_pull_request","arguments":{"head":"feature","base":"main","title":"synthetic"}}),
        );
        mock.join().unwrap();
        assert_ne!(created["result"]["isError"], true);
        let result = created["result"]["content"][0]["text"].as_str().unwrap();
        assert_eq!(
            serde_json::from_str::<Value>(result).unwrap()["pr_number"],
            42
        );
        assert!(!created.to_string().contains(token));
        assert!(!created.to_string().contains("access_token"));
        server.abort();

        crate::commands::gitea::gitea_delete_instance(instance.id).unwrap();
        instance_guard.0 = None;
        assert!(crate::credential_store::CredentialStore::get(
            &crate::credential_store::WindowsCredentialStore,
            &reference,
        )
        .unwrap()
        .is_none());
    }

    /// 仅供隔离的真实客户端试点；普通 cargo test 不运行，也不启动 Tauri 应用。
    #[cfg(windows)]
    #[tokio::test(flavor = "multi_thread")]
    #[ignore = "connect a disposable Codex/Claude project to the printed scoped MCP URL"]
    async fn isolated_scoped_mcp_client_pilot() {
        use std::io::{BufRead, BufReader, Read, Write};

        struct PilotGuard {
            instance_id: String,
            project_dir: std::path::PathBuf,
        }
        impl Drop for PilotGuard {
            fn drop(&mut self) {
                let _ = crate::commands::gitea::gitea_delete_instance(self.instance_id.clone());
                let _ = std::fs::remove_file(self.project_dir.join(".lian-jing/context.json"));
                let _ = std::fs::remove_file(self.project_dir.join(".codex/config.toml"));
                let _ = std::fs::remove_dir(self.project_dir.join(".lian-jing"));
                let _ = std::fs::remove_dir(self.project_dir.join(".codex"));
                let _ = std::fs::remove_dir(&self.project_dir);
            }
        }

        db::init_test_pool();
        let project_id = format!("synthetic-client-pilot-{}", uuid::Uuid::new_v4());
        let profile_id = format!("{project_id}-profile");
        let token = format!("qa-only-pilot-{}", uuid::Uuid::new_v4());
        let remote = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let remote_address = remote.local_addr().unwrap();
        let instance = crate::commands::gitea::gitea_save_instance(
            crate::commands::gitea::SaveGiteaInstanceInput {
                name: "isolated client pilot".into(),
                url: format!("http://{remote_address}"),
                access_token: token.clone(),
                is_default: Some(false),
            },
        )
        .unwrap();
        let project_dir =
            std::env::temp_dir().join(format!("lianjing-r11-{}", uuid::Uuid::new_v4()));
        let guard = PilotGuard {
            instance_id: instance.id.clone(),
            project_dir: project_dir.clone(),
        };
        let reference: String = {
            let conn = db::pool().get().unwrap();
            conn.query_row(
                "SELECT access_token_ref FROM gitea_instances WHERE id = ?1",
                [&instance.id],
                |row| row.get(0),
            )
            .unwrap()
        };
        {
            let conn = db::pool().get().unwrap();
            conn.execute(
                "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic Client Pilot')",
                [&profile_id],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)",
                rusqlite::params![project_id, profile_id],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES (?1, ?2, ?3, 'owner', 'repo')",
                rusqlite::params![uuid::Uuid::new_v4().to_string(), instance.id, project_id],
            )
            .unwrap();
        }

        let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
        let server = tokio::spawn(async move {
            axum::serve(listener, mcp_router(sessions)).await.unwrap();
        });
        let url = scoped_mcp_url(port, &project_id);
        std::fs::create_dir(&project_dir).unwrap();
        std::fs::create_dir(project_dir.join(".lian-jing")).unwrap();
        std::fs::create_dir(project_dir.join(".codex")).unwrap();
        std::fs::write(
            project_dir.join(".lian-jing/context.json"),
            serde_json::to_vec_pretty(&json!({"project_id": project_id, "schema_version": 1}))
                .unwrap(),
        )
        .unwrap();
        std::fs::write(
            project_dir.join(".codex/config.toml"),
            format!("[mcp_servers.lian-jing]\nurl = {url:?}\n"),
        )
        .unwrap();

        println!("R11_PILOT_PROJECT_DIR={}", project_dir.display());
        println!("R11_PILOT_SCOPED_URL={url}");
        println!("R11_PILOT_PROJECT_ID={project_id}");
        std::io::stdout().flush().unwrap();

        remote.set_nonblocking(true).unwrap();
        let mock = std::thread::spawn(move || {
            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(300);
            let (mut stream, _) = loop {
                match remote.accept() {
                    Ok(accepted) => break accepted,
                    Err(error)
                        if error.kind() == std::io::ErrorKind::WouldBlock
                            && std::time::Instant::now() < deadline =>
                    {
                        std::thread::sleep(std::time::Duration::from_millis(20));
                    }
                    Err(error) => panic!("5 分钟内未收到客户端的模拟 PR 请求: {error}"),
                }
            };
            stream
                .set_read_timeout(Some(std::time::Duration::from_secs(5)))
                .unwrap();
            let mut reader = BufReader::new(stream.try_clone().unwrap());
            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            assert!(line.starts_with("POST /api/v1/repos/owner/repo/pulls "));
            let mut authorized = false;
            let mut content_length = 0;
            loop {
                line.clear();
                reader.read_line(&mut line).unwrap();
                if line == "\r\n" || line.is_empty() {
                    break;
                }
                if line.eq_ignore_ascii_case(&format!("authorization: token {token}\r\n")) {
                    authorized = true;
                }
                if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
                    content_length = value.trim().parse().unwrap();
                }
            }
            assert!(authorized, "模拟 Gitea 未收到受保护存储中的合成凭证");
            let mut body = vec![0; content_length];
            reader.read_exact(&mut body).unwrap();
            let request: Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(request["head"], "synthetic-head");
            let response = r#"{"number":43,"html_url":"https://example.invalid/pr/43","title":"synthetic pilot"}"#;
            write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap();
        });
        mock.join().unwrap();
        // 让 MCP handler 把模拟 Gitea 的响应完整交付给外部客户端后再停服。
        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        server.abort();
        drop(guard);
        assert!(crate::credential_store::CredentialStore::get(
            &crate::credential_store::WindowsCredentialStore,
            &reference,
        )
        .unwrap()
        .is_none());
        assert!(!project_dir.exists());
    }

    /// V7 隔离真实客户端试点：仅内存数据库、随机合成凭证与临时 scoped MCP，认证走真实 helper。
    /// 阶段靠临时项目 `.pilot/` 标记推进：客户端登记后宿主代人填写合成账户（写 filled）；
    /// 见 `revoke` 标记后代人撤销（写 revoked）；见 `finish` 标记后检查改动边界并清理。
    #[cfg(windows)]
    #[tokio::test(flavor = "multi_thread")]
    #[ignore = "V7: connect disposable Codex sessions to the printed scoped MCP URL"]
    async fn isolated_dev_variables_client_pilot() {
        use crate::credential_store::WindowsCredentialStore;
        use std::io::Write;
        use std::time::{Duration, Instant};

        struct PilotGuard {
            project_id: String,
            project_dir: std::path::PathBuf,
        }
        impl Drop for PilotGuard {
            fn drop(&mut self) {
                if let Ok(conn) = db::pool().get() {
                    let _ = crate::commands::dev_variables::purge_project(
                        &conn,
                        &WindowsCredentialStore,
                        &self.project_id,
                    );
                }
                let _ = crate::credential_store::delete_mcp_auth_token_for_project(
                    &WindowsCredentialStore,
                    &self.project_id,
                );
                let _ = std::fs::remove_dir_all(&self.project_dir);
            }
        }

        async fn wait_until<T>(
            what: &str,
            limit: Duration,
            mut probe: impl FnMut() -> Option<T>,
        ) -> T {
            let deadline = Instant::now() + limit;
            loop {
                if let Some(value) = probe() {
                    return value;
                }
                assert!(Instant::now() < deadline, "等待超时：{what}");
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
        }

        db::init_test_pool();
        let project_id = format!("synthetic-dev-variables-pilot-{}", uuid::Uuid::new_v4());
        let profile_id = format!("{project_id}-profile");
        let project_dir =
            std::env::temp_dir().join(format!("lianjing-v7-{}", uuid::Uuid::new_v4()));
        let pilot_dir = project_dir.join(".pilot");
        let guard = PilotGuard {
            project_id: project_id.clone(),
            project_dir: project_dir.clone(),
        };
        {
            let conn = db::pool().get().unwrap();
            conn.execute(
                "INSERT INTO project_profiles (id, name) VALUES (?1, 'Synthetic Dev Variables Pilot')",
                [&profile_id],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO project_workspaces (id, profile_id) VALUES (?1, ?2)",
                rusqlite::params![project_id, profile_id],
            )
            .unwrap();
        }

        let token = crate::credential_store::get_or_create_mcp_auth_token(
            &WindowsCredentialStore,
            &project_id,
        )
        .unwrap();
        let digest = crate::credential_store::mcp_auth_digest(&project_id).unwrap();
        // 测试二进制位于 target/debug/deps/，helper 即同目录上一级的应用 debug 构建。
        let helper_exe = std::env::current_exe()
            .unwrap()
            .parent()
            .and_then(std::path::Path::parent)
            .unwrap()
            .join("dev-manager-tauri.exe");
        assert!(
            helper_exe.is_file(),
            "先 cargo build 生成 helper：{}",
            helper_exe.display()
        );
        let helper = format!(
            "\"{}\" --lianjing-mcp-header {digest}",
            helper_exe.display()
        );

        let listener = tokio::net::TcpListener::bind(mcp_bind_addr(0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
        let server = tokio::spawn(async move {
            axum::serve(listener, mcp_router(sessions)).await.unwrap();
        });
        let url = scoped_mcp_url(port, &project_id);
        std::fs::create_dir_all(project_dir.join(".lian-jing")).unwrap();
        std::fs::create_dir_all(project_dir.join(".codex")).unwrap();
        std::fs::create_dir_all(&pilot_dir).unwrap();
        std::fs::write(
            project_dir.join(".lian-jing/context.json"),
            serde_json::to_vec_pretty(&json!({"project_id": project_id, "schema_version": 1}))
                .unwrap(),
        )
        .unwrap();
        std::fs::write(
            project_dir.join(".codex/config.toml"),
            format!(
                "[mcp_servers.lian-jing]\nurl = {}\nhttp_headers_helper = {}\n",
                serde_json::to_string(&url).unwrap(),
                serde_json::to_string(&helper).unwrap()
            ),
        )
        .unwrap();

        let username = format!("pilot-user-{}", &uuid::Uuid::new_v4().simple().to_string()[..8]);
        let password = format!("pilot-pass-{}", uuid::Uuid::new_v4().simple());
        println!("V7_PILOT_PROJECT_DIR={}", project_dir.display());
        println!("V7_PILOT_SCOPED_URL={url}");
        println!("V7_PILOT_PROJECT_ID={project_id}");
        println!("V7_PILOT_HELPER={helper}");
        println!("V7_PILOT_EXPECTED_USERNAME={username}");
        std::io::stdout().flush().unwrap();

        // 阶段 1：等客户端登记测试账户需求，再代人填写合成值。
        let name = wait_until("客户端登记测试账户需求", Duration::from_secs(1800), || {
            let conn = db::pool().get().ok()?;
            conn.query_row(
                "SELECT name FROM project_dev_variables
                 WHERE project_id=?1 AND kind='test_account' ORDER BY name LIMIT 1",
                [&project_id],
                |row| row.get::<_, String>(0),
            )
            .ok()
        })
        .await;
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::save_test_account(
                &conn,
                &WindowsCredentialStore,
                &project_id,
                &name,
                &username,
                &password,
            )
            .unwrap();
        }
        std::fs::write(pilot_dir.join("filled"), &name).unwrap();
        println!("V7_PILOT_FILLED={name}");
        std::io::stdout().flush().unwrap();

        // 阶段 2：客户端跨会话读取后，由操作者放置 revoke 标记，宿主代人撤销。
        wait_until("撤销标记 .pilot/revoke", Duration::from_secs(1800), || {
            pilot_dir.join("revoke").exists().then_some(())
        })
        .await;
        {
            let conn = db::pool().get().unwrap();
            crate::commands::dev_variables::revoke(&conn, &WindowsCredentialStore, &project_id, &name)
                .unwrap();
        }
        std::fs::write(pilot_dir.join("revoked"), &name).unwrap();
        println!("V7_PILOT_REVOKED={name}");
        std::io::stdout().flush().unwrap();

        // 阶段 3：撤销后再读由客户端验证失败；finish 标记后检查改动边界。
        wait_until("结束标记 .pilot/finish", Duration::from_secs(1800), || {
            pilot_dir.join("finish").exists().then_some(())
        })
        .await;
        server.abort();
        {
            let conn = db::pool().get().unwrap();
            let statuses = crate::commands::dev_variables::list(
                &conn,
                &WindowsCredentialStore,
                &project_id,
            )
            .unwrap();
            assert!(statuses.iter().all(|status| !status.configured));
        }
        let mut pending = vec![project_dir.clone()];
        while let Some(dir) = pending.pop() {
            for entry in std::fs::read_dir(&dir).unwrap() {
                let path = entry.unwrap().path();
                if path.is_dir() {
                    pending.push(path);
                    continue;
                }
                let file_name = path.file_name().unwrap().to_string_lossy();
                assert!(!file_name.starts_with(".env"), "项目目录出现 {file_name}");
                let content = std::fs::read_to_string(&path).unwrap_or_default();
                assert!(!content.contains(&token), "{} 含 MCP token", path.display());
                assert!(!content.contains(&password), "{} 含测试密码", path.display());
            }
        }
        drop(guard);
        assert!(crate::credential_store::read_mcp_auth_token_for_digest(
            &WindowsCredentialStore,
            &digest
        )
        .is_err());
        assert!(!project_dir.exists());
    }

    #[tokio::test]
    async fn readiness_probe_does_not_satisfy_client_reload_evidence() {
        let project_id = "reload-evidence-fixture";
        let mut readiness = rpc("initialize");
        readiness.params = Some(json!({
            "clientInfo": { "name": "lian-jing-readiness", "version": "test" }
        }));
        process_rpc(&readiness, Some(project_id)).await.unwrap();
        assert!(!has_initialized_project_client(project_id));

        let mut codex = rpc("initialize");
        codex.params = Some(json!({
            "clientInfo": { "name": "codex", "version": "test" }
        }));
        process_rpc(&codex, Some(project_id)).await.unwrap();
        assert!(has_initialized_project_client(project_id));
    }

    #[test]
    fn loopback_urls_share_one_ipv4_base() {
        let port = 27190;
        let session_id = "test-session";

        assert_eq!(mcp_bind_addr(port).to_string(), "127.0.0.1:27190");
        assert_eq!(mcp_health_url(port), "http://127.0.0.1:27190/health");
        assert_eq!(
            scoped_mcp_url(port, "project one"),
            "http://127.0.0.1:27190/mcp/projects/project%20one"
        );
        assert_eq!(
            sse_messages_url(port, session_id),
            "http://127.0.0.1:27190/mcp/messages?session_id=test-session"
        );
    }
}

/// POST /mcp/messages?session_id=xxx
/// 立即返回 202，后台异步处理工具调用并经 SSE channel 推回响应
async fn handle_sse_message(
    State(sessions): State<Sessions>,
    Query(query): Query<SessionQuery>,
    headers: HeaderMap,
    Json(req): Json<RpcRequest>,
) -> Response {
    if authorize_sensitive_request(
        &req,
        None,
        &headers,
        &crate::credential_store::WindowsCredentialStore,
    )
    .is_err()
    {
        return StatusCode::FORBIDDEN.into_response();
    }

    if req.id.is_none() {
        return StatusCode::ACCEPTED.into_response();
    }

    let session_id = query.session_id.clone();
    tokio::spawn(async move {
        let id = req.id.clone();
        let response_json = match process_rpc(&req, None).await {
            Ok(result) => json!({ "jsonrpc": "2.0", "id": id, "result": result }),
            Err(msg) => json!({
                "jsonrpc": "2.0", "id": id,
                "error": { "code": -32601, "message": msg }
            }),
        };
        let mut guard = sessions.lock().await;
        if let Some(entry) = guard.get(&session_id) {
            if entry.project_scope.is_some() {
                return;
            }
            if entry
                .sender
                .send(("message".to_string(), response_json.to_string()))
                .await
                .is_err()
            {
                guard.remove(&session_id);
            }
        }
    });

    StatusCode::ACCEPTED.into_response()
}

async fn handle_scoped_sse_message(
    Path(project_id): Path<String>,
    State(sessions): State<Sessions>,
    Query(query): Query<SessionQuery>,
    headers: HeaderMap,
    Json(req): Json<RpcRequest>,
) -> Response {
    if authorize_sensitive_request(
        &req,
        Some(&project_id),
        &headers,
        &crate::credential_store::WindowsCredentialStore,
    )
    .is_err()
    {
        return StatusCode::FORBIDDEN.into_response();
    }
    if req.id.is_none() {
        return StatusCode::ACCEPTED.into_response();
    }
    let session_id = query.session_id.clone();
    let scope_matches = sessions
        .lock()
        .await
        .get(&session_id)
        .and_then(|entry| entry.project_scope.as_deref())
        == Some(project_id.as_str());
    if !scope_matches {
        return StatusCode::FORBIDDEN.into_response();
    }
    tokio::spawn(async move {
        let id = req.id.clone();
        let response_json = match process_rpc(&req, Some(&project_id)).await {
            Ok(result) => json!({ "jsonrpc": "2.0", "id": id, "result": result }),
            Err(msg) => json!({
                "jsonrpc": "2.0", "id": id,
                "error": { "code": -32601, "message": msg }
            }),
        };
        let mut guard = sessions.lock().await;
        if let Some(entry) = guard.get(&session_id) {
            if entry
                .sender
                .send(("message".to_string(), response_json.to_string()))
                .await
                .is_err()
            {
                guard.remove(&session_id);
            }
        }
    });
    StatusCode::ACCEPTED.into_response()
}