优化: 前端体验批次(Knowledge持久化+流式缓存失效 + dfnodes节点注册 + delta 50ms合批 + MCP schema必填校验+bind_directory对齐)

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

crates/df-mcp/src/server.rs

@@ -215,7 +215,19 @@ pub(crate) async fn dispatch(ctx: &Ctx, read_only: bool, id: Option<Value>, 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<String> = 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<String> {
+    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<String> = 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::<String>::new());
+        assert_eq!(
+            required_names(&json!({ "type": "object", "required": false })),
+            Vec::<String>::new()
+        );
+    }
+
     #[tokio::test]
     async fn visible_predicate() {
         assert!(visible(false, RiskLevel::Low));

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}");
+    }
 }

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)
 }
 

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::<crate::state::AppState>().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<df_ai::provider::TokenUsage> = None;
     // BUG-260617-12: DeepSeek thinking 模式推理内容累积（多轮需回传）
     let mut reasoning_content_acc: Option<String> = 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<tokio::time::Instant> = 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 双写
+                            // AR-8: 合批入 buffer,首个入 buffer 时置 flush 时刻(到点统一 emit 单条)。
-                            let ev = AiChatEvent::AiTextDelta {
+                            pending_delta.push_str(&chunk.delta);
-                                delta: chunk.delta,
+                            if next_flush_at.is_none() {
-                                conversation_id: Some(conv_id.to_string()),
+                                next_flush_at = Some(tokio::time::Instant::now() + DELTA_FLUSH_INTERVAL);
-                            };
+                            }
-                            let _ = app_handle.emit("ai-chat-event", ev.clone());
-                            let _ = app_handle.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
                         }
                         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::<crate::state::AppState>().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,

src-tauri/src/state.rs

@@ -618,6 +618,15 @@ fn build_registry(db: Arc<Database>) -> 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
 }
 

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 中途切走/关窗),取消防泄漏。

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,
   }
 }

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(跨刷新保留用户视图态)
+// topTab/activeKind/searchQuery 持久化 localStorage(跨刷新保留用户视图态)
-// 仅持久化视图态:不持久化详情选中(id 跨刷新可能已失效)。容错:解析失败/无效值回退默认。
+// 仅持久化视图态:不持久化详情选中(id 跨刷新可能已失效)。usePersistedRef 无类型校验,
-const LS_KEY = 'kn.view'
+// 解析失败/无效值回退 default,与 Tasks/Ideas 一致。
-function loadView(): { topTab: 'library' | 'inbox'; activeKind: string; searchQuery: string } {
+const topTab = usePersistedRef<'library' | 'inbox'>('knowledge.topTab', 'library')
-  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)
 
 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 activeKind = usePersistedRef('knowledge.activeKind', 'all')
-const searchQuery = ref(persisted.searchQuery)
+const searchQuery = usePersistedRef('knowledge.searchQuery', '')
 let searchTimer: ReturnType<typeof setTimeout> | 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()