绝尘
f82dd8ba90
- F-06 导入历史项目：scan.rs extract_description(README 首段 200 字) + import_project 命令(合并 create+bind，spawn_blocking 探测栈) + 前端 ProjectDetail 导入按钮 + store action；主代理补 lib.rs invoke_handler 注册(agent 未注册) - F-02 技能联想使用：核对发现 sendMessage→ai_chat_send skill 调用链已完整(联想选中→带 skill→注入 SKILL.md)，补 Esc 取消选中兜底 - AR-6 Low 失败语义统一(定向B)：audit.rs Low 失败 emit AiToolCallCompleted(错误 result) 替代 AiError，消除前端 false/后端 loop 续紊乱，错误回填 tool_result 让 LLM 自处理 cargo/vue-tsc 0 err
342 lines
15 KiB
Rust
342 lines
15 KiB
Rust
//! 工具调用审计 + pending 审批恢复 + 工具调用处理
|
||
|
||
use std::collections::HashMap;
|
||
use std::sync::Arc;
|
||
|
||
use tauri::{AppHandle, Emitter};
|
||
|
||
use df_ai::ai_tools::{AiToolRegistry, RiskLevel};
|
||
use df_ai::provider::ChatMessage;
|
||
use df_storage::crud::{AiToolExecutionRepo, ProjectRepo};
|
||
use df_storage::db::Database;
|
||
use df_storage::models::AiToolExecutionRecord;
|
||
|
||
use df_core::types::new_id;
|
||
|
||
use crate::state::AppState;
|
||
|
||
use crate::commands::now_millis;
|
||
|
||
use super::{AiChatEvent, AiSession, PendingApproval, ToolCallDraft};
|
||
|
||
/// RiskLevel → 审计记录字符串（low/medium/high）
|
||
pub(crate) fn risk_str(r: RiskLevel) -> &'static str {
|
||
match r {
|
||
RiskLevel::Low => "low",
|
||
RiskLevel::Medium => "medium",
|
||
RiskLevel::High => "high",
|
||
}
|
||
}
|
||
|
||
/// 审计记录字符串 → RiskLevel（启动重建 pending_approvals 用,未知串返回 None 跳过）
|
||
pub(crate) fn risk_from_str(s: &str) -> Option<RiskLevel> {
|
||
match s {
|
||
"low" => Some(RiskLevel::Low),
|
||
"medium" => Some(RiskLevel::Medium),
|
||
"high" => Some(RiskLevel::High),
|
||
_ => None,
|
||
}
|
||
}
|
||
|
||
/// 查项目可读标签：id → "「项目名」(id=x)"，查不到回退 "(id=x)"，空 id 返回空串。
|
||
///
|
||
/// AR-3：审批卡片需显示对象名而非裸 id（用户反馈"只返回 ID 不知道是什么数据"）。
|
||
async fn resolve_project_label(db: &Arc<Database>, id: &str) -> String {
|
||
if id.is_empty() {
|
||
return String::new();
|
||
}
|
||
let repo = ProjectRepo::new(db);
|
||
match repo.get_by_id(id).await {
|
||
Ok(Some(p)) => format!("「{}」(id={})", p.name, id),
|
||
_ => format!("(id={})", id),
|
||
}
|
||
}
|
||
|
||
/// 拼审批 reason：按工具名 + args 取可读字段，让用户知道"审批要做什么"。
|
||
///
|
||
/// 主要审批工具特化（带对象名/标识）：
|
||
/// - delete_project/restore_project/purge_project → 查项目名拼"删除项目「名」(id=x)"
|
||
/// - update_project → 查项目名 + field
|
||
/// - bind_directory → path + 查项目名
|
||
/// - create_task → title + 查 project_id 项目名；create_project → name；create_idea → title
|
||
/// - run_workflow → name
|
||
/// args 无可读字段或工具未特化时，fallback 原 risk 模板（含风险等级提示）。
|
||
async fn build_approval_reason(
|
||
tool_name: &str,
|
||
args: &serde_json::Value,
|
||
risk_level: RiskLevel,
|
||
db: &Arc<Database>,
|
||
) -> String {
|
||
let s = |key: &str| args.get(key).and_then(|v| v.as_str()).unwrap_or("");
|
||
let detail = match tool_name {
|
||
"delete_project" => {
|
||
let id = s("id");
|
||
if !id.is_empty() { format!("删除项目{}", resolve_project_label(db, id).await) } else { String::new() }
|
||
}
|
||
"restore_project" => {
|
||
let id = s("id");
|
||
if !id.is_empty() { format!("从回收站恢复项目{}", resolve_project_label(db, id).await) } else { String::new() }
|
||
}
|
||
"purge_project" => {
|
||
let id = s("id");
|
||
if !id.is_empty() { format!("永久删除项目及关联数据，不可恢复{}", resolve_project_label(db, id).await) } else { String::new() }
|
||
}
|
||
"update_project" => {
|
||
let id = s("id");
|
||
let field = s("field");
|
||
if !id.is_empty() {
|
||
format!("修改项目{}字段「{}」", resolve_project_label(db, id).await, field)
|
||
} else if !field.is_empty() {
|
||
format!("修改项目字段「{}」", field)
|
||
} else { String::new() }
|
||
}
|
||
"bind_directory" => {
|
||
let id = s("id");
|
||
let path = s("path");
|
||
if !path.is_empty() && !id.is_empty() {
|
||
format!("绑定目录：{}（项目{}）", path, resolve_project_label(db, id).await)
|
||
} else if !path.is_empty() {
|
||
format!("绑定目录：{}", path)
|
||
} else { String::new() }
|
||
}
|
||
"create_task" => {
|
||
let title = s("title");
|
||
let pid = s("project_id");
|
||
if !title.is_empty() && !pid.is_empty() {
|
||
format!("创建任务：{}（项目{}）", title, resolve_project_label(db, pid).await)
|
||
} else if !title.is_empty() {
|
||
format!("创建任务：{}", title)
|
||
} else { String::new() }
|
||
}
|
||
"create_project" => {
|
||
let name = s("name");
|
||
if !name.is_empty() { format!("创建项目：「{}」", name) } else { String::new() }
|
||
}
|
||
"create_idea" => {
|
||
let title = s("title");
|
||
if !title.is_empty() { format!("捕获灵感：{}", title) } else { String::new() }
|
||
}
|
||
"run_workflow" => {
|
||
let name = s("name");
|
||
if !name.is_empty() { format!("运行工作流：{}", name) } else { String::new() }
|
||
}
|
||
_ => String::new(),
|
||
};
|
||
if detail.is_empty() {
|
||
// fallback：无可读字段，保留风险等级提示模板
|
||
match risk_level {
|
||
RiskLevel::High => "高风险操作，必须人工批准".to_string(),
|
||
_ => "创建操作，请确认是否执行".to_string(),
|
||
}
|
||
} else {
|
||
// 拼上风险等级后缀（高风险标注，便于用户权衡）
|
||
match risk_level {
|
||
RiskLevel::High => format!("{}（高风险，需人工批准）", detail),
|
||
_ => format!("{}（请确认是否执行）", detail),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 启动恢复:从审计表重建 pending_approvals(重启前未审批的工具调用,内存态已丢)
|
||
///
|
||
/// pending_approvals 是 AiSession 内存 HashMap,重启必丢。ai_tool_executions 表已存
|
||
/// status=pending 的行(持久化真相源),此处读回重建内存态,使重启后待审批不丢。
|
||
/// 前端经 ai_pending_tool_calls 查询 + switchConversation 恢复 toolCard 的 pending_approval 态。
|
||
pub async fn restore_pending_approvals(state: &AppState) {
|
||
let pending = match state.ai_tool_executions.list_pending().await {
|
||
Ok(v) => v,
|
||
Err(e) => {
|
||
tracing::warn!("启动恢复 pending 审批失败(非阻断): {}", e);
|
||
return;
|
||
}
|
||
};
|
||
if pending.is_empty() {
|
||
return;
|
||
}
|
||
let mut session = state.ai_session.lock().await;
|
||
for rec in pending {
|
||
let args: serde_json::Value = serde_json::from_str(&rec.arguments).unwrap_or_default();
|
||
let Some(risk) = risk_from_str(&rec.risk_level) else { continue };
|
||
session.pending_approvals.insert(
|
||
rec.tool_call_id.clone(),
|
||
PendingApproval {
|
||
tool_call_id: rec.tool_call_id,
|
||
tool_name: rec.tool_name,
|
||
arguments: args,
|
||
risk_level: risk,
|
||
conversation_id: rec.conversation_id,
|
||
recovered: true,
|
||
},
|
||
);
|
||
}
|
||
tracing::info!("启动恢复: {} 条 pending 工具审批重建到内存", session.pending_approvals.len());
|
||
}
|
||
|
||
/// 写一条工具执行审计记录（insert 失败不阻断主流程，故 `let _ =`）
|
||
///
|
||
/// `decided_by` 有值（auto/human）= 已决策执行 → 记 executed_at；
|
||
/// `None`（pending 待审批）→ executed_at 留空，待 audit_finalize 回填。
|
||
pub(crate) async fn audit_tool_call(
|
||
repo: &AiToolExecutionRepo,
|
||
conv_id: &str,
|
||
tool_call_id: &str,
|
||
tool_name: &str,
|
||
arguments: &str,
|
||
status: &str,
|
||
risk_level: RiskLevel,
|
||
result: Option<String>,
|
||
decided_by: Option<&str>,
|
||
) {
|
||
let executed_at = if decided_by.is_some() { Some(now_millis()) } else { None };
|
||
let _ = repo
|
||
.insert(AiToolExecutionRecord {
|
||
id: new_id(),
|
||
conversation_id: Some(conv_id.to_string()),
|
||
tool_call_id: tool_call_id.to_string(),
|
||
tool_name: tool_name.to_string(),
|
||
arguments: arguments.to_string(),
|
||
result,
|
||
status: status.to_string(),
|
||
risk_level: risk_str(risk_level).to_string(),
|
||
requested_at: now_millis(),
|
||
executed_at,
|
||
decided_by: decided_by.map(|s| s.to_string()),
|
||
})
|
||
.await;
|
||
}
|
||
|
||
/// 审批后更新审计记录状态（按 tool_call_id 定位 pending 记录，回填 status/decided_by=human/executed_at/result）
|
||
///
|
||
/// 走专用 find_by_tool_call_id —— 通用 query 宏硬编码 ORDER BY created_at DESC，
|
||
/// 而 ai_tool_executions 无该列，调用会报 "no such column: created_at" 被 unwrap_or_default 吞掉，
|
||
/// 导致审批后审计记录永久卡 pending。
|
||
pub(crate) async fn audit_finalize(state: &AppState, tool_call_id: &str, status: &str, result: Option<String>) {
|
||
let Some(mut rec) = state
|
||
.ai_tool_executions
|
||
.find_by_tool_call_id(tool_call_id)
|
||
.await
|
||
.unwrap_or_default()
|
||
else {
|
||
tracing::warn!("audit_finalize: 未找到 tool_call_id={} 的审计记录", tool_call_id);
|
||
return;
|
||
};
|
||
rec.status = status.to_string();
|
||
rec.decided_by = Some("human".to_string());
|
||
rec.executed_at = Some(now_millis());
|
||
if let Some(r) = result {
|
||
rec.result = Some(r);
|
||
}
|
||
let _ = state.ai_tool_executions.update_full(&rec).await;
|
||
}
|
||
|
||
/// 处理流式接收的工具调用：Low 风险并行执行（join_all），Med/High 进审批门控
|
||
/// 返回待审批的工具数量（0 = 全部自动执行完成）
|
||
pub(crate) async fn process_tool_calls(
|
||
session: &mut 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);
|
||
let mut pending_count = 0usize;
|
||
let audit_repo = AiToolExecutionRepo::new(db);
|
||
|
||
// 解析 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()));
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallStarted {
|
||
id: draft.id.clone(),
|
||
name: draft.name.clone(),
|
||
args: args.clone(),
|
||
conversation_id: Some(conv_id.to_string()),
|
||
});
|
||
(idx, draft, args)
|
||
})
|
||
.collect();
|
||
|
||
// 分类：Low 收集并行执行，Med/High 立即进审批门控（push 占位 tool_result）
|
||
let mut low_risk: Vec<(ToolCallDraft, serde_json::Value)> = Vec::new();
|
||
for (_, draft, args) in drafts {
|
||
let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High);
|
||
match risk_level {
|
||
RiskLevel::Low => low_risk.push((draft, args)),
|
||
RiskLevel::Medium | RiskLevel::High => {
|
||
pending_count += 1;
|
||
session.pending_approvals.insert(draft.id.clone(), PendingApproval {
|
||
tool_call_id: draft.id.clone(),
|
||
tool_name: draft.name.clone(),
|
||
arguments: args.clone(),
|
||
risk_level,
|
||
conversation_id: Some(conv_id.to_string()),
|
||
recovered: false,
|
||
});
|
||
session.messages.push(ChatMessage::tool_result(&draft.id, "需要用户审批，等待确认"));
|
||
let reason = build_approval_reason(&draft.name, &args, risk_level, db).await;
|
||
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiApprovalRequired {
|
||
id: draft.id.clone(),
|
||
name: draft.name.clone(),
|
||
args: args.clone(),
|
||
reason,
|
||
conversation_id: Some(conv_id.to_string()),
|
||
});
|
||
audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, "pending", risk_level, None, None).await;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Low 风险并行执行：execute + 即时 emit 在闭包内（不持 session 锁），
|
||
// push tool_result / audit 在 join_all 后串行回填（持锁，与 Med/High 占位拼接）。
|
||
// join_all 保序——结果顺序 = low_risk 输入顺序 = tc_list 原始 index 顺序，不额外 sort
|
||
if !low_risk.is_empty() {
|
||
let results: Vec<(ToolCallDraft, Result<String, String>)> =
|
||
futures::future::join_all(low_risk.into_iter().map(|(draft, args)| {
|
||
let tools = tools_arc.clone();
|
||
let app_clone = app_handle.clone();
|
||
let conv_clone = conv_id.to_string();
|
||
async move {
|
||
let result = tools.execute(&draft.name, args).await;
|
||
match result {
|
||
Ok(val) => {
|
||
let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted {
|
||
id: draft.id.clone(),
|
||
result: val.clone(),
|
||
conversation_id: Some(conv_clone),
|
||
});
|
||
(draft, Ok(val.to_string()))
|
||
}
|
||
Err(e) => {
|
||
// AR-6（定向B）：Low 工具失败不 emit AiError。
|
||
// 原逻辑 emit AiError 会令前端置 streaming=false + 错误气泡，但 process_tool_calls
|
||
// 仍返回 pending_count=0，agentic loop 续下一轮 → 前端 false 后端跑，状态紊乱。
|
||
// 现改为正常 emit AiToolCallCompleted（result=错误信息），错误包进 tool_result
|
||
// 让 LLM 看到工具失败自行决定下一步，loop 正常续，前端状态一致。
|
||
let err_msg = format!("工具 {} 执行失败: {}", draft.name, e);
|
||
let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted {
|
||
id: draft.id.clone(),
|
||
result: serde_json::Value::String(err_msg.clone()),
|
||
conversation_id: Some(conv_clone),
|
||
});
|
||
(draft, Err(err_msg))
|
||
}
|
||
}
|
||
}
|
||
})).await;
|
||
|
||
// 串行回填 tool_result + 审计（持 session 锁）
|
||
for (draft, outcome) in results {
|
||
let (status, content) = match outcome {
|
||
Ok(c) => ("completed", c),
|
||
Err(c) => ("failed", c),
|
||
};
|
||
session.messages.push(ChatMessage::tool_result(&draft.id, content.clone()));
|
||
audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, status, RiskLevel::Low, Some(content), Some("auto")).await;
|
||
}
|
||
}
|
||
|
||
pending_count
|
||
}
|