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

1099 lines
46 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

use axum::{
body::Bytes,
extract::{Path, Query, State},
http::{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, convert::Infallible, net::SocketAddr, sync::Arc};
use tokio::sync::{mpsc, Mutex};
use tokio_stream::{wrappers::ReceiverStream, StreamExt as _};
use crate::{db, db::log_event};
/// SSE 会话表session_id → 发送端((event_name, data)
type Sessions = Arc<Mutex<HashMap<String, mpsc::Sender<(String, String)>>>>;
/// 从 settings 表读取 MCP 端口,默认 27190
pub fn read_port() -> u16 {
db::pool()
.get()
.ok()
.and_then(|c| {
c.query_row(
"SELECT value FROM settings WHERE key = 'mcp_port'",
[],
|r| r.get::<_, String>(0),
)
.ok()
})
.and_then(|v| v.parse().ok())
.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 = Router::new()
.route("/health", get(health))
// Streamable HTTPPOST— 保持向后兼容Python 直调仍可用
.route("/mcp/group/:group_id", post(handle_mcp).get(handle_sse_connect))
// SSE 消息端点:客户端通过此 URL 发送 JSON-RPC响应经 SSE 推回
.route("/mcp/group/:group_id/messages", post(handle_sse_message))
// Gitea webhook 观测面:接收 push 事件写入 commit_metrics飞轮数据源
.route("/webhook/gitea/:project_id", post(handle_gitea_webhook))
.with_state(sessions);
let addr = SocketAddr::from(([0, 0, 0, 0], port));
match tokio::net::TcpListener::bind(addr).await {
Ok(listener) => {
log_event("mcp_server", "info", &format!("MCP Server 已启动,端口 {port}HTTP + SSE"), None);
if let Err(e) = axum::serve(listener, app).await {
log_event("mcp_server", "error", &format!("Server 异常退出: {e}"), None);
}
}
Err(e) => log_event("mcp_server", "error", &format!("绑定端口 {port} 失败: {e}"), None),
}
}
async fn health() -> &'static str {
"炼境 MCP Server running"
}
// ── Gitea Webhook 观测面 ──────────────────────────────────────────────────────
/// push payload 中的一条 commit
#[derive(Debug, PartialEq)]
struct PushCommit {
sha: String,
message: String,
timestamp: Option<String>,
}
/// HMAC-SHA256 验签。Gitea 实际发送的签名头是 `X-Gitea-Signature`payload 的
/// HMAC-SHA256 hexR10以实际抓包行为为准非架构文档假设的 X-Gitea-Signature-256
fn verify_hmac_sha256(secret: &str, body: &[u8], sig_hex: &str) -> bool {
use hmac::{Hmac, Mac};
use sha2::Sha256;
let mut mac = match Hmac::<Sha256>::new_from_slice(secret.as_bytes()) {
Ok(m) => m,
Err(_) => return false,
};
mac.update(body);
let expected: String = mac
.finalize()
.into_bytes()
.iter()
.map(|b| format!("{b:02x}"))
.collect();
// 大小写不敏感比较Gitea 输出小写 hex
expected.eq_ignore_ascii_case(sig_hex.trim())
}
/// 解析 Gitea push payload 的 commits 数组
fn parse_push_payload(body: &[u8]) -> Result<Vec<PushCommit>, String> {
let json: Value = serde_json::from_slice(body).map_err(|e| format!("payload 非法 JSON: {e}"))?;
let commits = json["commits"].as_array().ok_or("payload 缺少 commits 数组")?;
Ok(commits
.iter()
.filter_map(|c| {
let sha = c["id"].as_str()?.to_string();
let message = c["message"].as_str().unwrap_or("").to_string();
let timestamp = c["timestamp"].as_str().map(|s| s.to_string());
Some(PushCommit { sha, message, timestamp })
})
.collect())
}
/// 从请求头提取签名:优先 X-Gitea-Signature兼容 X-Hub-Signature-256sha256= 前缀)
fn extract_signature(headers: &HeaderMap) -> Option<String> {
if let Some(v) = headers.get("x-gitea-signature").and_then(|v| v.to_str().ok()) {
return Some(v.to_string());
}
headers
.get("x-hub-signature-256")
.and_then(|v| v.to_str().ok())
.map(|v| v.trim_start_matches("sha256=").to_string())
}
async fn handle_gitea_webhook(
Path(project_id): Path<String>,
headers: HeaderMap,
body: Bytes,
) -> Response {
// 1. 查绑定与 secret
let secret: Option<String> = db::pool().get().ok().and_then(|c| {
c.query_row(
"SELECT webhook_secret FROM gitea_repos WHERE project_id = ?1",
[&project_id],
|r| r.get(0),
)
.ok()
});
let secret = match secret {
Some(s) => s,
None => {
return (StatusCode::NOT_FOUND, Json(json!({"error": "项目未绑定 Gitea 仓库或未设置 webhook"})))
.into_response()
}
};
// 2. 验签
let sig = extract_signature(&headers);
let verified = sig.as_deref().map(|s| verify_hmac_sha256(&secret, &body, s)).unwrap_or(false);
if !verified {
log_event(
"gitea_webhook",
"warn",
&format!("项目 {project_id} webhook 验签失败(签名{}", if sig.is_some() { "不匹配" } else { "缺失" }),
None,
);
return (StatusCode::UNAUTHORIZED, Json(json!({"error": "签名验证失败"}))).into_response();
}
// 3. 按事件类型分发
let event = headers
.get("x-gitea-event")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
match event {
"push" => handle_push(&project_id, &body).await,
"pull_request" => handle_pull_request(&project_id, &body).await,
other => Json(json!({"ignored": other})).into_response(),
}
}
async fn handle_push(project_id: &str, body: &Bytes) -> Response {
let commits = match parse_push_payload(body) {
Ok(c) => c,
Err(e) => return (StatusCode::BAD_REQUEST, Json(json!({"error": e}))).into_response(),
};
let conn = match db::pool().get() {
Ok(c) => c,
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": e.to_string()}))).into_response(),
};
let mut received = 0;
for c in &commits {
if crate::commands::commit_metrics::record_commit(
&conn, project_id, &c.sha, &c.message, c.timestamp.as_deref(),
).is_ok() { received += 1; }
}
log_event(
"gitea_webhook", "info",
&format!("项目 {project_id} push 事件已接收,写入 {received}/{} 条 commit", commits.len()),
None,
);
Json(json!({"received": received})).into_response()
}
async fn handle_pull_request(project_id: &str, body: &Bytes) -> Response {
let payload: serde_json::Value = match serde_json::from_slice(body) {
Ok(v) => v,
Err(e) => return (StatusCode::BAD_REQUEST, Json(json!({"error": e.to_string()}))).into_response(),
};
let raw_action = payload["action"].as_str().unwrap_or("");
let pr = &payload["pull_request"];
let pr_number = pr["number"].as_i64().unwrap_or(0) as i64;
// 只追踪有价值的生命周期节点
let action = match raw_action {
"opened" | "reopened" => "opened",
"closed" => {
if pr["merged"].as_bool().unwrap_or(false) { "merged" } else { "closed" }
}
_ => return Json(json!({"ignored": raw_action})).into_response(),
};
let title = pr["title"].as_str().unwrap_or("").to_string();
let head_branch = pr["head"]["label"].as_str().unwrap_or("").to_string();
let base_branch = pr["base"]["label"].as_str().unwrap_or("").to_string();
let author = pr["user"]["login"].as_str().unwrap_or("").to_string();
let id = uuid::Uuid::new_v4().to_string();
let conn = match db::pool().get() {
Ok(c) => c,
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": e.to_string()}))).into_response(),
};
let result = conn.execute(
"INSERT INTO pr_events (id, project_id, pr_number, title, action, head_branch, base_branch, author)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
rusqlite::params![id, project_id, pr_number, title, action, head_branch, base_branch, author],
);
match result {
Ok(_) => {
log_event(
"gitea_webhook", "info",
&format!("项目 {project_id} PR #{pr_number} {action}"),
None,
);
Json(json!({"recorded": action, "pr_number": pr_number})).into_response()
}
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": e.to_string()}))).into_response(),
}
}
// ── 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,
}
// ── 共享 RPC 处理逻辑 ─────────────────────────────────────────────────────────
async fn process_rpc(group_id: &str, req: &RpcRequest) -> Result<Value, String> {
match req.method.as_str() {
"initialize" => Ok(json!({
"protocolVersion": "2024-11-05",
"capabilities": { "tools": {} },
"serverInfo": {
"name": "炼境",
"version": env!("CARGO_PKG_VERSION")
}
})),
"tools/list" => Ok(tools_list_result()),
"tools/call" => Ok(tools_call(group_id, req.params.as_ref()).await),
_ => Err(format!("Method not found: {}", req.method)),
}
}
// ── Streamable HTTPPOST— 向后兼容Python 直调走此路径 ─────────────────────
async fn handle_mcp(Path(group_id): Path<String>, Json(req): Json<RpcRequest>) -> Response {
// MCP 通知(无 id→ 202 Accepted
if req.id.is_none() {
return StatusCode::ACCEPTED.into_response();
}
let id = req.id.clone();
match process_rpc(&group_id, &req).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/group/:group_id
/// Claude Code 发起 SSE 握手,建立长连接,首条事件告知后续 POST 的 URL。
async fn handle_sse_connect(
State(sessions): State<Sessions>,
Path(group_id): Path<String>,
) -> 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);
// 第一条事件:告知客户端后续 POST 的 endpoint URL
let port = read_port();
let endpoint_uri = format!(
"http://127.0.0.1:{}/mcp/group/{}/messages?session_id={}",
port, group_id, session_id
);
let _ = tx.send(("endpoint".to_string(), endpoint_uri)).await;
sessions.lock().await.insert(session_id, tx);
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())
}
/// POST /mcp/group/:group_id/messages?session_id=xxx
/// 客户端通过此 URL 发送 JSON-RPC 请求,响应经 SSE channel 推回。
async fn handle_sse_message(
State(sessions): State<Sessions>,
Path(group_id): Path<String>,
Query(query): Query<SessionQuery>,
Json(req): Json<RpcRequest>,
) -> Response {
if req.id.is_none() {
return StatusCode::ACCEPTED.into_response();
}
let id = req.id.clone();
let response_json = match process_rpc(&group_id, &req).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 sessions_guard = sessions.lock().await;
if let Some(tx) = sessions_guard.get(&query.session_id) {
if tx
.send(("message".to_string(), response_json.to_string()))
.await
.is_err()
{
// 连接已关闭,惰性清理
sessions_guard.remove(&query.session_id);
}
}
StatusCode::ACCEPTED.into_response()
}
// ── 工具定义 ──────────────────────────────────────────────────────────────────
fn tools_list_result() -> Value {
json!({
"tools": [
{
"name": "get_diagnostics",
"description": "查询炼境运行时诊断信息MCP Server 启动状态、各项目 MCP 配置注入结果、最近错误日志。排查功能是否正常时首先调用此工具。",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "list_group_projects",
"description": "列出产品组内所有成员项目,包含名称、路径、平台、技术栈、角色等信息",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "get_project_blueprint",
"description": "读取产品组内某个项目的蓝图文件(.blueprint/manifest.yaml了解该项目的模块与任务状态",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
},
{
"name": "get_project_file",
"description": "读取产品组内某个项目的任意文件内容,如 openapi.yaml、README.md、src/types.ts 等",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID"
},
"relative_path": {
"type": "string",
"description": "相对于项目根目录的文件路径,如 openapi.yaml 或 src/api/routes.ts"
}
},
"required": ["project_id", "relative_path"]
}
},
{
"name": "get_project_mentor_context",
"description": "获取产品组内某个项目的完整导师上下文——包含健康状态、蓝图进度摘要、近期错误日志,以及格式化好的 Markdown 报告(可直接用于质量分析和成长建议)",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
},
{
"name": "append_project_note",
"description": "将 Claude 的分析结论或复盘记录写入项目笔记,供未来会话和用户在炼境中查阅。完成 M/L 复杂度任务后必须调用,写入结构化复盘笔记。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
},
"content": {
"type": "string",
"description": "笔记内容。完成 M/L 任务后使用以下结构化格式:\n【复盘】<任务标题>\n任务类型: 后端命令|前端组件|数据层|MCP工具|架构调整\n复杂度: S|M|L\n实际轮数: N轮预估M轮\n是否返工: 是/否\n返工原因: <若返工填写,否则省略>\n有效策略: <本次奏效的做法>\n遗留风险: <若有填写,否则省略>\n\n其他场景(分析结论、建议、洞察)可自由格式。"
}
},
"required": ["project_id", "content"]
}
},
{
"name": "get_blueprint_status",
"description": "检测项目蓝图治理状态CONVENTIONS 规则版本是否最新、蓝图内容是否可能落后于代码synced/rules_outdated/content_stale/none。等价于炼境 UI 项目卡片上的蓝图状态徽章。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
},
{
"name": "generate_blueprint_prompt",
"description": "生成蓝图初始化/同步提示词(含 CONVENTIONS 规范与 manifest 严格格式约束)。等价于炼境 UI 治理面板的「初始化蓝图/更新蓝图描绘」按钮。拿到提示词后在该项目目录下执行它(或交给该项目的 AI 会话),完成后可用 get_blueprint_status / get_project_blueprint 验证结果。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
},
"mode": {
"type": "string",
"enum": ["init", "sync"],
"description": "init=项目尚无蓝图时从零初始化sync=已有蓝图,对照代码现状同步更新"
}
},
"required": ["project_id", "mode"]
}
},
{
"name": "sync_blueprint_rules",
"description": "将最新 CONVENTIONS 规则同步到项目(写入 .blueprint/CONVENTIONS.md + 更新 AGENTS.md/CLAUDE.md 的蓝图管理区块)。等价于炼境 UI 治理面板的「同步规则」按钮。仅当 get_blueprint_status 返回 rules_outdated 时需要调用。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
},
{
"name": "create_pull_request",
"description": "通过炼境存储的 Gitea 凭证为项目创建合并请求PR。炼境内部调用 Gitea API无需 Claude 自行提取 token。/ship 交付流程在推送分支后应调用此工具完成 PR 创建。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
},
"head": {
"type": "string",
"description": "源分支名,如 dev"
},
"base": {
"type": "string",
"description": "目标分支名,如 main"
},
"title": {
"type": "string",
"description": "PR 标题"
},
"body": {
"type": "string",
"description": "PR 描述Markdown可为空字符串"
}
},
"required": ["project_id", "head", "base", "title"]
}
},
{
"name": "get_project_snapshots",
"description": "获取项目的历史健康快照(每日一条),包含 git 提交间隔、蓝图完成率、阻塞任务数、错误情况。用于分析趋势、识别高价值改进变量。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
},
"days": {
"type": "integer",
"description": "查询最近 N 天的快照,默认 30"
}
},
"required": ["project_id"]
}
},
{
"name": "poll_now",
"description": "立即触发一次 Gitea 轮询:拉取所有绑定仓库的最新 commit 和 PR 状态写入飞轮。等价于炼境 UI「立即拉取」按钮。适用于公网 Gitea + 内网炼境webhook 不可达)场景。",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "setup_webhook",
"description": "为项目重新注册 Gitea webhook订阅 push + pull_request 事件)。等价于炼境 UI「重设」按钮。导入新项目后、或需要更新事件订阅时调用。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
},
{
"name": "inject_mcp",
"description": "向项目的 .claude/settings.json 注入炼境 MCP 配置,同时写入 bypassPermissions 和 liangjing.json。等价于炼境 UI「注入 MCP」按钮。导入新项目后或炼境重启后调用。",
"inputSchema": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "项目 workspace ID来自 list_group_projects 的返回结果)"
}
},
"required": ["project_id"]
}
}
]
})
}
// ── 工具调用 ──────────────────────────────────────────────────────────────────
async fn tools_call(group_id: &str, params: Option<&Value>) -> Value {
let tool_name = params
.and_then(|p| p.get("name"))
.and_then(|n| n.as_str())
.unwrap_or("")
.to_string();
let args = params
.and_then(|p| p.get("arguments"))
.cloned()
.unwrap_or(json!({}));
let gid = group_id.to_string();
let outcome = tokio::task::spawn_blocking(move || match tool_name.as_str() {
"get_diagnostics" => get_diagnostics(),
"list_group_projects" => list_group_projects(&gid),
"get_project_blueprint" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
get_project_blueprint(&gid, pid)
}
"get_blueprint_status" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
resolve_project_root(&gid, pid).and_then(|root| {
crate::commands::blueprint::get_blueprint_status(root.to_string_lossy().to_string())
.and_then(|s| serde_json::to_string_pretty(&s).map_err(|e| e.to_string()))
})
}
"generate_blueprint_prompt" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("sync");
if mode != "init" && mode != "sync" {
Err(format!("mode 只允许 init 或 sync收到: {mode}"))
} else {
resolve_project_root(&gid, pid).and_then(|root| {
crate::commands::blueprint::generate_blueprint_prompt(
root.to_string_lossy().to_string(),
mode.to_string(),
)
})
}
}
"sync_blueprint_rules" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
resolve_project_root(&gid, pid).and_then(|root| {
crate::commands::blueprint::sync_blueprint_rules(root.to_string_lossy().to_string())
.and_then(|r| serde_json::to_string_pretty(&r).map_err(|e| e.to_string()))
})
}
"get_project_file" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let rel = args
.get("relative_path")
.and_then(|v| v.as_str())
.unwrap_or("");
get_project_file(&gid, pid, rel)
}
"get_project_mentor_context" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
get_project_mentor_context(&gid, pid)
}
"append_project_note" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let content = args.get("content").and_then(|v| v.as_str()).unwrap_or("");
append_project_note(&gid, pid, content)
}
"get_project_snapshots" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let days = args.get("days").and_then(|v| v.as_u64()).unwrap_or(30) as u32;
get_project_snapshots(&gid, pid, days)
}
"create_pull_request" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let head = args.get("head").and_then(|v| v.as_str()).unwrap_or("");
let base = args.get("base").and_then(|v| v.as_str()).unwrap_or("main");
let title = args.get("title").and_then(|v| v.as_str()).unwrap_or("");
let body = args.get("body").and_then(|v| v.as_str()).unwrap_or("");
crate::commands::gitea::gitea_api_create_pull_request(pid, head, base, title, body)
}
"poll_now" => {
crate::commands::gitea::poll_all_repos()
.map(|(polled, ingested)| format!("轮询完成:{polled} 个仓库,新 commit {ingested}PR 事件已同步"))
}
"setup_webhook" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let port = read_port();
let callback_base = format!("http://127.0.0.1:{}", port);
crate::commands::gitea::gitea_setup_webhook(pid.to_string(), callback_base)
.map(|repo| format!(
"webhook 注册成功:{}/{} hook_id={}",
repo.gitea_owner, repo.gitea_repo_name,
repo.webhook_id.map(|id| id.to_string()).unwrap_or_default()
))
}
"inject_mcp" => {
let pid = args.get("project_id").and_then(|v| v.as_str()).unwrap_or("");
let conn = db::pool().get().map_err(|e| e.to_string())?;
let group_id: String = conn.query_row(
"SELECT group_id FROM project_group_map WHERE project_id = ?1 LIMIT 1",
rusqlite::params![pid],
|r| r.get(0),
).map_err(|_| format!("项目 {} 未加入任何产品组", pid))?;
drop(conn);
crate::mcp_inject::inject_project_mcp(pid, &group_id)
.map(|_| format!("MCP 注入成功project_id={} group_id={}", pid, group_id))
}
_ => Err(format!("未知工具: {}", tool_name)),
})
.await;
match outcome {
Ok(Ok(text)) => json!({ "content": [{ "type": "text", "text": text }] }),
Ok(Err(e)) => json!({ "content": [{ "type": "text", "text": e }], "isError": true }),
Err(e) => json!({ "content": [{ "type": "text", "text": e.to_string() }], "isError": true }),
}
}
// ── 工具实现(同步,由 spawn_blocking 包裹)──────────────────────────────────
fn get_diagnostics() -> Result<String, String> {
let conn = db::pool().get().map_err(|e| e.to_string())?;
let port = read_port();
// 服务启动时间(首条 mcp_server info 日志)
let start_time: Option<String> = conn.query_row(
"SELECT ts FROM runtime_logs WHERE category = 'mcp_server' AND level = 'info' ORDER BY id ASC LIMIT 1",
[],
|r| r.get(0),
).ok();
let mut out = String::new();
out.push_str("## 炼境 MCP 诊断报告\n\n");
out.push_str(&format!("- **服务端口**: {}\n", port));
out.push_str(&format!("- **启动时间**: {}\n\n", start_time.as_deref().unwrap_or("未知(可能在本次日志之前启动)")));
// MCP 注入状态:每个 project_id 取最后一条
out.push_str("### MCP 注入状态(每个项目最后一次结果)\n\n");
let mut stmt = conn.prepare(
"SELECT context, level, message, ts
FROM runtime_logs
WHERE category = 'mcp_inject'
AND id IN (
SELECT MAX(id) FROM runtime_logs
WHERE category = 'mcp_inject'
GROUP BY json_extract(context, '$.project_id')
)
ORDER BY ts DESC
LIMIT 50",
).map_err(|e| e.to_string())?;
let inject_rows: Vec<(Option<String>, String, String, String)> = stmt
.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))
.map_err(|e| e.to_string())?
.filter_map(|r| r.ok())
.collect();
if inject_rows.is_empty() {
out.push_str("_暂无注入记录成员加入产品组时会产生_\n\n");
} else {
for (ctx, level, msg, ts) in &inject_rows {
let project_id = ctx.as_deref()
.and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
.and_then(|v| v.get("project_id").and_then(|x| x.as_str()).map(|s| s.to_string()))
.unwrap_or_else(|| "unknown".to_string());
let icon = if level == "info" { "" } else { "⚠️" };
out.push_str(&format!("- {} `{}` — {} _({})\n", icon, project_id, msg, ts));
}
out.push('\n');
}
// 最近 20 条 error/warn
out.push_str("### 最近错误日志(最多 20 条)\n\n");
let mut stmt2 = conn.prepare(
"SELECT ts, category, level, message FROM runtime_logs
WHERE level IN ('error', 'warn')
ORDER BY id DESC LIMIT 20",
).map_err(|e| e.to_string())?;
let error_rows: Vec<(String, String, String, String)> = stmt2
.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))
.map_err(|e| e.to_string())?
.filter_map(|r| r.ok())
.collect();
if error_rows.is_empty() {
out.push_str("_无错误记录_ ✅\n");
} else {
for (ts, cat, level, msg) in &error_rows {
out.push_str(&format!("- `[{}]` **{}** `{}` — {}\n", ts, level, cat, msg));
}
}
// ── 系统健康(复用 health check 逻辑)────────────────────────────────────
out.push_str("\n### 系统健康\n\n");
let report = crate::commands::settings::collect_health_report();
out.push_str(&format!("检查时间:{}\n\n", report.checked_at));
out.push_str("| 检查项 | 状态 | 详情 |\n|--------|------|------|\n");
for item in &report.items {
let icon = match item.status.as_str() {
"ok" => "",
"warn" => "⚠️",
"error" => "",
_ => "",
};
out.push_str(&format!(
"| {} | {} {} | {} |\n",
item.name, icon, item.status,
item.detail.as_deref().unwrap_or("-")
));
}
// ── 质量指标 ──────────────────────────────────────────────────────────────
out.push_str("\n### 质量指标\n\n");
let q = &report.quality;
out.push_str(&format!("- **休眠项目**>30天未 push{}\n", q.dormant_projects.len()));
for p in &q.dormant_projects {
out.push_str(&format!(" - {} {} 天)\n", p.name, p.days_since_push));
}
out.push_str(&format!("- **孤立项目**(无分组 & 无 repo{}\n", q.orphaned_projects.len()));
for p in &q.orphaned_projects {
out.push_str(&format!(" - {}\n", p.name));
}
out.push_str(&format!(
"- **MCP 注入异常产品组**(近 7 天):{}\n",
q.groups_with_inject_warn.len()
));
for gid in &q.groups_with_inject_warn {
out.push_str(&format!(" - `{}`\n", gid));
}
out.push_str(&format!("- **无本地路径项目**{}\n", q.projects_without_path));
Ok(out)
}
fn list_group_projects(group_id: &str) -> Result<String, String> {
let conn = db::pool().get().map_err(|e| e.to_string())?;
let mut stmt = conn
.prepare(
"SELECT pw.id, pp.name, pw.platform, pw.win_path, pw.wsl_path,
pp.tech_stack, pgm.role
FROM project_group_map pgm
JOIN project_workspaces pw ON pw.id = pgm.project_id
JOIN project_profiles pp ON pp.id = pw.profile_id
WHERE pgm.group_id = ?1",
)
.map_err(|e| e.to_string())?;
let rows: Vec<String> = stmt
.query_map([group_id], |row| {
let id: String = row.get(0)?;
let name: String = row.get(1)?;
let platform: Option<String> = row.get(2)?;
let win_path: Option<String> = row.get(3)?;
let wsl_path: Option<String> = row.get(4)?;
let tech_stack: Option<String> = row.get(5)?;
let role: Option<String> = row.get(6)?;
Ok(format!(
"- **{}** (id: `{}`)\n 平台: {} | 技术栈: {}{}\n Windows 路径: {}\n WSL 路径: {}",
name,
id,
platform.as_deref().unwrap_or("unknown"),
tech_stack.as_deref().unwrap_or("未配置"),
role.map(|r| format!(" | 角色: {}", r)).unwrap_or_default(),
win_path.as_deref().unwrap_or("未配置"),
wsl_path.as_deref().unwrap_or("未配置"),
))
})
.map_err(|e| e.to_string())?
.filter_map(|r| r.ok())
.collect();
if rows.is_empty() {
return Ok("该产品组暂无成员项目。".to_string());
}
Ok(format!(
"产品组成员(共 {} 个项目):\n\n{}",
rows.len(),
rows.join("\n\n")
))
}
fn get_project_blueprint(group_id: &str, project_id: &str) -> Result<String, String> {
let root = resolve_project_root(group_id, project_id)?;
let path = root.join(".blueprint").join("manifest.yaml");
std::fs::read_to_string(&path)
.map_err(|e| format!("读取蓝图失败({}: {}", path.display(), e))
}
fn get_project_file(group_id: &str, project_id: &str, relative_path: &str) -> Result<String, String> {
if relative_path.contains("..") {
return Err("路径不允许包含 \"..\"".to_string());
}
let root = resolve_project_root(group_id, project_id)?;
let path = root.join(relative_path);
std::fs::read_to_string(&path)
.map_err(|e| format!("读取文件失败({}: {}", path.display(), e))
}
/// 根据 group_id + project_id 解析项目根目录路径
/// WSL 项目自动转换为 Windows UNC 路径(\\wsl.localhost\<distro>\...
fn resolve_project_root(group_id: &str, project_id: &str) -> Result<std::path::PathBuf, String> {
let conn = db::pool().get().map_err(|e| e.to_string())?;
let row: rusqlite::Result<(Option<String>, Option<String>, Option<String>)> = conn.query_row(
"SELECT pw.win_path, pw.wsl_path, pw.platform
FROM project_group_map pgm
JOIN project_workspaces pw ON pw.id = pgm.project_id
WHERE pgm.group_id = ?1 AND pgm.project_id = ?2",
[group_id, project_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
);
let (win_path, wsl_path, platform) = row.map_err(|_| {
format!("项目 {} 不属于产品组 {},或项目不存在", project_id, group_id)
})?;
if platform.as_deref() == Some("wsl") {
let wsl = wsl_path.ok_or_else(|| "WSL 项目未配置 wsl_path".to_string())?;
let distro: String = conn
.query_row(
"SELECT value FROM settings WHERE key = 'wsl_distro'",
[],
|r| r.get(0),
)
.unwrap_or_else(|_| "Ubuntu".to_string());
// /home/user/project → \\wsl.localhost\Ubuntu\home\user\project
let inner = wsl.trim_start_matches('/').replace('/', "\\");
Ok(std::path::PathBuf::from(format!(
"\\\\wsl.localhost\\{}\\{}",
distro, inner
)))
} else {
let win = win_path.ok_or_else(|| "Windows 项目未配置 win_path".to_string())?;
Ok(std::path::PathBuf::from(win))
}
}
fn get_project_mentor_context(group_id: &str, project_id: &str) -> Result<String, String> {
// 验证项目属于该产品组(与其他 MCP 工具一致)
let conn = db::pool().get().map_err(|e| e.to_string())?;
let exists: bool = conn
.query_row(
"SELECT COUNT(*) FROM project_group_map WHERE group_id = ?1 AND project_id = ?2",
[group_id, project_id],
|r| r.get::<_, i64>(0),
)
.map(|n| n > 0)
.unwrap_or(false);
if !exists {
return Err(format!("项目 {} 不属于产品组 {},或项目不存在", project_id, group_id));
}
let ctx = crate::commands::mentor::get_mentor_context(project_id.to_string())
.map_err(|e| format!("获取导师上下文失败: {e}"))?;
Ok(ctx.mentor_markdown)
}
fn append_project_note(group_id: &str, project_id: &str, content: &str) -> Result<String, String> {
if content.trim().is_empty() {
return Err("笔记内容不能为空".to_string());
}
let conn = db::pool().get().map_err(|e| e.to_string())?;
// 校验 project_id 属于该 group
let exists: bool = conn
.query_row(
"SELECT COUNT(*) FROM project_group_map WHERE group_id = ?1 AND project_id = ?2",
[group_id, project_id],
|r| r.get::<_, i64>(0),
)
.map(|n| n > 0)
.unwrap_or(false);
if !exists {
return Err(format!("项目 {} 不属于产品组 {},或项目不存在", project_id, group_id));
}
conn.execute(
"INSERT INTO project_notes (project_id, source, content) VALUES (?1, 'mcp', ?2)",
rusqlite::params![project_id, content],
)
.map_err(|e| e.to_string())?;
Ok("笔记已保存 ✅".to_string())
}
fn get_project_snapshots(group_id: &str, project_id: &str, days: u32) -> Result<String, String> {
let conn = db::pool().get().map_err(|e| e.to_string())?;
// 校验项目属于该产品组
let exists: bool = conn
.query_row(
"SELECT COUNT(*) FROM project_group_map WHERE group_id = ?1 AND project_id = ?2",
[group_id, project_id],
|r| r.get::<_, i64>(0),
)
.map(|n| n > 0)
.unwrap_or(false);
if !exists {
return Err(format!("项目 {} 不属于产品组 {},或项目不存在", project_id, group_id));
}
let mut stmt = conn
.prepare(
"SELECT snapshot_date, days_since_commit, blueprint_done_rate,
blueprint_in_progress, blueprint_blocked, has_errors
FROM project_snapshots
WHERE project_id = ?1
AND snapshot_date >= date('now', 'localtime', ?2)
ORDER BY snapshot_date ASC",
)
.map_err(|e| e.to_string())?;
let offset = format!("-{} days", days);
let rows: Vec<(String, Option<i64>, Option<f64>, i64, i64, i64)> = stmt
.query_map(rusqlite::params![project_id, offset], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?))
})
.map_err(|e| e.to_string())?
.filter_map(|r| r.ok())
.collect();
if rows.is_empty() {
return Ok(format!(
"项目 `{}` 暂无历史快照(快照在每次应用启动时写入,最多保留 {} 天)。",
project_id, days
));
}
let mut md = format!(
"# 项目历史快照:`{}`\n\n最近 {} 天,共 {} 条记录\n\n",
project_id,
days,
rows.len()
);
md.push_str("| 日期 | 距上次提交(天) | 蓝图完成率 | 进行中模块 | 阻塞任务 | 有错误 |\n");
md.push_str("|------|--------------|-----------|-----------|---------|------|\n");
for (date, days_commit, done_rate, in_prog, blocked, has_err) in &rows {
let commit_col = days_commit
.map(|d| d.to_string())
.unwrap_or_else(|| "".to_string());
let rate_col = done_rate
.map(|r| format!("{:.0}%", r * 100.0))
.unwrap_or_else(|| "".to_string());
let err_col = if *has_err == 1 { "⚠️" } else { "" };
md.push_str(&format!(
"| {} | {} | {} | {} | {} | {} |\n",
date, commit_col, rate_col, in_prog, blocked, err_col
));
}
md.push_str("\n---\n\n请根据以上趋势数据,分析哪些变量与项目停滞或健康度下降最相关,并给出改进建议。\n");
Ok(md)
}
// ── Gitea Webhook 单元测试 ────────────────────────────────────────────────────
#[cfg(test)]
mod webhook_tests {
use super::*;
/// 用相同算法生成签名,验证正反两个方向
fn sign(secret: &str, body: &[u8]) -> String {
use hmac::{Hmac, Mac};
use sha2::Sha256;
let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).unwrap();
mac.update(body);
mac.finalize().into_bytes().iter().map(|b| format!("{b:02x}")).collect()
}
#[test]
fn hmac_verify_accepts_valid_signature() {
let body = br#"{"commits":[]}"#;
let sig = sign("my-secret", body);
assert!(verify_hmac_sha256("my-secret", body, &sig));
// 大写 hex 也应通过
assert!(verify_hmac_sha256("my-secret", body, &sig.to_uppercase()));
}
#[test]
fn hmac_verify_rejects_wrong_secret_or_tampered_body() {
let body = br#"{"commits":[]}"#;
let sig = sign("my-secret", body);
assert!(!verify_hmac_sha256("other-secret", body, &sig));
assert!(!verify_hmac_sha256("my-secret", br#"{"commits":[1]}"#, &sig));
assert!(!verify_hmac_sha256("my-secret", body, "deadbeef"));
}
#[test]
fn parse_push_payload_extracts_commits() {
let body = br#"{
"ref": "refs/heads/master",
"commits": [
{"id": "abc123", "message": "feat(core): add thing", "timestamp": "2026-07-02T10:00:00+08:00"},
{"id": "def456", "message": "fix(ui): repair", "timestamp": null}
]
}"#;
let commits = parse_push_payload(body).unwrap();
assert_eq!(commits.len(), 2);
assert_eq!(commits[0].sha, "abc123");
assert_eq!(commits[0].message, "feat(core): add thing");
assert_eq!(commits[0].timestamp.as_deref(), Some("2026-07-02T10:00:00+08:00"));
assert_eq!(commits[1].timestamp, None);
}
#[test]
fn parse_push_payload_rejects_invalid() {
assert!(parse_push_payload(b"not json").is_err());
assert!(parse_push_payload(br#"{"no_commits": true}"#).is_err());
// commits 为空数组是合法的(如仅 tag push
assert_eq!(parse_push_payload(br#"{"commits":[]}"#).unwrap().len(), 0);
}
}