diff --git a/crates/df-mcp/src/lib.rs b/crates/df-mcp/src/lib.rs index c3a233a..1b2630c 100644 --- a/crates/df-mcp/src/lib.rs +++ b/crates/df-mcp/src/lib.rs @@ -17,4 +17,4 @@ pub mod server; pub mod server_http; pub mod tools; -pub use server::run_server; +pub use server::{run_server, ServerConfig}; diff --git a/crates/df-mcp/src/server.rs b/crates/df-mcp/src/server.rs index bf4fac0..7403604 100644 --- a/crates/df-mcp/src/server.rs +++ b/crates/df-mcp/src/server.rs @@ -8,9 +8,11 @@ //! //! 高风险(High)工具:tools/list 不暴露(从清单剔除),tools/call 即便绕过也由 handler 兜底拒绝。 //! read-only:tools/list 仅留 Low,tools/call Medium/High 一律拒绝。 +//! 生命周期:空闲超时(默认 60s 无请求)自动退出,防客户端强杀后进程残留;写操作(risk != Low)可选回调(stdio 预留接线点)。 use std::path::Path; use std::sync::Arc; +use std::time::Duration; use df_storage::db::Database; use serde_json::{json, Value}; @@ -27,31 +29,62 @@ pub(crate) const PROTOCOL_VERSION: &str = "2025-06-18"; pub(crate) const SERVER_NAME: &str = "devflow-mcp"; pub(crate) const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION"); -/// 启动 MCP server。 +/// MCP server 运行配置。 +/// +/// 聚合 stdio 形态的运行参数,避免 `run_server` 参数膨胀。 +/// [`Default`] 即常用生产配置(可写 + 空闲超时 60s + 无写回调)。 +pub struct ServerConfig { + /// 只读模式:true 则仅暴露 list/get 工具,Medium/High 写操作一律拒绝。 + pub read_only: bool, + /// 空闲超时:连续 N 时长无 stdin 输入则自动退出进程(防客户端强杀后孤儿残留)。 + /// `None` = 永不因空闲退出(常驻)。默认 60s。 + pub idle_timeout: Option, + /// 写操作回调:成功执行(未被 read-only/High 拒绝)的写工具(risk != Low)触发。 + /// 桌面内嵌形态经 server_http → on_tool_call → emit df-data-changed 刷新 GUI; + /// stdio 独立进程暂无 AppHandle,传 None 预留接线点,后续可注入。 + pub on_write_call: Option>, +} + +impl Default for ServerConfig { + fn default() -> Self { + Self { + read_only: false, + idle_timeout: Some(Duration::from_secs(60)), + on_write_call: None, + } + } +} + +/// 启动 MCP server(stdin/stdout 接实际进程句柄)。 /// /// - `db_path`:SQLite 数据库文件路径(应用同库,开 WAL 模式) -/// - `read_only`:true 则仅暴露 list/get 工具 +/// - `config`:运行配置(只读/空闲超时/写操作回调),见 [`ServerConfig`] /// /// 复用 [`Database::open`](df_storage::Database::open)(已含 `PRAGMA journal_mode=WAL`), /// 故 SQLite WAL 状态:随 df-storage 一起已启用,无需额外处理。 -pub async fn run_server(db_path: &Path, read_only: bool) -> anyhow::Result<()> { +pub async fn run_server(db_path: &Path, config: ServerConfig) -> anyhow::Result<()> { let db = Arc::new(Database::open(db_path).await?); let ctx = Ctx::new(db); let stdin = tokio::io::stdin(); let stdout = tokio::io::stdout(); - main_loop(stdin, stdout, &ctx, read_only).await + main_loop(stdin, stdout, &ctx, &config).await } /// 可单测的主循环(参数化 stdin/stdout)。 /// /// 协议正确性:每行一个 JSON-RPC 消息,Response 单行写回(末尾 \n)。 /// Notification(id=None)不回响应。 +/// +/// 生命周期: +/// - 空闲超时:仅在**等待下一个请求**(`read_line`)时计时;正在处理的请求不受影响, +/// 超时即 break 优雅退出(防客户端强杀后进程残留)。 +/// - 写操作回调:成功执行的写工具(risk != Low)在响应写回后触发,供外部(GUI)感知数据变更。 pub async fn main_loop( stdin: R, stdout: W, ctx: &Ctx, - read_only: bool, + config: &ServerConfig, ) -> anyhow::Result<()> where R: tokio::io::AsyncRead + Unpin, @@ -63,7 +96,17 @@ where loop { line.clear(); - let n = reader.read_line(&mut line).await?; + // 空闲超时:只包裹「等待下一请求」,不包裹 dispatch/写响应,处理期间绝不误杀。 + let n = match config.idle_timeout { + Some(dur) => match tokio::time::timeout(dur, reader.read_line(&mut line)).await { + Ok(n) => n, + Err(_elapsed) => { + tracing::info!(target: "df_mcp", idle_secs = dur.as_secs(), "空闲超时无新请求,自动退出"); + break; + } + }?, + None => reader.read_line(&mut line).await?, + }; if n == 0 { // EOF(stdin 关闭),优雅退出 break; @@ -101,8 +144,16 @@ where continue; } - let resp = dispatch(ctx, read_only, req.id.clone(), method).await; + // 写回调需在 dispatch 后判定,先预取工具名(仅 tools/call 需要,避免 clone 整包请求) + let tool_name = match &method { + McpMethod::ToolsCall { name, .. } => Some(name.clone()), + _ => None, + }; + let resp = dispatch(ctx, config.read_only, req.id.clone(), method).await; write_response(&mut writer, &resp).await?; + if let Some(name) = tool_name { + fire_write_hook(config, &name); + } } Ok(()) @@ -153,19 +204,15 @@ pub(crate) async fn dispatch(ctx: &Ctx, read_only: bool, id: Option, meth .unwrap_or(Value::Null), ); }; - // read-only 模式:Medium/High 全拒 - if read_only && spec.risk != RiskLevel::Low { - let r = CallToolResult::error(format!( - "只读模式拒绝执行 {name}(风险等级 {:?})", - spec.risk - )); - return Response::ok(id, serde_json::to_value(r).unwrap_or(Value::Null)); - } - // 非 read-only:High 兜底拒绝(handler 内也会拒,双保险) - if spec.risk == RiskLevel::High { - let r = CallToolResult::error(format!( - "High 风险操作 {name} 默认拒绝,请在 DevFlow 应用内执行。" - )); + // 执行前防御:read-only 拒 Medium/High;High 兜底拒(handler 内也会拒,双保险)。 + // 判定收口到 should_execute,与 main_loop 写回调共用同一事实源,避免两份逻辑漂移。 + if !should_execute(read_only, spec.risk) { + let msg = if read_only && spec.risk != RiskLevel::Low { + format!("只读模式拒绝执行 {name}(风险等级 {:?})", spec.risk) + } else { + format!("High 风险操作 {name} 默认拒绝,请在 DevFlow 应用内执行。") + }; + let r = CallToolResult::error(msg); return Response::ok(id, serde_json::to_value(r).unwrap_or(Value::Null)); } // Low / Medium:执行 @@ -181,6 +228,23 @@ pub(crate) async fn dispatch(ctx: &Ctx, read_only: bool, id: Option, meth } } +/// 工具是否会被执行(dispatch 与 main_loop 写回调共用的判定)。 +/// +/// read-only 下仅 Low 可执行;非 read-only 下 High 仍兜底拒绝。两条件都过 → 可执行。 +fn should_execute(read_only: bool, risk: RiskLevel) -> bool { + !(read_only && risk != RiskLevel::Low) && risk != RiskLevel::High +} + +/// 触发写操作回调(若有)。仅当工具为写操作(risk != Low)且未被 read-only/High 拒绝时触发, +/// 与 dispatch 的执行判定一致。回调仅作通知(如 GUI 刷新),不承载返回结果。 +fn fire_write_hook(config: &ServerConfig, name: &str) { + let Some(cb) = &config.on_write_call else { return }; + let Some(spec) = tools::find(name) else { return }; + if spec.risk != RiskLevel::Low && should_execute(config.read_only, spec.risk) { + cb(name); + } +} + /// 工具可见性:read-only 仅 Low,否则 Low + Medium(High 永不可见) pub(crate) fn visible(read_only: bool, risk: RiskLevel) -> bool { if read_only { @@ -213,6 +277,8 @@ async fn write_response( mod tests { use super::*; use crate::tools::RiskLevel; + use std::sync::Mutex; + use tokio::io::AsyncWriteExt; /// 构造内存 DB + Ctx async fn test_ctx() -> Ctx { @@ -358,4 +424,122 @@ mod tests { assert!(!visible(true, RiskLevel::Medium)); assert!(!visible(true, RiskLevel::High)); } + + // ── should_execute 判定(dispatch 与写回调共用)──────────────────── + + #[test] + fn should_execute_predicate() { + assert!(should_execute(false, RiskLevel::Low)); + assert!(should_execute(false, RiskLevel::Medium)); + assert!(!should_execute(false, RiskLevel::High)); + assert!(should_execute(true, RiskLevel::Low)); + assert!(!should_execute(true, RiskLevel::Medium)); + assert!(!should_execute(true, RiskLevel::High)); + } + + // ── 空闲超时 / 写操作回调(main_loop 集成)──────────────────────── + + /// 构造测试配置:只读开关 + 默认空闲超时 + 默认无回调(字段可覆盖) + async fn test_config(read_only: bool) -> ServerConfig { + ServerConfig { + read_only, + idle_timeout: Some(Duration::from_secs(60)), + on_write_call: None, + } + } + + #[tokio::test] + async fn main_loop_exits_on_idle_timeout() { + let ctx = test_ctx().await; + // 用 duplex 造一个「开着但永不写数据」的 stdin:read_line 会一直挂起, + // 空闲超时(100ms)触发后应正常 break 退出,而非阻塞或 panic。 + let (_tx, rx) = tokio::io::duplex(1024); + let config = ServerConfig { + idle_timeout: Some(Duration::from_millis(100)), + ..test_config(false).await + }; + let outer = tokio::time::timeout( + Duration::from_secs(2), + main_loop(rx, tokio::io::sink(), &ctx, &config), + ) + .await; + let inner = outer.expect("main_loop 应在空闲超时后返回,而非一直阻塞"); + assert!(inner.is_ok(), "空闲超时退出应为 Ok,实际: {inner:?}"); + } + + #[tokio::test] + async fn main_loop_fires_write_callback_for_write_tool() { + let ctx = test_ctx().await; + let calls: Arc>> = Arc::new(Mutex::new(Vec::new())); + let calls_cb = calls.clone(); + let config = ServerConfig { + on_write_call: Some(Arc::new(move |name| { + calls_cb.lock().unwrap().push(name.to_string()); + })), + ..test_config(false).await + }; + let (mut tx, rx) = tokio::io::duplex(1024); + let line = r#"{"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"create_project","arguments":{"name":"McpProj","description":"via mcp"}}}"#; + tx.write_all(format!("{line}\n").as_bytes()).await.unwrap(); + drop(tx); // 关 stdin → 处理后 EOF,正常退出 + main_loop(rx, tokio::io::sink(), &ctx, &config) + .await + .unwrap(); + assert!( + calls.lock().unwrap().contains(&"create_project".to_string()), + "写工具 create_project 应触发写回调,实际: {:?}", + *calls.lock().unwrap() + ); + } + + #[tokio::test] + async fn main_loop_no_write_callback_for_read_tool() { + let ctx = test_ctx().await; + let calls: Arc>> = Arc::new(Mutex::new(Vec::new())); + let calls_cb = calls.clone(); + let config = ServerConfig { + on_write_call: Some(Arc::new(move |name| { + calls_cb.lock().unwrap().push(name.to_string()); + })), + ..test_config(false).await + }; + let (mut tx, rx) = tokio::io::duplex(1024); + let line = r#"{"jsonrpc":"2.0","id":8,"method":"tools/call","params":{"name":"list_projects","arguments":{}}}"#; + tx.write_all(format!("{line}\n").as_bytes()).await.unwrap(); + drop(tx); + main_loop(rx, tokio::io::sink(), &ctx, &config) + .await + .unwrap(); + assert!( + calls.lock().unwrap().is_empty(), + "只读工具 list_projects 不应触发写回调,实际: {:?}", + *calls.lock().unwrap() + ); + } + + #[tokio::test] + async fn main_loop_no_write_callback_when_read_only_denies_write() { + let ctx = test_ctx().await; + let calls: Arc>> = Arc::new(Mutex::new(Vec::new())); + let calls_cb = calls.clone(); + let config = ServerConfig { + read_only: true, + on_write_call: Some(Arc::new(move |name| { + calls_cb.lock().unwrap().push(name.to_string()); + })), + ..test_config(true).await + }; + let (mut tx, rx) = tokio::io::duplex(1024); + let line = r#"{"jsonrpc":"2.0","id":9,"method":"tools/call","params":{"name":"create_project","arguments":{"name":"X","description":"d"}}}"#; + tx.write_all(format!("{line}\n").as_bytes()).await.unwrap(); + drop(tx); + main_loop(rx, tokio::io::sink(), &ctx, &config) + .await + .unwrap(); + assert!( + calls.lock().unwrap().is_empty(), + "read-only 下 Medium 写被拒,不应触发写回调,实际: {:?}", + *calls.lock().unwrap() + ); + } } diff --git a/crates/df-mcp/src/tools.rs b/crates/df-mcp/src/tools.rs index 38180a4..5cbd715 100644 --- a/crates/df-mcp/src/tools.rs +++ b/crates/df-mcp/src/tools.rs @@ -15,7 +15,7 @@ // name→id 解析:src-tauri 有机制层解析(audit/mod.rs auto_resolve),MCP 面暂不同步。 use std::sync::{Arc, OnceLock}; -use df_storage::crud::{IdeaRepo, ProjectRepo, TaskRepo}; +use df_storage::crud::{IdeaQuery, IdeaRepo, ProjectQuery, ProjectRepo, TaskQuery, TaskRepo}; use df_storage::db::Database; use df_storage::models::{IdeaRecord, ProjectRecord, TaskRecord}; use df_types::types::{IdeaStatus, ProjectStatus, TaskStatus, new_id}; @@ -102,25 +102,25 @@ pub fn all_tools() -> &'static Vec<&'static ToolSpec> { use RiskLevel::*; vec![ // ─── 项目 ─── - spec("list_projects", "列出所有未删除项目", object_schema(json!({}), &[]), Low, list_projects), + spec("list_projects", "列出所有未删除项目(分页:offset/limit,默认 limit=50 上限 100)", object_schema(json!({"offset": int_field("偏移量(可空,默认 0)"), "limit": int_field("返回上限(可空,默认 50,上限 100)")}), &[]), Low, list_projects), spec("get_project", "按 ID 获取项目", object_schema(json!({"id": str_field("项目 ID")}), &["id"]), Low, get_project), spec("create_project", "创建项目(Medium 风险,默认允许+审计日志)", object_schema(json!({"name": str_field("项目名"), "description": str_field("描述"), "status": opt_str_field("状态(默认 planning)")}), &["name", "description"]), Medium, create_project), - spec("update_project", "更新项目(部分更新:仅传需要改的字段,未传字段保留原值)", object_schema(json!({"id": str_field("项目 ID"), "name": opt_str_field("项目名(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)"), "status": opt_str_field("状态(可空=保留原值)")}), &["id"]), Medium, update_project), + spec("update_project", "更新项目(部分更新:仅传需要改的字段,未传字段保留原值)", object_schema(json!({"id": str_field("项目 ID"), "name": opt_str_field("项目名(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)"), "status": opt_str_field("状态(可空=保留原值)"), "expected_updated_at": int_field("乐观锁版本(可空):上次读取到的 updated_at 毫秒时间戳,不一致则拒绝写入")}), &["id"]), Medium, update_project), spec("delete_project", "软删项目(进回收站,可恢复)——High 风险,默认拒绝,请在 DevFlow 应用内执行", object_schema(json!({"id": str_field("项目 ID")}), &["id"]), High, delete_project), spec("bind_directory", "为项目绑定本地代码目录(会做路径冲突检测,Medium 风险+审计日志)", object_schema(json!({"id": str_field("项目 ID"), "path": str_field("本地目录绝对路径")}), &["id", "path"]), Medium, bind_directory), // ─── 任务 ─── - spec("list_tasks", "列出所有未删除任务(可按 project_id/status 过滤)", object_schema(json!({"project_id": opt_str_field("按项目过滤(可空)"), "status": opt_str_field("按状态过滤(todo/in_progress/in_review/testing/blocked/done/cancelled,可空)")}), &[]), Low, list_tasks), + spec("list_tasks", "列出所有未删除任务(可按 project_id/status 过滤;分页 offset/limit,默认 limit=50 上限 100)", object_schema(json!({"project_id": opt_str_field("按项目过滤(可空)"), "status": opt_str_field("按状态过滤(todo/in_progress/in_review/testing/blocked/done/cancelled,可空)"), "offset": int_field("偏移量(可空,默认 0)"), "limit": int_field("返回上限(可空,默认 50,上限 100)")}), &[]), Low, list_tasks), spec("create_task", "创建任务(Medium 风险,默认允许+审计日志;可选 parent_id 父任务 ID,限 1 级嵌套,父任务自身不能是子任务)", object_schema(json!({"project_id": str_field("项目 ID"), "title": str_field("标题"), "description": str_field("描述"), "priority": int_field("优先级(可空,默认 0)"), "parent_id": opt_str_field("父任务 ID(可空)")}), &["project_id", "title", "description"]), Medium, create_task), - spec("update_task", "更新任务(部分更新:仅传需要改的字段,未传字段保留原值;状态须走 advance_task)", object_schema(json!({"id": str_field("任务 ID"), "project_id": opt_str_field("项目 ID(可空=保留原值)"), "title": opt_str_field("标题(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)")}), &["id"]), Medium, update_task), + spec("update_task", "更新任务(部分更新:仅传需要改的字段,未传字段保留原值;状态须走 advance_task)", object_schema(json!({"id": str_field("任务 ID"), "project_id": opt_str_field("项目 ID(可空=保留原值)"), "title": opt_str_field("标题(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)"), "expected_updated_at": int_field("乐观锁版本(可空):上次读取到的 updated_at 毫秒时间戳,不一致则拒绝写入")}), &["id"]), Medium, update_task), spec("advance_task", "推进任务状态(传目标 status,内部读当前态+状态机校验,Medium 风险+审计日志)", object_schema(json!({"id": str_field("任务 ID"), "to": str_field("目标 status(todo/in_progress/in_review/testing/blocked/done/cancelled)")}), &["id", "to"]), Medium, advance_task), spec("delete_task", "软删任务(进回收站)——High 风险,默认拒绝,请在 DevFlow 应用内执行", object_schema(json!({"id": str_field("任务 ID")}), &["id"]), High, delete_task), // ─── 灵感 ─── - spec("list_ideas", "列出所有想法/灵感", object_schema(json!({}), &[]), Low, list_ideas), + spec("list_ideas", "列出所有想法/灵感(分页:offset/limit,默认 limit=50 上限 100)", object_schema(json!({"offset": int_field("偏移量(可空,默认 0)"), "limit": int_field("返回上限(可空,默认 50,上限 100)")}), &[]), Low, list_ideas), spec("create_idea", "创建想法(Medium 风险,默认允许+审计日志)", object_schema(json!({"title": str_field("标题"), "description": str_field("描述"), "priority": int_field("优先级(可空,默认 0)")}), &["title", "description"]), Medium, create_idea), - spec("update_idea", "更新想法(部分更新:仅传需要改的字段,未传字段保留原值)", object_schema(json!({"id": str_field("想法 ID"), "title": opt_str_field("标题(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)")}), &["id"]), Medium, update_idea), + spec("update_idea", "更新想法(部分更新:仅传需要改的字段,未传字段保留原值)", object_schema(json!({"id": str_field("想法 ID"), "title": opt_str_field("标题(可空=保留原值)"), "description": opt_str_field("描述(可空=保留原值)"), "expected_updated_at": int_field("乐观锁版本(可空):上次读取到的 updated_at 毫秒时间戳,不一致则拒绝写入")}), &["id"]), Medium, update_idea), spec("delete_idea", "软删想法——High 风险,默认拒绝,请在 DevFlow 应用内执行", object_schema(json!({"id": str_field("想法 ID")}), &["id"]), High, delete_idea), spec("evaluate_idea", "对想法做启发式评估(只读:只返分数不写库,基于 description/title 计算 feasibility/impact/urgency/overall)", object_schema(json!({"id": str_field("想法 ID")}), &["id"]), Low, evaluate_idea), - spec("score_idea", "评分并写库(Medium 风险+审计日志):对想法做启发式评估,把 scores 写回 DB 并返回更新后的记录", object_schema(json!({"id": str_field("想法 ID")}), &["id"]), Medium, score_idea), + spec("score_idea", "评分并写库(Medium 风险+审计日志):对想法做启发式评估,把 scores 写回 DB 并返回更新后的记录", object_schema(json!({"id": str_field("想法 ID"), "expected_updated_at": int_field("乐观锁版本(可空):上次读取到的 updated_at 毫秒时间戳,不一致则拒绝写入")}), &["id"]), Medium, score_idea), // ─── 工作流(High) ─── spec("run_workflow", "触发工作流——High 风险,默认拒绝,请在 DevFlow 应用内执行", object_schema(json!({"project_id": str_field("项目 ID"), "task_id": opt_str_field("任务 ID(可空)")}), &["project_id"]), High, run_workflow), // ─── 回收站 ─── @@ -203,6 +203,41 @@ fn arg_int_or(args: &Value, key: &str, default: i32) -> i32 { .unwrap_or(default) } +/// 解析分页参数 offset/limit(offset 默认 0;limit 默认 50,钳制上限 100)。 +/// 供 list_projects / list_tasks / list_ideas 三个列表工具统一使用。 +fn pagination(args: &Value) -> (u32, u32) { + let offset = args + .get("offset") + .and_then(|v| v.as_i64()) + .map(|i| i.max(0) as u32) + .unwrap_or(0); + let limit = args + .get("limit") + .and_then(|v| v.as_i64()) + .map(|i| i.max(1) as u32) + .unwrap_or(50); + (offset, limit.min(100)) +} + +/// 乐观锁版本校验(CAS):调用方传入 expected_updated_at(毫秒时间戳)时, +/// 与 DB 当前 updated_at 比对,不一致即拒绝写入(数据已被其他进程修改)。 +/// 未传则跳过校验(向后兼容,不破坏旧调用方)。接受数字或字符串两种传法。 +fn check_expected_updated_at(args: &Value, db_updated_at: &str) -> Result<(), CallToolResult> { + let Some(expected) = args.get("expected_updated_at").and_then(|v| { + v.as_i64() + .or_else(|| v.as_str().and_then(|s| s.trim().parse().ok())) + }) else { + return Ok(()); + }; + let current: i64 = db_updated_at.trim().parse().unwrap_or(i64::MIN); + if current != expected { + return Err(CallToolResult::error(format!( + "数据已被其他进程修改(当前 updated_at={db_updated_at}),请刷新后重试" + ))); + } + Ok(()) +} + // ============================================================ // 跨实体校验(防 B-260801-01:跨实体误操作) // ============================================================ @@ -258,12 +293,30 @@ async fn cross_entity_err(db: &Arc, id: &str, excluding: &str) -> Opti // handler 实现 — 项目 // ============================================================ -fn list_projects(ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult> { +fn list_projects(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { let db = ctx.db.clone(); + let (offset, limit) = pagination(&args); Box::pin(async move { let repo = ProjectRepo::new(&db); - match repo.list_active().await { - Ok(list) => json_ok(json!({ "projects": list, "count": list.len() })), + // 分页:取 limit+1 条探测是否有下一页(has_more),再截断到 limit。 + // 复用 list_by_query(默认 created_at DESC,与 list_active 排序一致)。 + let q = ProjectQuery { + limit: Some(limit + 1), + offset: Some(offset), + ..Default::default() + }; + match repo.list_by_query(q).await { + Ok(list) => { + let has_more = list.len() as u32 > limit; + let page: Vec<_> = list.into_iter().take(limit as usize).collect(); + json_ok(json!({ + "projects": page, + "count": page.len(), + "offset": offset, + "limit": limit, + "has_more": has_more + })) + } Err(e) => err_str(e), } }) @@ -345,6 +398,10 @@ fn update_project(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> } Err(e) => return err_str(e), }; + // 乐观锁 CAS:调用方传入 expected_updated_at 则与 DB 当前版本比对,不一致拒绝写入 + if let Err(r) = check_expected_updated_at(&args, &existing.updated_at) { + return r; + } // 部分更新:name/description/status 缺省回退 existing,避免空默认清空数据 let name = arg_str(&args, "name").unwrap_or_else(|_| existing.name.clone()); let description = arg_str(&args, "description").unwrap_or_else(|_| existing.description.clone()); @@ -431,22 +488,29 @@ fn list_tasks(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { let db = ctx.db.clone(); let project_id_filter = args.get("project_id").and_then(|v| v.as_str()).map(|s| s.to_owned()); let status_filter = args.get("status").and_then(|v| v.as_str()).map(|s| s.to_owned()); + let (offset, limit) = pagination(&args); Box::pin(async move { let repo = TaskRepo::new(&db); - let query = df_storage::crud::TaskQuery { + // 分页:取 limit+1 条探测是否有下一页(has_more),再截断到 limit。 + let query = TaskQuery { project_id: project_id_filter, status: status_filter, - priority: None, - assignee: None, - keyword: None, - queue: None, - parent_id: None, - order_by: None, - limit: None, - offset: None, + limit: Some(limit + 1), + offset: Some(offset), + ..Default::default() }; match repo.list_by_query(&query).await { - Ok(list) => json_ok(json!({ "tasks": list, "count": list.len() })), + Ok(list) => { + let has_more = list.len() as u32 > limit; + let page: Vec<_> = list.into_iter().take(limit as usize).collect(); + json_ok(json!({ + "tasks": page, + "count": page.len(), + "offset": offset, + "limit": limit, + "has_more": has_more + })) + } Err(e) => err_str(e), } }) @@ -540,6 +604,10 @@ fn update_task(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { } Err(e) => return err_str(e), }; + // 乐观锁 CAS:调用方传入 expected_updated_at 则与 DB 当前版本比对,不一致拒绝写入 + if let Err(r) = check_expected_updated_at(&args, &existing.updated_at) { + return r; + } // 部分更新:project_id/title/description 缺省回退 existing,避免空默认清空数据 let project_id = arg_str(&args, "project_id").unwrap_or_else(|_| existing.project_id.clone()); let title = arg_str(&args, "title").unwrap_or_else(|_| existing.title.clone()); @@ -615,12 +683,30 @@ fn delete_task(_ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult> { // handler 实现 — 灵感 // ============================================================ -fn list_ideas(ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult> { +fn list_ideas(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { let db = ctx.db.clone(); + let (offset, limit) = pagination(&args); Box::pin(async move { let repo = IdeaRepo::new(&db); - match repo.list_all().await { - Ok(list) => json_ok(json!({ "ideas": list, "count": list.len() })), + // 分页:取 limit+1 条探测是否有下一页(has_more),再截断到 limit。 + // 复用 list_by_query(默认 created_at DESC,与 list_all 排序一致)。 + let q = IdeaQuery { + limit: Some(limit + 1), + offset: Some(offset), + ..Default::default() + }; + match repo.list_by_query(&q).await { + Ok(list) => { + let has_more = list.len() as u32 > limit; + let page: Vec<_> = list.into_iter().take(limit as usize).collect(); + json_ok(json!({ + "ideas": page, + "count": page.len(), + "offset": offset, + "limit": limit, + "has_more": has_more + })) + } Err(e) => err_str(e), } }) @@ -684,6 +770,10 @@ fn update_idea(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { } Err(e) => return err_str(e), }; + // 乐观锁 CAS:调用方传入 expected_updated_at 则与 DB 当前版本比对,不一致拒绝写入 + if let Err(r) = check_expected_updated_at(&args, &existing.updated_at) { + return r; + } // 部分更新:title/description 缺省回退 existing,避免空默认清空数据 let title = arg_str(&args, "title").unwrap_or_else(|_| existing.title.clone()); let description = arg_str(&args, "description").unwrap_or_else(|_| existing.description.clone()); @@ -764,6 +854,10 @@ fn score_idea(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { Ok(None) => return CallToolResult::error(format!("想法不存在: {id}")), Err(e) => return err_str(e), }; + // 乐观锁 CAS:调用方传入 expected_updated_at 则与 DB 当前版本比对,不一致拒绝写入 + if let Err(r) = check_expected_updated_at(&args, &idea.updated_at) { + return r; + } // 与 evaluate_idea 共用的纯函数评分 let scores = heuristic_scores(&idea.title, &idea.description); let now = now_millis(); @@ -1271,4 +1365,150 @@ mod tests { // 三表都没有 → None assert_eq!(detect_entity_owner(&ctx.db, "ghost", "task").await, None); } + + // ── 乐观锁 CAS(update 读-改-写竞态防护)───────────────────────── + + /// update_project 传错误的 expected_updated_at → 拒绝写入。 + #[tokio::test] + async fn update_project_cas_mismatch_rejects() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let p = ProjectRepo::new(&ctx.db).get_by_id(&pid).await.unwrap().unwrap(); + let wrong: i64 = p.updated_at.parse::().unwrap() + 1; + let r = update_project(&ctx, json!({ "id": pid, "name": "新名", "expected_updated_at": wrong })).await; + assert_eq!(r.is_error, Some(true)); + assert!( + text_of(&r).contains("数据已被其他进程修改"), + "应报版本冲突,实际: {}", + text_of(&r) + ); + } + + /// update_project 传正确的 expected_updated_at → 成功。 + #[tokio::test] + async fn update_project_cas_match_succeeds() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let p = ProjectRepo::new(&ctx.db).get_by_id(&pid).await.unwrap().unwrap(); + let expected: i64 = p.updated_at.parse().unwrap(); + let r = update_project(&ctx, json!({ "id": pid, "name": "新名", "expected_updated_at": expected })).await; + assert!(r.is_error.is_none(), "版本一致应成功: {:?}", text_of(&r)); + assert_eq!(json_of(&r)["project"]["name"], "新名"); + } + + /// update_project 不传 expected_updated_at → 跳过校验(向后兼容)。 + #[tokio::test] + async fn update_project_cas_absent_is_backward_compatible() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let r = update_project(&ctx, json!({ "id": pid, "name": "新名" })).await; + assert!(r.is_error.is_none(), "不传版本应成功: {:?}", text_of(&r)); + assert_eq!(json_of(&r)["project"]["name"], "新名"); + } + + /// score_idea 传错误的 expected_updated_at → 拒绝写入且 DB 不落库。 + #[tokio::test] + async fn score_idea_cas_mismatch_rejects() { + let ctx = test_ctx().await; + let id = seed_idea(&ctx, "核心功能重构", "需要立即重构关键模块").await; + let idea = IdeaRepo::new(&ctx.db).get_by_id(&id).await.unwrap().unwrap(); + let wrong: i64 = idea.updated_at.parse::().unwrap() + 1; + let r = score_idea(&ctx, json!({ "id": id, "expected_updated_at": wrong })).await; + assert_eq!(r.is_error, Some(true)); + assert!( + text_of(&r).contains("数据已被其他进程修改"), + "应报版本冲突,实际: {}", + text_of(&r) + ); + assert!(db_scores(&ctx, &id).await.is_none(), "CAS 拒绝时不应写库"); + } + + /// score_idea 传正确的 expected_updated_at → 成功写库。 + #[tokio::test] + async fn score_idea_cas_match_succeeds() { + let ctx = test_ctx().await; + let id = seed_idea(&ctx, "核心功能重构", "需要立即重构关键模块").await; + let idea = IdeaRepo::new(&ctx.db).get_by_id(&id).await.unwrap().unwrap(); + let expected: i64 = idea.updated_at.parse().unwrap(); + let r = score_idea(&ctx, json!({ "id": id, "expected_updated_at": expected })).await; + assert!(r.is_error.is_none(), "版本一致应成功: {:?}", text_of(&r)); + assert!(db_scores(&ctx, &id).await.is_some(), "版本一致应写库"); + } + + /// update_task 传错误的 expected_updated_at → 拒绝写入。 + #[tokio::test] + async fn update_task_cas_mismatch_rejects() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "宿主").await; + let tid = seed_task(&ctx, &pid, "原标题").await; + let t = TaskRepo::new(&ctx.db).get_by_id(&tid).await.unwrap().unwrap(); + let wrong: i64 = t.updated_at.parse::().unwrap() + 1; + let r = update_task(&ctx, json!({ "id": tid, "title": "新标题", "expected_updated_at": wrong })).await; + assert_eq!(r.is_error, Some(true)); + assert!(text_of(&r).contains("数据已被其他进程修改"), "实际: {}", text_of(&r)); + } + + // ── list 分页(offset/limit/has_more)───────────────────────────── + + /// list_ideas 分页:limit 截断 + has_more 翻页标记。 + #[tokio::test] + async fn list_ideas_pagination_has_more_and_page() { + let ctx = test_ctx().await; + for i in 0..5 { + seed_idea(&ctx, &format!("灵感{i}"), "x").await; + } + // 首页 limit=2 → 2 条 + has_more=true + let r = list_ideas(&ctx, json!({ "limit": 2, "offset": 0 })).await; + assert!(r.is_error.is_none(), "{:?}", text_of(&r)); + let v = json_of(&r); + assert_eq!(v["ideas"].as_array().unwrap().len(), 2); + assert_eq!(v["count"], 2); + assert_eq!(v["has_more"], true); + assert_eq!(v["limit"], 2); + assert_eq!(v["offset"], 0); + // 第二页 offset=2 → 又 2 条,仍有下一页(共 5 条) + let r = list_ideas(&ctx, json!({ "limit": 2, "offset": 2 })).await; + let v = json_of(&r); + assert_eq!(v["ideas"].as_array().unwrap().len(), 2); + assert_eq!(v["has_more"], true); + // 第三页 offset=4 → 1 条,has_more=false(到尾) + let r = list_ideas(&ctx, json!({ "limit": 2, "offset": 4 })).await; + let v = json_of(&r); + assert_eq!(v["ideas"].as_array().unwrap().len(), 1); + assert_eq!(v["has_more"], false); + } + + /// list_tasks 分页:limit 超上限钳到 100;不足一页无下一页。 + #[tokio::test] + async fn list_tasks_pagination_caps_limit_and_no_has_more() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "宿主").await; + for i in 0..3 { + seed_task(&ctx, &pid, &format!("任务{i}")).await; + } + let r = list_tasks(&ctx, json!({ "limit": 999 })).await; + assert!(r.is_error.is_none(), "{:?}", text_of(&r)); + let v = json_of(&r); + assert_eq!(v["limit"], 100, "limit 超上限应钳到 100"); + assert_eq!(v["tasks"].as_array().unwrap().len(), 3); + assert_eq!(v["has_more"], false, "3 条 < 100,无下一页"); + } + + /// list_projects 分页:offset 生效 + has_more 标记。 + #[tokio::test] + async fn list_projects_pagination_offset_and_has_more() { + let ctx = test_ctx().await; + for i in 0..4 { + seed_project(&ctx, &format!("项目{i}")).await; + } + let r = list_projects(&ctx, json!({ "limit": 3, "offset": 0 })).await; + assert!(r.is_error.is_none(), "{:?}", text_of(&r)); + let v = json_of(&r); + assert_eq!(v["projects"].as_array().unwrap().len(), 3); + assert_eq!(v["has_more"], true, "4 条取 3,还有下一页"); + let r = list_projects(&ctx, json!({ "limit": 3, "offset": 3 })).await; + let v = json_of(&r); + assert_eq!(v["projects"].as_array().unwrap().len(), 1); + assert_eq!(v["has_more"], false); + } } diff --git a/crates/df-storage/src/db.rs b/crates/df-storage/src/db.rs index 5ca884f..4f553c9 100644 --- a/crates/df-storage/src/db.rs +++ b/crates/df-storage/src/db.rs @@ -19,6 +19,9 @@ impl Database { /// 打开（或创建）数据库文件 pub async fn open(path: &Path) -> Result { let conn = Connection::open(path)?; + // GUI 与 MCP server 多进程写同库(WAL 下写写仍互斥),无 busy_timeout 时并发写 + // 立即报 SQLITE_BUSY。设 5s 等待,让短暂持锁的一方先完成而非直接失败。 + conn.busy_timeout(std::time::Duration::from_millis(5000))?; conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?; // 执行迁移 diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index adeb101..1a55213 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -4,10 +4,42 @@ use std::path::PathBuf; use std::process::ExitCode; +/// MCP 进程本地时区日志时间戳格式化器(与 GUI lib.rs 同款,对齐日志格式)。 +/// tracing_subscriber 默认输出 UTC,改为 chrono::Local 输出含时区偏移的本地时间。 +struct LocalTimer; +impl tracing_subscriber::fmt::time::FormatTime for LocalTimer { + fn format_time(&self, w: &mut tracing_subscriber::fmt::format::Writer<'_>) -> std::fmt::Result { + write!(w, "{}", chrono::Local::now().format("%Y-%m-%dT%H:%M:%S%.6f%:z")) + } +} + +/// 初始化 MCP 进程的文件日志:追加到 %TEMP%/devflow-trace.log(与 GUI 同文件), +/// RUST_LOG 控制级别(默认 info)。返回的 guard 需保持存活到 server 退出, +/// 否则 non_blocking worker 提前关闭会丢失日志。 +fn init_mcp_tracing() -> tracing_appender::non_blocking::WorkerGuard { + let log_path = std::env::temp_dir().join("devflow-trace.log"); + let log_file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&log_path) + .expect("创建日志文件失败"); + let (non_blocking, guard) = tracing_appender::non_blocking(log_file); + tracing_subscriber::fmt() + .with_timer(LocalTimer) + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")) + ) + .with_writer(non_blocking) + .with_ansi(false) + .init(); + guard +} + fn main() -> ExitCode { // CLI 子命令:手动解析 argv(避免引入 clap 重依赖)。 // - // `devflow mcp-server [--db ] [--read-only] [--stdio] [--print-config]` + // `devflow mcp-server [--db ] [--read-only] [--stdio] [--idle-timeout ] [--print-config]` // // 仅识别首个非程序名参数是否为 `mcp-server`;其余情况走 GUI。 // stdio 句柄由父进程(MCP 客户端)继承,无需额外配置。 @@ -22,14 +54,17 @@ fn main() -> ExitCode { /// 解析 mcp-server 子命令参数并启动 MCP server。 /// /// - `--db `:指定 SQLite 路径(默认:系统 app_data_dir/devflow[-dev].db) -/// - `--read-only`:仅暴露 list/get 工具 +/// - `--read-only`:仅暴露 list/get 工具(生产环境建议开启,拒绝全部写库) /// - `--stdio`:显式声明 stdio 形态(默认即 stdio;headless/CI/GUI 未启动场景) +/// - `--idle-timeout `:空闲 N 秒无请求自动退出(0=禁用;默认 60) /// - `--print-config`:输出 MCP 配置到 stdout 后退出(默认 HTTP 形态,加 --stdio 输出 stdio command 形态) fn run_mcp_server(args: &[String]) -> ExitCode { let mut db_path: Option = None; let mut read_only = false; let mut stdio_mode = false; let mut print_config = false; + // 空闲超时:None=未显式指定(用默认 60s);Some(0)=禁用;Some(n)=n 秒 + let mut idle_timeout_arg: Option = None; let mut i = 0; while i < args.len() { @@ -45,13 +80,28 @@ fn run_mcp_server(args: &[String]) -> ExitCode { "--read-only" => read_only = true, "--stdio" => stdio_mode = true, "--print-config" => print_config = true, + "--idle-timeout" => { + i += 1; + if i >= args.len() { + eprintln!("--idle-timeout 需要参数(整数秒,0=禁用)"); + return ExitCode::FAILURE; + } + match args[i].parse::() { + Ok(secs) => idle_timeout_arg = Some(secs), + Err(_) => { + eprintln!("--idle-timeout 参数需为整数秒: {}", args[i]); + return ExitCode::FAILURE; + } + } + } "-h" | "--help" => { - println!("devflow mcp-server [--db ] [--read-only] [--stdio] [--print-config]"); + println!("devflow mcp-server [--db ] [--read-only] [--stdio] [--idle-timeout ] [--print-config]"); println!(); - println!(" --db 指定 SQLite 数据库路径(默认:app_data_dir//devflow[-dev].db)"); - println!(" --read-only 仅暴露 list/get 工具(拒绝 create/update/delete)"); - println!(" --stdio 显式声明 stdio 形态(默认即 stdio;headless/CI/GUI 未启动场景)"); - println!(" --print-config 输出 MCP 配置(默认 HTTP,加 --stdio 输出 stdio command 形态)"); + println!(" --db 指定 SQLite 数据库路径(默认:app_data_dir//devflow[-dev].db)"); + println!(" --read-only 仅暴露 list/get 工具(拒绝全部写库,生产环境建议开启)"); + println!(" --stdio 显式声明 stdio 形态(默认即 stdio;headless/CI/GUI 未启动场景)"); + println!(" --idle-timeout 空闲 N 秒无请求自动退出(防孤儿进程残留;0=禁用;默认 60)"); + println!(" --print-config 输出 MCP 配置(默认 HTTP,加 --stdio 输出 stdio command 形态)"); return ExitCode::SUCCESS; } other => { @@ -78,21 +128,27 @@ fn run_mcp_server(args: &[String]) -> ExitCode { } // 输出 Claude Code / Cursor 兼容的 MCP 配置片段(command/args 形态,显式带 --stdio)。 - // Claude Code: claude mcp add devflow -- devflow mcp-server --stdio [--read-only] + // Claude Code: claude mcp add devflow -- devflow mcp-server --stdio [--read-only] [--idle-timeout N] // Cursor: 在 ~/.cursor/mcp.json 加 mcpServers.devflow let read_only_flag = if read_only { r#","--read-only""# } else { "" }; + // 空闲超时:用户显式指定过才写进配置片段(未指定由 server 用默认 60s,不必冗余)。 + let idle_flag = match idle_timeout_arg { + Some(secs) => format!(r#","--idle-timeout","{secs}""#), + None => String::new(), + }; // 用当前可执行文件绝对路径(command 需要能被客户端直接执行;硬编码 devflow // 在 Windows 上是 DevFlow.exe,且 PATH 未必含安装目录)。 let exe = std::env::current_exe() .map(|p| p.to_string_lossy().replace('\\', "/")) .unwrap_or_else(|_| "devflow".to_owned()); let json = format!( - r#"{{"mcpServers":{{"devflow":{{"command":"{exe}","args":["mcp-server","--stdio"{read_only_flag}]}}}}}}"# + r#"{{"mcpServers":{{"devflow":{{"command":"{exe}","args":["mcp-server","--stdio"{read_only_flag}{idle_flag}]}}}}}}"# ); println!("{json}"); eprintln!("# Claude Code 接入(stdio 形态):"); eprintln!("# claude mcp add devflow -- {exe} mcp-server --stdio"); eprintln!("# Cursor 接入:把上面 JSON 写入 ~/.cursor/mcp.json"); + eprintln!("# 安全提示:生产环境建议加 --read-only(只读),写库请在 DevFlow 应用内操作"); return ExitCode::SUCCESS; } @@ -109,6 +165,27 @@ fn run_mcp_server(args: &[String]) -> ExitCode { }, }; + // MCP 是独立进程,不经 GUI lib.rs 的 run() 路径,须单独 init tracing subscriber; + // 否则 df_mcp 内 medium_audit 的 tracing::warn! 无 subscriber 是 no-op,审计不落盘。 + // guard 绑定在此函数作用域,存活到 server 退出,保证日志 worker 持续运行。 + let _log_guard = init_mcp_tracing(); + tracing::info!(path = %db_path.display(), "[startup] MCP server 日志已初始化"); + + // 空闲超时:未显式指定用默认 60s(对齐 ServerConfig::default);显式 0 = 禁用(常驻)。 + let idle_timeout = match idle_timeout_arg { + Some(0) => None, + Some(secs) => Some(std::time::Duration::from_secs(secs)), + None => Some(std::time::Duration::from_secs(60)), + }; + let config = df_mcp::ServerConfig { + read_only, + idle_timeout, + // 写操作回调:GUI 内嵌 HTTP 形态经 src-tauri/src/mcp.rs → on_tool_call 已接 + // emit df-data-changed 刷新前端;stdio 独立进程暂无 AppHandle,暂传 None, + // 架构上预留接线点,后续注入 AppHandle 即可让 stdio 形态 MCP 写库也触发 GUI 刷新。 + on_write_call: None, + }; + // 启动 MCP server(tokio runtime) let rt = match tokio::runtime::Runtime::new() { Ok(rt) => rt, @@ -117,7 +194,7 @@ fn run_mcp_server(args: &[String]) -> ExitCode { return ExitCode::FAILURE; } }; - match rt.block_on(df_mcp::run_server(&db_path, read_only)) { + match rt.block_on(df_mcp::run_server(&db_path, config)) { Ok(()) => ExitCode::SUCCESS, Err(e) => { eprintln!("MCP server 异常退出: {e}");