Files
DevFlow/src-tauri/src/commands/ai/audit/mod.rs
T
lxy f736f435bc 优化: 所有剩余UI/UX待办一批完成(持久化+AuditLog+解耦+total+原12大改+P2)
持久化(P1-c):新建 usePersistedRef composable,Tasks/AuditLog/ProjectDetail 等接入 localStorage

AuditLog(P1-d):后端 list_tool_executions 加 WHERE 筛选+返 {items,total,has_more},前端对接+长列折叠+筛选持久化

数据源解耦(P1-g):ProjectDetail projectTasks 按 project_id 独立加载 + ChatInput @项目联想独立加载(不读 store.tasks 当前页)

GitChanges(12a):后端 get_module_commits 加 git rev-list --count 返 total,前端显真实总数

原12大改:Dashboard 统计卡压底行(1)/Projects 列表卡片视图(2)/project_event 埋点排序(3)/TaskDetail 重设计(4)/IdeaDetail 重设计(5)/KnowledgeDetail 重设计(6)/界面持久化+侧栏Ctrl+B+审批数字键(7)/ProjectDetail 三栏改两栏(10)

P2打磨:文件浏览器(FileTree去重/FilePreview行号.md Diff/selectedFilePath归位)/settings反馈(假保存/端口校验)/AI会话(try-catch/scrollIntoView)/后端计数(move_queue事件/timeline total/workflow分页/import_batch分块)/杂项(TopBar/ConfirmDialog键盘/CIStatus i18n/ToolResultBody/ModuleNode/ApprovalDialog全选)
2026-08-02 13:11:06 +08:00

738 lines
42 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 工具调用审计 + pending 审批恢复 + 工具调用处理
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tauri::{AppHandle, Emitter, Manager};
use tokio::sync::Mutex;
use df_ai::ai_tools::{AiToolRegistry, RiskLevel};
use df_ai::provider::ChatMessage;
use df_storage::crud::AiToolExecutionRepo;
use df_storage::db::Database;
use super::{AiChatEvent, AiSession, ApprovalKind, PathAuthRequest, PendingApproval, ToolCallDraft};
// utils(audit/utils.rs):RiskLevel ↔ 字符串转换 + 字符串安全截断纯 helper。
// 第五批从本文件抽离,行为零变更。pub(crate) use 保持 finalize / restore 子模块
// super::risk_str / super::risk_from_str 引用路径不变;truncate_chars 供本文件 process_tool_calls
// DTO 装配 + record.rs 审计面板 IPC 共用。
mod utils;
pub(crate) use utils::{risk_from_str, risk_str, truncate_chars};
// diff(audit/diff.rs):write_file 审批预览 diff 生成。
// 第五批从本文件抽离,行为零变更。
mod diff;
// path_auth(audit/path_auth.rs):路径授权预校验。
// 第六批从本文件抽离,行为零变更。pub(super) use 供 tests 子模块引用 + process_tool_calls 裸名调用。
mod path_auth;
pub(super) use path_auth::{check_file_tool_auth, FileToolAuthOutcome};
// approval(audit/approval.rs):审批门控逻辑。
// 第六批从本文件抽离,行为零变更。use 保持 process_tool_calls 裸名调用。
mod approval;
use approval::{detect_retry_count, handle_approval_tool};
// record(audit/record.rs):审计记录写入 + 历史 DTO + 审计面板查询 IPC。
// 第六批从本文件抽离,行为零变更。pub(crate) use 供本文件 process_tool_calls /
// approval.rs 裸名调用 audit_tool_callpub use 保持 commands::ai::audit::* 路径透明
// list_tool_executions 是 #[tauri::command]ToolExecutionDto 供前端 DTO 序列化)。
pub mod record;
#[allow(unused_imports)]
pub(crate) use record::{audit_tool_call, query_audit_history, record_audit};
#[allow(unused_imports)]
pub use record::{list_tool_executions, ToolExecutionDto, ToolExecutionPage, ToolExecQuery};
// reason 拼装(resolve_project_label / resolve_task_label / build_approval_reason
// 拆至子模块 audit/reason.rs(第一批 helper 抽离,行为零变更)。
mod reason;
// build_approval_reason 现由 approval.rsinsert_pending_approval)内部调用,
// 通过 `use super::reason::build_approval_reason` 引入,本文件不再直接使用。
// restore(audit/restore.rs):启动恢复重建 pending_approvals。
// 第二批从本文件抽离,行为零变更。re-export 保持 commands::ai::audit::* 路径透明。
mod restore;
pub use restore::restore_pending_approvals;
// finalize(audit/finalize.rs):audit_finalize 审批后状态回填。
// audit_tool_call 已移至 record.rs(写入归 record 子模块)。
// 第二批从本文件抽离,行为零变更。pub(crate) use 供外部
//agentic/approval_timeout.rs / commands/chat.rs)裸名调用 audit_finalize。
mod finalize;
pub(crate) use finalize::audit_finalize;
// cache(audit/cache.rs):高危工具去重缓存 + 只读工具缓存。
// 第三批从本文件抽离,行为零变更。
mod cache;
pub(super) use cache::{find_cached_readonly_result, pending_placeholder_for};
// data_change(audit/data_change.rs):AR-11 数据变更联动刷新。
// 第四批从本文件抽离,行为零变更。pub(crate) use 保持 emit_data_changed 对 crate 内可见
// chat.rs 通过 commands::ai::audit::emit_data_changed 引入并调用),data_change_for_tool 私有。
mod data_change;
pub(crate) use data_change::emit_data_changed;
// idea_source(audit/idea_source.rs):灵感来源消息级溯源补全。
// create_idea 工具执行后,若 AI 未填 source 且有 message_id → 补 conv_msg:{id}(低侵入,不改 handler 接口)。
// pub(crate) use 供本文件 process_tool_calls + chat.rs 审批执行路径调用(单点逻辑,多调用点)。
mod idea_source;
pub(crate) use idea_source::maybe_fill_idea_source;
// ============================================================
// 工具调用心跳 + 审批处理
// ============================================================
/// 处理流式接收的工具调用:Low 风险并行执行(join_all),Med/High 进审批门控
/// 返回待审批的工具数量(0 = 全部自动执行完成)
/// 工具执行 + 心跳保活:execute 期间每 30s emit AiHeartbeat(对齐 stream_recv.rs
/// stream_llm select! 心跳语义),execute 完 abort 心跳 task。
///
/// 根治:工具执行在 stream_llm 之外(本模块),原本无 AiHeartbeat。
/// 单次 execute 超过前端 STREAM_TIMEOUT_MS130s)——bash 跑 cargo/测试、read 大文件、
/// 全盘 search 等开发长命令——前端 watchdog 误判断流,抛"工具已执行完成后续中断"误报
/// (实测:用户报"一边流一边抛",前一轮 delta 文本在屏 + 当前轮工具执行静默 > 130s)。
/// 补工具执行阶段的心跳缺口,前端 watchdog 在静默期也能收到 reset,不再误杀长工具。
async fn execute_with_heartbeat(
tools: &AiToolRegistry,
name: &str,
args: serde_json::Value,
app: &AppHandle,
conv_id: &str,
tool_call_id: &str,
) -> anyhow::Result<serde_json::Value> {
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::time::Duration;
// RAII guarddrop 时 stop+abort 心跳 task,确保 execute panicunwind)也不泄漏心跳 task。
struct HeartbeatGuard {
stop: Arc<AtomicBool>,
handle: tokio::task::JoinHandle<()>,
}
impl Drop for HeartbeatGuard {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
self.handle.abort();
}
}
let stop = Arc::new(AtomicBool::new(false));
let app_h = app.clone();
let conv = conv_id.to_string();
let stop_c = stop.clone();
// 心跳 task:弃首 ticktokio interval 首 tick 立即返回,对齐 stream_recv.rs:174),
// 后每 30s emit AiHeartbeat< 130s watchdog 确保覆盖)。
let heartbeat = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(30));
interval.tick().await;
loop {
interval.tick().await;
if stop_c.load(Ordering::SeqCst) {
break;
}
let ev = AiChatEvent::AiHeartbeat { conversation_id: Some(conv.clone()) };
let _ = app_h.emit("ai-chat-event", ev.clone());
let _ = app_h.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
}
});
// RAIIexecute 无论 Ok/Err/panicunwind),_guard drop 自动 stop+abort,无心跳 task 泄漏。
let _guard = HeartbeatGuard { stop, handle: heartbeat };
// tools.execute 无 timeout 时,卡死工具(run_command 长命令/read_file 大文件/同步
// 阻塞工具)永久挂起 → process_tool_calls 持 session lock 永久 → guard.reset 等 lock → AiCompleted
// 永不发 → 前端"回答完卡住/超时清空"。外层 timeout 兜底:超时返错误 tool_result,锁释放,loop 续跑。
// 心跳 30s 续命前端 watchdog。
//
// 外层 timeout 取值(BUF-run-cmd-timeout):原硬编码 60s 会先于 run_command 自带 timeout_secs
// (上限 600s)drop,致 run_command 跑长构建(cargo/npm)永远到不了用户配的 timeout_secs。
// run_command 分支从 args.timeout_secs 取值(clamp [60,600],默认 60,与 tool_registry.rs:2707
// run_command 内部 clamp 同源),其余工具仍 60s。run_command 内部还有自己的子 timeout,
// 外层只需 >= 内部 timeout 即不抢断(run_command 内部超时会返带语义的 tool_result,优于外层裸中止)。
let outer_secs: u64 = if name == "run_command" {
args.get("timeout_secs")
.and_then(|v| v.as_u64())
.unwrap_or(60)
.clamp(60, 600)
} else {
60
};
// run_command 实时流式:经 task-local sink 注入(AppHandle + tool_call_id + conv_id),
// handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。
// 非 run_command 不注入 sink(handler 内 emit_output 静默 noop),零行为变更。
let sink = if name == "run_command" {
Some(crate::commands::ai::command_stream::CommandSink::new(
app.clone(),
tool_call_id.to_string(),
Some(conv_id.to_string()),
))
} else {
None
};
let exec_fut = crate::commands::ai::command_stream::scope(sink, tools.execute(name, args));
match tokio::time::timeout(Duration::from_secs(outer_secs), exec_fut).await {
Ok(result) => result,
Err(_elapsed) => {
tracing::error!(
conv_id = %conv_id,
tool = %name,
outer_secs,
"[ai] 工具执行超时({}s),返回错误 tool_result(防 process_tool_calls 持 session lock 永久卡死)",
outer_secs
);
Err(anyhow::anyhow!("工具执行超时({}s),已中止(防死锁)", outer_secs))
}
}
}
pub(crate) async fn process_tool_calls(
session_arc: &Arc<Mutex<AiSession>>,
tool_calls_acc: HashMap<u32, ToolCallDraft>,
tools_arc: &Arc<AiToolRegistry>,
db: &Arc<Database>,
app_handle: &AppHandle,
conv_id: &str,
) -> usize {
let mut tc_list: Vec<_> = tool_calls_acc.into_iter().collect();
tc_list.sort_unstable_by_key(|(i, _)| *i);
// 治本兜底:LLM 异常复用同 tool_use.idstream_recv 按 content_block index 分桶,
// 同 id 不同 index draft 可并存 → 每 draft emit AiToolCallStarted 致同 id emit 两次 → 前端 push 两卡,
// Completed 按 id 只 update 首张 → 次张残留 running 0行)。process 层按 id 去重——同 id 保留
// 最小 index 的首个,丢弃后续,保证 emit Started 的 id 唯一。前端 useAiEvents.ts:205 findToolCall
// 守卫双保险。
let mut seen_ids: HashSet<String> = HashSet::new();
tc_list.retain(|(_, draft)| seen_ids.insert(draft.id.clone()));
let mut pending_count = 0usize;
let audit_repo = AiToolExecutionRepo::new(db);
// P1 消息级溯源:取当前 assistant 消息 id。
// 调用前 agentic/mod.rs 已把本轮 assistant_with_tools 消息(LLM 返回带 tool_calls 的那条)
// push 到 per_conv.messagesaudit/mod.rs:882),此处取末条 assistant id 作为本轮工具
// 调用所属的溯源 message_id,贯穿所有 audit_tool_call 写入。None 表示无 assistant 消息
// (异常路径/老数据无 id),audit 落 message_id=None,展示侧兼容。
//
// 短 lock 段读 current_message_id(纯读,无 await),clone 出来后续传参。
// 整个 process_tool_calls 不再要求调用方持锁,内部所有 session.xxx 访问均短 lock 段,
// 慢操作(execute_with_heartbeat/audit_tool_call/detect_retry_count/DB 查)全在锁外 await。
// current_message_id 读加 200ms timeout(防 session lock 竞争卡死致
// process_tool_calls 进不去 emit AiToolCallStarted → 前端工具卡片不呈现 + 45s 看门狗超时)。
// 超时用 None(溯源降级,非致命)。治本:消除 session lock 长持有(save clone 段已优化)。
let current_message_id: Option<String> = match tokio::time::timeout(
std::time::Duration::from_millis(200),
async {
let __lock_t = std::time::Instant::now();
let session = session_arc.lock().await;
let __id = session.conv_read(conv_id)
.and_then(|c| c.messages.last_assistant_message_id());
let __hold = __lock_t.elapsed();
if __hold > std::time::Duration::from_millis(30) {
eprintln!("[LOCK-SLOW] process_tool_calls:192 持锁 {:?} (含 lock 等待)", __hold);
}
__id
},
).await {
Ok(id) => id,
Err(_) => {
tracing::warn!(conv_id = %conv_id, "[ai] process_tool_calls current_message_id lock 200ms 超时,溯源降级 None(session lock 竞争)");
None
}
};
let current_message_id = current_message_id.as_deref();
// 解析 args + 批量发 Started(前端骨架按原始 index 顺序展示)
let drafts: Vec<(u32, ToolCallDraft, serde_json::Value)> = tc_list.into_iter()
.map(|(idx, draft)| {
let args = serde_json::from_str(&draft.args).unwrap_or(serde_json::Value::Object(Default::default()));
// L3 emit 双写:AiToolCallStarted publish 到事件总线(tunnel subscriber 透传 miniapp)。
// AiTextDelta/AiToolCall* 高频事件不双写原则不适用此处(工具调用生命周期事件属关键状态变更,需透传)。
let ev = AiChatEvent::AiToolCallStarted {
id: draft.id.clone(),
name: draft.name.clone(),
args: args.clone(),
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);
(idx, draft, args)
})
.collect();
// ── 文件工具路径授权预校验 ──
// 在 RiskLevel 分类前,对文件工具(read/write/list/patch/info/append/delete/rename/search
// 逐条预校验路径授权(persistent + 会话 session_allowed_dirs + 黑名单):
// - 任一路径命中黑名单 → Deniedpush 错误 tool_result + emit Completed,不挂起(Phase C
// - 任一路径未命中白名单且非黑名单 → NeedsAuth:挂起 loopinsert pending + emit AiDirAuthRequired),
// 等用户经 ai_authorize_dir IPC 决定"仅本次/未来都允许/拒绝"后恢复(Phase B
// - 全部已授权 → 放行,进入下方 RiskLevel 分类走原 Low/Med/High 流程
// 预校验通过的 drafts 留在原 Vec 继续走原流程;NeedsAuth/Denied 的 drafts 从 Vec 移除单独处理。
let allowed_snapshot = app_handle
.state::<crate::state::AppState>()
.allowed_dirs
.clone();
let allowed_guard = allowed_snapshot.read().await.clone();
// 收集 NeedsAuth/Denied drafts(从 drafts 移除,不进下方 RiskLevel 循环)
let mut path_auth_pending: Vec<(ToolCallDraft, serde_json::Value, PathAuthRequest)> = Vec::new();
let mut path_denied: Vec<(ToolCallDraft, String)> = Vec::new();
let mut authorized_drafts: Vec<(u32, ToolCallDraft, serde_json::Value)> = Vec::new();
for (idx, draft, args) in drafts {
// 会话级临时授权目录在 allowed_guard.session 内(进程级,随 active 会话切换清空),
// handler 闭包 read lock 与此处预校验读同一字段,两端授权判定一致。
match check_file_tool_auth(&draft.name, &args, &allowed_guard) {
FileToolAuthOutcome::Authorized => authorized_drafts.push((idx, draft, args)),
FileToolAuthOutcome::NeedsAuth(req) => path_auth_pending.push((draft, args, req)),
FileToolAuthOutcome::Denied(reason) => path_denied.push((draft, reason)),
}
}
let drafts = authorized_drafts;
// Phase C: Denied 路径 → 硬拒(push 错误 tool_result + emit Completed),不挂起 loop。
// 工具返 Err 让 LLM 知路径被禁,自行调整;loop 继续下一轮(不暂停)。
for (draft, reason) in path_denied {
// 拒绝消息改结构化 JSON,防裸文本破坏 tool role 消息协议
let err_msg = serde_json::json!({
"status": "rejected",
"reason": "path_blacklist",
"message": format!("路径授权拒绝: {}", reason)
}).to_string();
// 短 lock 段:push tool_result(纯写,无 await)
{
let __lock_t = std::time::Instant::now();
let mut session = session_arc.lock().await;
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &err_msg));
let __hold = __lock_t.elapsed();
if __hold > std::time::Duration::from_millis(30) {
eprintln!("[LOCK-SLOW] process_tool_calls:262 持锁 {:?} (含 lock 等待)", __hold);
}
}
// L3 emit 双写:路径黑名单拒绝 emit Completed 双路发布。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(err_msg.clone()),
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);
// 审计:路径黑名单拒绝(decided_by=auto_blacklist),留痕可追溯
let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High);
audit_tool_call(
&audit_repo, conv_id, &draft.id, &draft.name, &draft.args,
"rejected", risk_level, Some(err_msg), Some("auto_blacklist"),
current_message_id,
).await;
}
// Phase B: NeedsAuth 路径 → 挂起 loopinsert pending + emit AiDirAuthRequired + push 占位 tool_result)。
// 复用审批挂起架构:pending_approvals 以 tool_call_id 为键,ai_authorize_dir IPC remove 后恢复。
// pending_count 计入(让 agentic loop 检测到挂起并暂停,等 ai_authorize_dir → try_continue 恢复)。
//
// detect_retry_count(DB 读)/emit/audit_tool_call(DB 写)均在锁外,
// 仅 push tool_result + pending_approvals.insert 两处纯写改短 lock 段。
for (draft, args, req) in path_auth_pending {
pending_count += 1;
// L1 补丁:req.dirs 含所有未授权父目录;emit 用首个作主展示目录(前端弹窗主显)。
// ai_authorize_dir 消费时遍历所有 dirs 写白名单。
let dir_str = req.dirs.first()
.map(|d| d.to_string_lossy().to_string())
.unwrap_or_default();
let path_str = req.raw_paths.first().cloned().unwrap_or_default();
// 锁外:(容错/恢复,开关 df-ai-approval-retry):同 tc_id 重试检测。
let retry_count = detect_retry_count(&audit_repo, &draft.id).await;
if retry_count >= 1 {
let skip_msg = format!(
"已跳过重试(同 tool_call_id={} 此前已审批执行过,防 LLM 死循环重试同卡死工具)",
draft.id
);
// 短 lock 段:push tool_result(纯写)
{
let __lock_t = std::time::Instant::now();
let mut session = session_arc.lock().await;
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &skip_msg));
let __hold = __lock_t.elapsed();
if __hold > std::time::Duration::from_millis(30) {
eprintln!("[LOCK-SLOW] process_tool_calls:305 持锁 {:?} (含 lock 等待)", __hold);
}
}
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(skip_msg.clone()),
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);
let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High);
audit_tool_call(
&audit_repo, conv_id, &draft.id, &draft.name, &draft.args,
"skipped_retry", risk_level, Some(skip_msg), Some("auto_retry_guard"),
current_message_id,
).await;
continue;
}
// 短 lock 段:pending_approvals.insert + 占位 tool_result push(两处纯写,无 await)
{
let __lock_t = std::time::Instant::now();
let mut session = session_arc.lock().await;
session.pending_approvals.insert(
draft.id.clone(),
PendingApproval {
tool_call_id: draft.id.clone(),
tool_name: draft.name.clone(),
arguments: args.clone(),
conversation_id: Some(conv_id.to_string()),
recovered: false,
// 路径授权挂起标 kind=Path(req)(下沉原 path_auth 字段)。
kind: ApprovalKind::Path(req),
retry_count,
created_at: Some(std::time::SystemTime::now()),
},
);
// 占位 tool_result(与 RiskLevel 审批一致),ai_authorize_dir 批准后替换为真实结果。
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &pending_placeholder_for(&draft.id)));
let __hold = __lock_t.elapsed();
if __hold > std::time::Duration::from_millis(30) {
eprintln!("[LOCK-SLOW] process_tool_calls:325 持锁 {:?} (含 lock 等待)", __hold);
}
}
tracing::debug!(target: "ai_dirauth", conv = %conv_id, tool = %draft.name, tc_id = %draft.id, "emit AiDirAuthRequired(路径授权挂起,等 ai_authorize_dir 恢复)");
let ev = AiChatEvent::AiDirAuthRequired {
id: draft.id.clone(),
tool: draft.name.clone(),
path: path_str,
dir: dir_str,
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);
let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High);
audit_tool_call(
&audit_repo, conv_id, &draft.id, &draft.name, &draft.args,
"pending", risk_level, None, None,
current_message_id,
).await;
}
// 分类:Low 收集并行执行,Med/High 立即进审批门控(push 占位 tool_result
//
// High risk 在进审批门前先查去重缓存(find_cached_high_risk_result)。
// 若 LLM 重试同命令(同 tool_name + 同 args,键序无关),命中已落定的旧 tool_result
// 把缓存结果作为新 tool_call_id 的 tool_result 回传 LLM,跳过 insert pending + 跳过审批,
// 断「超时→重试→重新审批」循环。Med 不去重(去重易误伤),Low 无审批本就不进此分支。
// F-#97low_risk 向量携带 risk_level(治 securityReview blocker:710 审计失真)。
// 原 Vec 仅 (draft,args),回填审计硬编码 RiskLevel::Low#97 让 Med/Highall/medium 模式)
// 也进此向量,需透传真实等级,审计方可追溯「all 模式执行了多少高危命令」。
let mut low_risk: Vec<(ToolCallDraft, serde_json::Value, RiskLevel)> = Vec::new();
// F-#97 自动执行范围三档(2026-06-22):KV df-ai-auto-execute-mode 取值 low/medium/all,默认 low=现状)。
// low:仅 Low 自动(等价旧行为);mediumLow+Medium 自动;all:全自动无审批(完全 AI 接管)。
let auto_exec_mode = app_handle
.state::<crate::state::AppState>()
.settings
.get("df-ai-auto-execute-mode")
.await
.ok()
.flatten()
.unwrap_or_else(|| "low".to_string());
// F-#97 诊断:用户报「选 all 仍审批」时辅助定位 KV 是否真落库。
// df-ai-trace-approve 开关控制(默认开)。常见误以为已生效但实际未点确认按钮致 KV 未落。
let trace_on = app_handle
.state::<crate::state::AppState>()
.settings
.get("df-ai-trace-approve")
.await
.ok()
.flatten()
.map(|v| v == "true" || v == "1")
.unwrap_or(true);
if trace_on {
tracing::info!(
conv_id = %conv_id,
auto_exec_mode = %auto_exec_mode,
drafts_count = drafts.len(),
"[ai] 审批判定读取的自动执行模式"
);
}
// trust_hits 收集:AE-04 会话信任命中(同会话已批准同类操作),真实执行但移到锁外 spawn
// (对齐 Low risk 不持锁模式)。命中时此处只 emit toast + 收集,不 .await execute。
let mut trust_hits: Vec<(ToolCallDraft, serde_json::Value, String, RiskLevel)> = Vec::new();
for (_, draft, args) in drafts {
let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High);
handle_approval_tool(
draft, args, risk_level, &auto_exec_mode,
session_arc, conv_id, tools_arc, &audit_repo,
app_handle, db, &mut low_risk, &mut trust_hits,
&mut pending_count, current_message_id,
).await;
}
// trust-hit 并行执行:execute + 即时 emit 在闭包内(闭包不访问 session,
// 调用方不再持锁,本段所有 await 完全在锁外),
// push tool_result / audit 在 join_all 后串行回填(短 lock push + 锁外 audit)。对齐 Low risk 并行模式。
// CR-51 修:原 inline .await execute 串行执行每个工具(阻塞期间锁被持有,run_command 慢命令
// 阻塞同会话 IPC);改 join_all 并行多工具减少总阻塞时间。P0 重构后锁外并行,根本消除阻塞。
// join_all 保序——结果顺序 = trust_hits 输入顺序 = tc_list 原始 index 顺序,不额外 sort
if !trust_hits.is_empty() {
let results: Vec<(ToolCallDraft, RiskLevel, Result<String, String>)> =
futures::future::join_all(trust_hits.into_iter().map(|(draft, args, _dir_label, risk_level)| {
let tools = tools_arc.clone();
let app_clone = app_handle.clone();
let conv_clone = conv_id.to_string();
async move {
let exec_result = execute_with_heartbeat(&tools, &draft.name, args, &app_clone, &conv_clone, &draft.id).await;
match exec_result {
Ok(val) => {
let content = val.to_string();
// L3 emit 双写:trust 放行工具执行成功 emit Completed 双路发布。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(content.clone()),
conversation_id: Some(conv_clone),
};
let _ = app_clone.emit("ai-chat-event", ev.clone());
let _ = app_clone.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
// AR-11:数据变更类工具执行成功后 emit df-data-changed 联动刷新
// write_file/run_command 不在 data_change_for_tool 映射内,emit_data_changed 内部 None 即 noop
emit_data_changed(&app_clone, &draft.name);
(draft, risk_level, Ok(content))
}
Err(e) => {
let content = e.to_string();
// L3 emit 双写:trust 放行工具执行失败 emit Completed 双路发布。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(content.clone()),
conversation_id: Some(conv_clone),
};
let _ = app_clone.emit("ai-chat-event", ev.clone());
let _ = app_clone.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
(draft, risk_level, Err(content))
}
}
}
})).await;
// 串行回填 tool_result + 审计(短 lock push + 锁外 audit
for (draft, risk_level, outcome) in results {
let (status, content) = match outcome {
Ok(c) => ("completed", c),
Err(c) => ("failed", c),
};
// 短 lock 段:push tool_result(纯写,无 await)
{
let mut session = session_arc.lock().await;
// T2: 大工具结果进 namespace
// 失败包结构化 envelope {"status":"failed","error":...}:让断路器按 status 判
// (不解析内容文本,理论可靠)。emit 给前端的 result 与 audit 仍用原 err_msg 纯文本
// (前端卡片/审计可读,与断路器信号解耦)。
let msg_content = if status == "failed" {
serde_json::json!({"status":"failed","error":&content}).to_string()
} else if df_ai::namespace_store::should_use_namespace(&content, &draft.name) {
session.namespace_store.store(&draft.name, &content)
} else {
content.clone()
};
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &msg_content));
}
// 锁外:审计(trust 放行仍记一条,decided_by=auto_trust),留痕可追溯
audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, status, risk_level, Some(content), Some("auto_trust"), current_message_id).await;
}
}
// Low 风险并行执行:execute + 即时 emit 在闭包内(不持 session 锁),
// push tool_result / audit 在 join_all 后串行回填(短 lock push + 锁外 audit,与 Med/High 占位拼接)。
// join_all 保序——结果顺序 = low_risk 输入顺序 = tc_list 原始 index 顺序,不额外 sort
if !low_risk.is_empty() {
// 只读幂等工具去重缓存(治 LLM 死循环重调)。
// 实测 9357c27c 会话:LLM 对 read_file(application.xml) 连调 17 次、search_files 连调 13 次,
// 每次返回几乎相同结果却反复重调,是 prompt 64 万的直接元凶。此处先查会话内是否已对
// 同参只读工具成功执行过,命中则直接回填缓存结果跳过真执行,断 LLM 失忆死循环。
// 安全边界见 find_cached_readonly_result 文档(仅白名单只读工具 + 仅 completed 成功结果)。
//
// find_cached_readonly_result 内部短 lock + 锁外 DB 查,本段不持锁。
let mut low_risk_uncached: Vec<(ToolCallDraft, serde_json::Value, RiskLevel)> = Vec::with_capacity(low_risk.len());
for (draft, args, risk_level) in low_risk {
let cached = find_cached_readonly_result(session_arc, conv_id, &audit_repo, &draft.name, &args).await;
if let Some(cached_content) = cached {
// 缓存命中:直接 push tool_result + 审计(decided_by=cache_hit 标记缓存来源),
// 不走真执行 + 不重emit Started/Completed(避免误导前端工具又执行了一次)。
// emit Completed 携带缓存结果供前端折叠卡片展示(与 find_cached_high_risk_result 一致)。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(cached_content.clone()),
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);
// 短 lock 段:push tool_result(纯写,无 await)
{
let mut session = session_arc.lock().await;
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &cached_content));
}
audit_tool_call(
&audit_repo, conv_id, &draft.id, &draft.name, &draft.args,
"completed", risk_level, Some(cached_content), Some("cache_hit"),
current_message_id,
).await;
tracing::info!(
conv_id = %conv_id,
tool = %draft.name,
tc_id = %draft.id,
"[ai] 只读工具缓存命中,跳过真执行(治 LLM 死循环重调)"
);
} else {
low_risk_uncached.push((draft, args, risk_level));
}
}
let low_risk = low_risk_uncached;
// 携带 args + 原始 JSON resultcreate_idea source 补全需解析 result.id + args.source)。
let results: Vec<(ToolCallDraft, serde_json::Value, RiskLevel, Result<String, String>)> =
futures::future::join_all(low_risk.into_iter().map(|(draft, args, risk_level)| {
let tools = tools_arc.clone();
let app_clone = app_handle.clone();
let conv_clone = conv_id.to_string();
async move {
let result = execute_with_heartbeat(&tools, &draft.name, args, &app_clone, &conv_clone, &draft.id).await;
match result {
Ok(val) => {
// L3 emit 双写:Low 风险工具执行成功 emit Completed 双路发布。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: val.clone(),
conversation_id: Some(conv_clone),
};
let _ = app_clone.emit("ai-chat-event", ev.clone());
let _ = app_clone.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
// AR-11(方案A):数据变更工具执行成功后 emit df-data-changed,
// 前端 store listen 刷新列表(仅命中映射的工具 emit,见 emit_data_changed)。
emit_data_changed(&app_clone, &draft.name);
(draft, val.clone(), risk_level, Ok(val.to_string()))
}
Err(e) => {
// AR-6(定向B):Low 工具失败不 emit AiError。
// 原逻辑 emit AiError 会令前端置 streaming=false + 错误气泡,但 process_tool_calls
// 仍返回 pending_count=0agentic loop 续下一轮 → 前端 false 后端跑,状态紊乱。
// 现改为正常 emit AiToolCallCompletedresult=错误信息),错误包进 tool_result
// 让 LLM 看到工具失败自行决定下一步,loop 正常续,前端状态一致。
let err_msg = format!("工具 {} 执行失败: {}", draft.name, e);
// L3 emit 双写:Low 风险工具执行失败 emit Completed 双路发布。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(err_msg.clone()),
conversation_id: Some(conv_clone),
};
let _ = app_clone.emit("ai-chat-event", ev.clone());
let _ = app_clone.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
(draft, serde_json::Value::Null, risk_level, Err(err_msg))
}
}
}
})).await;
// 串行回填 tool_result + 审计(短 lock push + 锁外 audit
for (draft, raw_result, risk_level, outcome) in results {
let (status, content) = match outcome {
Ok(c) => ("completed", c),
Err(c) => ("failed", c),
};
// source 消息级溯源补全(仅 source 空 + 有 message_id)。
// 注:低风险路径目前 create_idea 不会进(Medium→pending),此处为防御/未来若调级别覆盖。
// 仅 Ok(completed) 时补(失败 result 无 idea_id 意义);args 从 draft.args 反解(JSON 原样)。
if status == "completed" {
let args_val = serde_json::from_str(&draft.args).unwrap_or(serde_json::Value::Null);
maybe_fill_idea_source(db, &draft.name, &args_val, &raw_result, current_message_id).await;
}
// 短 lock 段:写 per_conv.messages(纯写,无 await)
{
let mut session = session_arc.lock().await;
// T2: 大工具结果进 namespace,主队列只留引用
// 失败包结构化 envelope {"status":"failed","error":...}:让断路器按 status 判
// (不解析内容文本,理论可靠)。emit 给前端的 result 与 audit 仍用原 err_msg 纯文本
// (前端卡片/审计可读,与断路器信号解耦)。
let msg_content = if status == "failed" {
serde_json::json!({"status":"failed","error":&content}).to_string()
} else if df_ai::namespace_store::should_use_namespace(&content, &draft.name) {
session.namespace_store.store(&draft.name, &content)
} else {
content.clone()
};
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &msg_content));
}
// F-#97 审计留痕:low_risk 向量中 risk_level != Low 即 mode 放行(Med 仅 medium/all、High 仅 all 才进)。
// decided_by 区分接管来源,审计表可追溯 all 模式执行了多少高危命令(治 securityReview blocker:
// 原硬编码 RiskLevel::Low 致 Med/High 接管操作全标 Low/auto 失真,等于没记)。
let decided_by: &str = match risk_level {
RiskLevel::Low => "auto",
RiskLevel::Medium => "auto_takeover_medium",
RiskLevel::High => "auto_takeover_all",
};
audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, status, risk_level, Some(content), Some(decided_by), current_message_id).await;
}
}
// 单表 pending_approvals 统计 pending 总数(path/risk 合一,kind 区分)。
// 短 lock 段:读 pending_approvals 统计(纯读,无 await)
let (path_count, risk_count, pending_approvals_len) = {
let session = session_arc.lock().await;
let len = session.pending_approvals.len();
let (p, r) = session.pending_approvals.values()
.fold((0usize, 0usize), |(p, r), a| match a.kind {
ApprovalKind::Path(_) => (p + 1, r),
ApprovalKind::Risk { .. } => (p, r + 1),
});
(p, r, len)
};
tracing::debug!(target: "ai_dirauth", pending_count, pending_approvals_len, path_count, risk_count, "process_tool_calls 返回(路径/风险挂起分布)");
pending_count
}
#[cfg(test)]
mod tests {
use super::*;
use super::path_auth::extract_file_tool_paths;
/// grep 走单路径授权申请路径(NeedsAuth),非 search_files 盲拒(Denied)。
///
/// grep 加入 extract_file_tool_paths 单路径分支,未授权路径触发
/// AiDirAuthRequired 申请(对齐 read_file),不像 search_files 被硬拒。
/// 锁定此差异:grep 与 read_file 同款授权弹窗语义。
#[test]
fn test_grep_unauthorized_path_triggers_auth_not_denied() {
// 空白名单(不含 workspace_root 的目录):任何路径都未授权 → NeedsAuth
let allowed = crate::state::AllowedDirs::default();
let args = serde_json::json!({ "pattern": "foo", "path": "C:/some/unauthorized/dir" });
let outcome = check_file_tool_auth("grep", &args, &allowed);
match outcome {
FileToolAuthOutcome::NeedsAuth(_) => { /* 期望:grep 触发授权申请 */ }
FileToolAuthOutcome::Authorized => panic!("grep 未授权应返 NeedsAuth(申请授权),实际 Authorized"),
FileToolAuthOutcome::Denied(_) => panic!("grep 未授权应返 NeedsAuth(申请授权),实际 Deniedgrep 不应像 search_files 盲拒)"),
}
// 对照:search_files 同样未授权但被硬拒(Denied),锁定两工具差异
let search_outcome = check_file_tool_auth("search_files", &args, &allowed);
match search_outcome {
FileToolAuthOutcome::Denied(_) => { /* 期望:search_files 盲拒 */ }
FileToolAuthOutcome::NeedsAuth(_) => panic!("search_files 未授权应返 Denied(盲拒),实际 NeedsAuth"),
FileToolAuthOutcome::Authorized => panic!("search_files 未授权应返 Denied(盲拒),实际 Authorized"),
}
}
/// grep 黑名单路径(.ssh 等)直接 Denied(不申请授权),对齐 read_file。
#[test]
fn test_grep_blacklisted_path_denied() {
let allowed = crate::state::AllowedDirs::default_with_root();
// .ssh 命中系统黑名单(is_in_system_blacklist
let args = serde_json::json!({ "pattern": "foo", "path": "/home/user/.ssh/config" });
let outcome = check_file_tool_auth("grep", &args, &allowed);
match outcome {
FileToolAuthOutcome::Denied(_) => { /* 期望:黑名单硬拒 */ }
FileToolAuthOutcome::NeedsAuth(_) => panic!("grep 黑名单路径应返 Denied,实际 NeedsAuth"),
FileToolAuthOutcome::Authorized => panic!("grep 黑名单路径应返 Denied,实际 Authorized"),
}
}
/// extract_file_tool_pathsgrep 取 args["path"](单路径分支)
#[test]
fn test_extract_grep_path() {
let args = serde_json::json!({ "pattern": "foo", "path": "src/main.rs", "glob": "*.rs" });
let paths = extract_file_tool_paths("grep", &args);
assert_eq!(paths, vec!["src/main.rs".to_string()], "grep 应取 path 参数(忽略 glob 等其他参数)");
}
}