diff --git a/crates/df-mcp/Cargo.toml b/crates/df-mcp/Cargo.toml index 0cef617..ec3cee2 100644 --- a/crates/df-mcp/Cargo.toml +++ b/crates/df-mcp/Cargo.toml @@ -11,6 +11,9 @@ df-types = { path = "../df-types" } # 含 is_valid_state+can_transition+同态拒绝三层校验),避免 MCP 直调底层 # advance_status_atomic 绕过状态机(防止外部客户端非法跳态 todo→done)。 df-nodes = { path = "../df-nodes" } +# df-project: bind_directory 复用技术栈探测(detect_stack)与路径规范化(normalize_path), +# 对齐 GUI tool_registry.rs::bind_dir_to_project,避免在本 crate 重复实现镜像。 +df-project = { path = "../df-project" } serde.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/crates/df-mcp/src/server.rs b/crates/df-mcp/src/server.rs index 7403604..9c42445 100644 --- a/crates/df-mcp/src/server.rs +++ b/crates/df-mcp/src/server.rs @@ -215,7 +215,19 @@ pub(crate) async fn dispatch(ctx: &Ctx, read_only: bool, id: Option, meth let r = CallToolResult::error(msg); return Response::ok(id, serde_json::to_value(r).unwrap_or(Value::Null)); } - // Low / Medium:执行 + // Low / Medium:执行前先校验 schema 必填参数(缺必填直接拒, + // 防 handler 内 arg_str_or 静默兜底把缺参当空串/默认值写入) + let missing: Vec = required_names(&spec.tool.input_schema) + .into_iter() + .filter(|name| arguments.get(name).map_or(true, |v| v.is_null())) + .collect(); + if !missing.is_empty() { + let r = CallToolResult::error(format!( + "缺少必填参数: {}", + missing.join(", ") + )); + return Response::ok(id, serde_json::to_value(r).unwrap_or(Value::Null)); + } let result = (spec.handler)(ctx, arguments).await; Response::ok(id, serde_json::to_value(result).unwrap_or(Value::Null)) } @@ -235,6 +247,30 @@ fn should_execute(read_only: bool, risk: RiskLevel) -> bool { !(read_only && risk != RiskLevel::Low) && risk != RiskLevel::High } +/// 从工具 inputSchema 提取必填参数名列表。 +/// +/// 兼容 MCP schema required 的两种形态:array 显式列出必填属性;bool true 表示全部 +/// properties 必填;false/缺失表示无必填。供 tools/call 执行前校验参数完整性, +/// 缺必填直接拒绝,杜绝 handler 内 arg_str_or 静默兜底把缺参当空串/默认值写入。 +fn required_names(schema: &Value) -> Vec { + match schema.get("required") { + Some(Value::Array(items)) => items + .iter() + .filter_map(|v| v.as_str().map(|s| s.to_owned())) + .collect(), + Some(Value::Bool(true)) => { + let mut names: Vec = schema + .get("properties") + .and_then(|p| p.as_object()) + .map(|props| props.keys().cloned().collect()) + .unwrap_or_default(); + names.sort(); + names + } + _ => Vec::new(), + } +} + /// 触发写操作回调(若有)。仅当工具为写操作(risk != Low)且未被 read-only/High 拒绝时触发, /// 与 dispatch 的执行判定一致。回调仅作通知(如 GUI 刷新),不承载返回结果。 fn fire_write_hook(config: &ServerConfig, name: &str) { @@ -415,6 +451,54 @@ mod tests { assert_eq!(list_v["projects"][0]["name"], "McpProj"); } + // ── schema required 校验(dispatch 层缺必填直接拒)──────────────── + + #[tokio::test] + async fn tools_call_missing_required_arg_returns_error() { + // create_project schema required 含 name:缺 name 应被 dispatch 层拒绝 + // (而非 handler 内 arg_str_or 静默兜底把 description 当空串) + let input = + r#"{"jsonrpc":"2.0","id":9,"method":"tools/call","params":{"name":"create_project","arguments":{"description":"缺 name"}}}"#; + let out = run_io_lines(&[input], false).await; + let v: Value = serde_json::from_str(&out[0]).unwrap(); + assert_eq!(v["result"]["isError"], true); + let text = v["result"]["content"][0]["text"].as_str().unwrap(); + assert!(text.contains("缺少必填参数: name"), "实际: {text}"); + } + + #[test] + fn required_names_from_array() { + let schema = json!({ + "type": "object", + "properties": { "id": {}, "name": {} }, + "required": ["id", "name"] + }); + assert_eq!(required_names(&schema), vec!["id".to_string(), "name".to_string()]); + } + + #[test] + fn required_names_from_bool_true_means_all_properties() { + let schema = json!({ + "type": "object", + "properties": { "id": {}, "name": {}, "desc": {} }, + "required": true + }); + // bool true = 全部 properties 必填,返回全部属性名(排序保证确定性) + assert_eq!( + required_names(&schema), + vec!["desc".to_string(), "id".to_string(), "name".to_string()] + ); + } + + #[test] + fn required_names_false_or_missing_is_empty() { + assert_eq!(required_names(&json!({ "type": "object" })), Vec::::new()); + assert_eq!( + required_names(&json!({ "type": "object", "required": false })), + Vec::::new() + ); + } + #[tokio::test] async fn visible_predicate() { assert!(visible(false, RiskLevel::Low)); diff --git a/crates/df-mcp/src/tools.rs b/crates/df-mcp/src/tools.rs index e6e9111..addc70e 100644 --- a/crates/df-mcp/src/tools.rs +++ b/crates/df-mcp/src/tools.rs @@ -104,19 +104,19 @@ pub fn all_tools() -> &'static Vec<&'static ToolSpec> { // ─── 项目 ─── 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("create_project", "创建项目(Medium 风险,默认允许+审计日志)", object_schema(json!({"name": str_field("项目名"), "description": str_field("描述"), "status": opt_str_field("状态(默认 planning)")}), &["name"]), 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("状态(可空=保留原值)"), "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 过滤;分页 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("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"]), 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("描述(可空=保留原值)"), "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", "列出所有想法/灵感(分页: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("create_idea", "创建想法(Medium 风险,默认允许+审计日志)", object_schema(json!({"title": str_field("标题"), "description": str_field("描述"), "priority": int_field("优先级(可空,默认 0)")}), &["title"]), Medium, create_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), @@ -455,26 +455,55 @@ fn bind_directory(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> medium_audit("bind_directory", &format!("{id} <- {path}")); Box::pin(async move { let repo = ProjectRepo::new(&db); + // 目录必须真实存在(对齐 GUI bind_dir_to_project:先 is_dir 校验再往下走) + if !std::path::Path::new(&path).is_dir() { + return CallToolResult::error(format!("目录不存在: {path}")); + } // 分段检测 `..`(防穿越)——纯子串 contains("..") 会误伤 my..file 这类合法名, // 改用逐段判断对齐 tool_registry.rs:validate_path 的分段检测逻辑。 let has_traversal = path.split(|c| c == '\\' || c == '/').any(|seg| seg == ".."); if has_traversal { return CallToolResult::error(format!("路径不得包含 '..' 段: {}", path)); } - let norm = normalize_path(&path); - // 路径冲突检测 + // canonicalize 解析真实绝对路径:目录已存在,符号链接在此被解析, + // 防经 symlink 逃逸到预期外目录(对齐 GUI normalize_path 的 canonicalize 优先)。 + let canon = match std::path::Path::new(&path).canonicalize() { + Ok(c) => c, + Err(e) => return CallToolResult::error(format!("路径解析失败: {path}: {e}")), + }; + // 路径冲突检测(规范化比较,防路径写法差异绕过,复用 Repo 统一实现) + let norm = df_project::scan::normalize_path(&path); if let Some(conflict) = repo.find_path_conflict(&norm, Some(&id)).await.ok().flatten() { return CallToolResult::error(format!( "路径已被项目「{}」({})绑定,请先解绑", conflict.name, conflict.id )); } - // 仅更新 path 字段(用 normalize 后的规范化路径,保留其它) + // 技术栈探测:detect_stack 内含多次同步 fs IO,spawn_blocking 防阻塞 tokio runtime + // (与 GUI bind_dir_to_project 同法;用 canonicalize 后的真实目录探测) + let stack_dir = canon; + let stack = match tokio::task::spawn_blocking(move || { + df_project::scan::detect_stack(&stack_dir) + }) + .await + { + Ok(Ok(s)) => s, + Ok(Err(e)) => return CallToolResult::error(format!("技术栈探测失败: {e}")), + Err(e) => return CallToolResult::error(format!("技术栈探测任务失败: {e}")), + }; + let stack_json = match serde_json::to_string(&stack) { + Ok(s) => s, + Err(e) => return CallToolResult::error(format!("技术栈序列化失败: {e}")), + }; + // 写回 path(规范化后的真实路径)与 stack 字段,其余字段保留 match repo.update_field(&id, "path", &norm).await { Ok(true) => {} Ok(false) => return CallToolResult::error(format!("项目不存在: {id}")), Err(e) => return err_str(e), } + if let Err(e) = repo.update_field(&id, "stack", &stack_json).await { + return err_str(e); + } let updated = repo.get_by_id(&id).await.ok().flatten(); json_ok(json!({ "id": id, "project": updated })) }) @@ -951,39 +980,6 @@ fn restore_project(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> }) } -// ============================================================ -// 路径规范化(镜像 df_project::scan::normalize_path,本 crate 不依赖 df-project) -// ============================================================ - -fn normalize_path(p: &str) -> String { - // 先尝试 canonicalize(解析符号链接 + 绝对化 + .. 折叠,仅已存在的路径有效) - if let Ok(abs) = std::path::Path::new(p).canonicalize() { - return abs.to_string_lossy().replace('\\', "/").to_lowercase(); - } - // fallback:路径尚未创建,手动做以下处理: - // ① 统一分隔符 - // ② 逐段折叠 ..(防 foo/../bar → foo/bar) - // ③ 去尾斜杠 - // ④ 小写化 - let normalized = p.replace('\\', "/"); - let mut segments: Vec<&str> = Vec::new(); - for seg in normalized.split('/') { - match seg { - "." | "" => continue, // 当前目录 / 空段(连续斜杠) - ".." if segments.is_empty() => segments.push(".."), // 根级 .. 保留(相对路径语义) - ".." => { segments.pop(); } // 上级 → 弹出上一段 - _ => segments.push(seg), - } - } - let result = if segments.is_empty() { - String::new() - } else { - segments.join("/") - }; - // 去尾斜杠 - result.trim_end_matches('/').to_lowercase() -} - // ============================================================ // 单测:evaluate_idea(只读,不写库)/ score_idea(写库)/ 风险契约 // ============================================================ @@ -1514,4 +1510,51 @@ mod tests { assert_eq!(v["projects"].as_array().unwrap().len(), 1); assert_eq!(v["has_more"], false); } + + // ── bind_directory(对齐 GUI:is_dir 校验 + canonicalize + stack 探测)── + + #[tokio::test] + async fn bind_directory_rejects_non_existent_dir() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let fake = std::env::temp_dir().join(format!("df_mcp_no_such_{}", uuid::Uuid::new_v4())); + let r = bind_directory(&ctx, json!({ "id": pid, "path": fake.to_string_lossy() })).await; + assert_eq!(r.is_error, Some(true)); + assert!(text_of(&r).contains("目录不存在"), "实际: {}", text_of(&r)); + } + + #[tokio::test] + async fn bind_directory_rejects_traversal_path() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let root = std::env::temp_dir().join(format!("df_mcp_trav_{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(root.join("sub")).unwrap(); + // 真实存在但含 `..` 段的路径:is_dir 通过后应由遍历检测拒绝 + let path = root.join("sub").join("..").join("sub"); + let r = bind_directory(&ctx, json!({ "id": pid, "path": path.to_string_lossy() })).await; + std::fs::remove_dir_all(&root).ok(); + assert_eq!(r.is_error, Some(true)); + assert!(text_of(&r).contains("'..'"), "实际: {}", text_of(&r)); + } + + #[tokio::test] + async fn bind_directory_binds_real_dir_and_writes_stack() { + let ctx = test_ctx().await; + let pid = seed_project(&ctx, "项目").await; + let dir = std::env::temp_dir().join(format!("df_mcp_bind_{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + // 放一个 Cargo.toml 让 detect_stack 识别为 rust,验证 stack 探测生效 + std::fs::write( + dir.join("Cargo.toml"), + "[package]\nname = \"x\"\nversion = \"0.1.0\"\n", + ) + .unwrap(); + let r = bind_directory(&ctx, json!({ "id": pid, "path": dir.to_string_lossy() })).await; + std::fs::remove_dir_all(&dir).ok(); + assert!(r.is_error.is_none(), "绑定失败: {:?}", text_of(&r)); + let v = json_of(&r); + assert_eq!(v["id"], pid); + let stack = v["project"]["stack"].as_str().unwrap_or_default(); + assert!(stack.contains("rust"), "应探测到 rust 技术栈,实际 stack: {stack}"); + } } diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 073126f..2a116c0 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -176,7 +176,18 @@ async fn resolve_and_inject( return String::new(); } let loc = sanitize::locality_of(provider); - let (augs, _errors) = state.resolvers.resolve_all(&refs, loc).await; + let (augs, errors) = state.resolvers.resolve_all(&refs, loc).await; + // 技能/提及注入失败透出:单条 resolve 失败不阻断整批注入(既有语义保留), + // 但失败不能静默——warn 日志让技能注入失败可见(不改前端行为)。 + if !errors.is_empty() { + tracing::warn!( + skill = ?skill, + fail_count = errors.len(), + "[ai] 技能/提及注入失败 {} 条(已跳过,不阻断注入): {:?}", + errors.len(), + errors + ); + } build_augmentation_segment(&augs, lang) } diff --git a/src-tauri/src/commands/ai/stream_recv.rs b/src-tauri/src/commands/ai/stream_recv.rs index df640ab..4937118 100644 --- a/src-tauri/src/commands/ai/stream_recv.rs +++ b/src-tauri/src/commands/ai/stream_recv.rs @@ -148,6 +148,27 @@ fn accumulate_tool_calls( } } +/// 双写 emit: Tauri 事件 + ai_event_bus 总线(跨端透传)。消费 ev。 +fn emit_ai_chat_event(app_handle: &AppHandle, ev: AiChatEvent) { + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); +} + +/// AR-8: flush 合批的 AiTextDelta(双写 emit + event_bus)。空 buffer 不 emit。 +fn flush_delta(app_handle: &AppHandle, conv_id: &str, pending_delta: &mut String) { + if pending_delta.is_empty() { + return; + } + let delta = std::mem::take(pending_delta); + emit_ai_chat_event( + app_handle, + AiChatEvent::AiTextDelta { + delta, + conversation_id: Some(conv_id.to_string()), + }, + ); +} + /// 流式接收 LLM 响应。 /// /// 三类异常处理（返回 StreamResult 显式区分出口）： @@ -184,6 +205,10 @@ pub(crate) async fn stream_llm( const FIRST_CHUNK_TIMEOUT: Duration = Duration::from_secs(10); /// 心跳间隔：静默期向前端报「LLM 仍在跑」，reset watchdog const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30); + /// AR-8: 后端 delta 合批窗口。50ms 内的多次 AiTextDelta 合并为一条再 emit, + /// 降低 IPC 事件风暴(前端 rAF 只节流渲染不节流 IPC);合并后前端 currentText + /// 逐条累加,最终一致。 + const DELTA_FLUSH_INTERVAL: Duration = Duration::from_millis(50); // ================================================================ // BUG-2026-07-17 根治: 整个 LLM 流式请求运行在独立 OS 线程的 @@ -264,6 +289,10 @@ pub(crate) async fn stream_llm( let mut final_usage: Option = None; // BUG-260617-12: DeepSeek thinking 模式推理内容累积（多轮需回传） let mut reasoning_content_acc: Option = None; + // AR-8: per-conv delta 合批累加器。pending_delta 累积未 flush 的文本, + // next_flush_at 记录下次 flush 的绝对时刻(首个入 buffer 时置 now+50ms)。 + let mut pending_delta = String::new(); + let mut next_flush_at: Option = None; // B-260615-15:heartbeat interval 提至 loop 外复用,避免每轮重建计时器 // (每轮重建会丢已积累的节拍,且 interval 首次 tick 立即返回的特性会被误用)。 @@ -290,6 +319,7 @@ pub(crate) async fn stream_llm( // BUG-2026-07-14 根治: wall-clock deadline 检查(不依赖 select! timeout 语义)。 if tokio::time::Instant::now() >= idle_deadline { if !first_chunk_done { + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::InitFailed { retryable: true, error: format!( @@ -304,6 +334,7 @@ pub(crate) async fn stream_llm( text_len = full_text.len(), "[ai] 流中途 idle timeout(deadline),保文不重试(incomplete)", ); + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Partial { text: full_text, tool_calls: tool_calls_acc, @@ -312,10 +343,24 @@ pub(crate) async fn stream_llm( }; } - // 三分支 select!: + // AR-8: 到点 flush 合批的 delta(单条 AiTextDelta 双写 emit + event_bus)。 + if let Some(deadline) = next_flush_at { + if tokio::time::Instant::now() >= deadline { + next_flush_at = None; + flush_delta(app_handle, conv_id, &mut pending_delta); + } + } + + // 多分支 select!: // 1) chunk_rx.recv():来自专用线程的 LLM chunk (15s 保底 timeout,防 select! 死等) // 2) heartbeat.tick():静默期发 AiHeartbeat reset 前端 watchdog // 3) notify.notified():用户停止即时打断 + // 4) flush 定时器(sleep_until next_flush_at):尾部 delta 不因流静默而延迟(AR-8)。 + let flush_deadline = next_flush_at + .unwrap_or_else(|| tokio::time::Instant::now() + Duration::from_secs(3600)); + let flush_fut = tokio::time::sleep_until(flush_deadline); + tokio::pin!(flush_fut); + tokio::select! { chunk_result = tokio::time::timeout( Duration::from_secs(15), @@ -325,6 +370,7 @@ pub(crate) async fn stream_llm( Err(_elapsed) => { // 15s 保底 timeout 触发。正常路径不会到这里:心跳 30s 会先触发 select! 返回。 if !first_chunk_done { + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::InitFailed { retryable: true, error: format!( @@ -339,6 +385,7 @@ pub(crate) async fn stream_llm( text_len = full_text.len(), "[ai] 流中途 idle timeout,保文不重试(incomplete)", ); + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Partial { text: full_text, tool_calls: tool_calls_acc, @@ -355,13 +402,11 @@ pub(crate) async fn stream_llm( idle_deadline = tokio::time::Instant::now() + STREAM_IDLE_TIMEOUT; if !chunk.delta.is_empty() { full_text.push_str(&chunk.delta); - // L3 emit 双写 - let ev = AiChatEvent::AiTextDelta { - delta: chunk.delta, - conversation_id: Some(conv_id.to_string()), - }; - let _ = app_handle.emit("ai-chat-event", ev.clone()); - let _ = app_handle.state::().ai_event_bus.publish_event(ev); + // AR-8: 合批入 buffer,首个入 buffer 时置 flush 时刻(到点统一 emit 单条)。 + pending_delta.push_str(&chunk.delta); + if next_flush_at.is_none() { + next_flush_at = Some(tokio::time::Instant::now() + DELTA_FLUSH_INTERVAL); + } } if let Some(tc_deltas) = &chunk.tool_calls { accumulate_tool_calls(tc_deltas, &mut tool_calls_acc); @@ -395,6 +440,7 @@ pub(crate) async fn stream_llm( "[ai] provider 流式错误事件", ); if full_text.is_empty() && tool_calls_acc.is_empty() { + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::InitFailed { retryable, error: fmt_diag( @@ -405,6 +451,7 @@ pub(crate) async fn stream_llm( ), }; } + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Partial { text: full_text, tool_calls: tool_calls_acc, @@ -429,6 +476,7 @@ pub(crate) async fn stream_llm( "[ai] 流式接收中途错误(from 专用线程)", ); if full_text.is_empty() && tool_calls_acc.is_empty() { + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::InitFailed { retryable: classify_status_or_class(&status_or_class), error: fmt_diag( @@ -439,6 +487,7 @@ pub(crate) async fn stream_llm( ), }; } + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Partial { text: full_text, tool_calls: tool_calls_acc, @@ -450,11 +499,9 @@ pub(crate) async fn stream_llm( } // 2) 心跳:静默期 30s 发 AiHeartbeat,前端 watchdog reset(B-260615-02) _ = heartbeat.tick() => { - let ev = AiChatEvent::AiHeartbeat { + emit_ai_chat_event(app_handle, AiChatEvent::AiHeartbeat { conversation_id: Some(conv_id.to_string()), - }; - let _ = app_handle.emit("ai-chat-event", ev.clone()); - let _ = app_handle.state::().ai_event_bus.publish_event(ev); + }); } // 3) 即时停止唤醒(B-260615-14) _ = notify.notified() => { @@ -463,11 +510,19 @@ pub(crate) async fn stream_llm( break; } } + // 4) AR-8: delta 合批到点 flush(流静默/工具执行等待期也能按时发出尾部文本) + _ = &mut flush_fut => { + if next_flush_at.is_some() { + next_flush_at = None; + flush_delta(app_handle, conv_id, &mut pending_delta); + } + } } } // 用户停止 if stopped { + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Complete { text: full_text, tool_calls: tool_calls_acc, @@ -490,6 +545,7 @@ pub(crate) async fn stream_llm( text_len = full_text.len(), "[ai] 流尽未收到 finished 但有 partial_text,保文不重试(incomplete)", ); + flush_delta(app_handle, conv_id, &mut pending_delta); return StreamResult::Partial { text: full_text, tool_calls: tool_calls_acc, @@ -499,6 +555,7 @@ pub(crate) async fn stream_llm( } // 正常完成 + flush_delta(app_handle, conv_id, &mut pending_delta); StreamResult::Complete { text: full_text, tool_calls: tool_calls_acc, diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 1362119..18a6586 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -618,6 +618,15 @@ fn build_registry(db: Arc) -> NodeRegistry { registry.register("task_advance", move |_config| { Box::new(df_nodes::task_advance_node::TaskAdvanceNode::new(db.clone())) }); + // 补注册 5 个实现完整的孤儿节点(git/docker/http/notify/subflow): + // 均为无参单元结构体(impl Node),与 human 节点同风格注册(工厂闭包直接 Box 结构体, + // 无需捕获 db)。此前未接线,前端 DAG 用对应节点类型会报「未注册」。 + // "script" 节点保持禁用(ScriptNode 走 cmd/sh 无审批,见上方 R-PD-2 说明)。 + registry.register("git", |_config| Box::new(df_nodes::git_node::GitNode)); + registry.register("docker", |_config| Box::new(df_nodes::docker_node::DockerNode)); + registry.register("http", |_config| Box::new(df_nodes::http_node::HttpNode)); + registry.register("notify", |_config| Box::new(df_nodes::notify_node::NotifyNode)); + registry.register("subflow", |_config| Box::new(df_nodes::subflow_node::SubflowNode)); registry } diff --git a/src/components/ai/MessageList.vue b/src/components/ai/MessageList.vue index 9a43caf..6de80d9 100644 --- a/src/components/ai/MessageList.vue +++ b/src/components/ai/MessageList.vue @@ -171,6 +171,7 @@ const { scheduleStreamParse, clearStreamingState, cancelPendingRaf, + resetStreamCache, } = useStreamRenderer({ saveSelection, restoreSelection, @@ -715,6 +716,8 @@ watch(() => store.state.activeConversationId, () => { } else { cancelPendingRaf() } + // 切会话同步清流式块级 memo 缓存(_blockCache),防旧会话已缓存块被新会话命中。 + resetStreamCache() }) // 流式 rAF 清理:组件卸载时若仍有 pending rAF(streaming 中途切走/关窗),取消防泄漏。 diff --git a/src/composables/ai/useStreamRenderer.ts b/src/composables/ai/useStreamRenderer.ts index b731032..8c19690 100644 --- a/src/composables/ai/useStreamRenderer.ts +++ b/src/composables/ai/useStreamRenderer.ts @@ -207,6 +207,12 @@ export function useStreamRenderer(opts: UseStreamRendererOptions) { if (rafId !== null) { cancelAnimationFrame(rafId); rafId = null } } + /// 清空实例级块缓存(内容寻址 Map + LRU)。会话切换/开关变化时调用, + /// 防旧会话已缓存块被新会话命中(跨会话内容错乱)。仅清缓存,不动 streamingBlocks。 + function resetStreamCache(): void { + _blockCache.clear() + } + return { mdReady, renderMd, @@ -216,5 +222,6 @@ export function useStreamRenderer(opts: UseStreamRendererOptions) { renderStreamingBlocks, clearStreamingState, cancelPendingRaf, + resetStreamCache, } } diff --git a/src/views/Knowledge.vue b/src/views/Knowledge.vue index c1ab287..de3a382 100644 --- a/src/views/Knowledge.vue +++ b/src/views/Knowledge.vue @@ -161,6 +161,7 @@ import { } from '@/stores/knowledge' import { useRendered } from '@/composables/useMarkdown' import { useToast } from '@/composables/useToast' +import { usePersistedRef } from '@/composables/usePersistedRef' import { stripMd } from '@/utils/markdown' import KnowledgeDetail from '@/components/knowledge/KnowledgeDetail.vue' import type { KnowledgeDetailPayload, KnowledgeRecord } from '@/api/types' @@ -170,38 +171,10 @@ const store = useKnowledgeStore() // ⑤ 编辑/审核反馈:成功/失败明确提示(对齐 Settings.vue/Projects.vue useToast 模式) const { toast, showToast } = useToast() -// ④ topTab/activeKind/searchQuery 持久化 localStorage(跨刷新保留用户视图态) -// 仅持久化视图态:不持久化详情选中(id 跨刷新可能已失效)。容错:解析失败/无效值回退默认。 -const LS_KEY = 'kn.view' -function loadView(): { topTab: 'library' | 'inbox'; activeKind: string; searchQuery: string } { - try { - const raw = localStorage.getItem(LS_KEY) - if (!raw) return { topTab: 'library', activeKind: 'all', searchQuery: '' } - const v = JSON.parse(raw) - return { - topTab: v.topTab === 'inbox' ? 'inbox' : 'library', - activeKind: typeof v.activeKind === 'string' ? v.activeKind : 'all', - searchQuery: typeof v.searchQuery === 'string' ? v.searchQuery : '', - } - } catch { - return { topTab: 'library', activeKind: 'all', searchQuery: '' } - } -} -function saveView() { - try { - localStorage.setItem(LS_KEY, JSON.stringify({ - topTab: topTab.value, - activeKind: activeKind.value, - searchQuery: searchQuery.value, - })) - } catch { - /* localStorage 不可用(隐私模式/配额)静默降级,视图态非关键 */ - } -} -const persisted = loadView() - -// 顶层 Tab: library(知识库) | inbox(审核收件箱) -const topTab = ref<'library' | 'inbox'>(persisted.topTab) +// topTab/activeKind/searchQuery 持久化 localStorage(跨刷新保留用户视图态) +// 仅持久化视图态:不持久化详情选中(id 跨刷新可能已失效)。usePersistedRef 无类型校验, +// 解析失败/无效值回退 default,与 Tasks/Ideas 一致。 +const topTab = usePersistedRef<'library' | 'inbox'>('knowledge.topTab', 'library') const categories = [ { key: 'all', icon: '📦' }, @@ -211,8 +184,8 @@ const categories = [ function catLabel(key: string): string { return key === 'all' ? t('knowledge.categoryAll') : kindText(key) } -const activeKind = ref(persisted.activeKind) -const searchQuery = ref(persisted.searchQuery) +const activeKind = usePersistedRef('knowledge.activeKind', 'all') +const searchQuery = usePersistedRef('knowledge.searchQuery', '') let searchTimer: ReturnType | null = null function onSearchInput() { @@ -246,9 +219,6 @@ function switchTopTab(tab: 'library' | 'inbox') { } } -// ④ 视图态(topTab/activeKind/searchQuery)持久化:任一变化即写 localStorage -watch([topTab, activeKind, searchQuery], saveView) - // 关闭错误条 function clearError() { store.clearError()