From 55bc1e1b232f276161aa44687584a93b13d781ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Fri, 19 Jun 2026 04:31:19 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84:=20audit=E7=AC=AC=E4=B8=89?= =?UTF-8?q?=E6=89=B9cache=20+=20anthropic=5Fcompat=E6=8B=86=E5=88=86(strat?= =?UTF-8?q?egy=E5=B9=B6=E8=A1=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - audit 第三批: 新建 audit/cache.rs(151行 find_cached_high_risk_result/canonical_args_key/sort_object_keys/PENDING_APPROVAL_PLACEHOLDER), audit/mod.rs 586→451(-135); pub(crate) re-export, F-09 per_conv保留 - anthropic_compat 拆分: 新建 df-ai/anthropic_helpers.rs(223行 apply_anthropic_event + 5 struct + 2常量), anthropic_compat.rs 1036→842; Provider impl保留(impl块约束), 字段pub(crate) 主代兜底: cargo check --workspace 0 + test df-ai 119 + test devflow 98 strategy: 核心库自底向上, 单批1-2文件原子, impl块约束(Provider方法保留只抽纯函数/类型) git add指定(audit/* + df-ai/*, 不含其他会话) --- crates/df-ai/src/anthropic_compat.rs | 216 ++-------------------- crates/df-ai/src/anthropic_helpers.rs | 223 +++++++++++++++++++++++ crates/df-ai/src/lib.rs | 1 + src-tauri/src/commands/ai/audit/cache.rs | 151 +++++++++++++++ src-tauri/src/commands/ai/audit/mod.rs | 145 +-------------- 5 files changed, 391 insertions(+), 345 deletions(-) create mode 100644 crates/df-ai/src/anthropic_helpers.rs create mode 100644 src-tauri/src/commands/ai/audit/cache.rs diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index 3ca3467..f70d686 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -5,18 +5,20 @@ //! //! 与 OpenAI 协议的关键差异由本模块内部完成转换，对外仍暴露统一的 LlmProvider trait， //! 上层 (Agentic Loop / AiNode) 无需感知协议。 +//! +//! 协议数据结构（请求/响应 struct）与 SSE 事件解析纯函数已抽至 `anthropic_helpers`， +//! 本模块仅保留 Provider struct + impl（HTTP 调用），Rust impl 块不可跨文件故作此切分。 use async_trait::async_trait; use eventsource_stream::Eventsource; use futures::StreamExt; use reqwest::Client; -use serde::{Deserialize, Serialize}; use std::time::Duration; use tracing::{debug, error, warn}; use crate::provider::{ CompletionRequest, CompletionResponse, LlmProvider, MessageRole, - StreamChunk, StreamResult, TokenUsage, ToolCall, ToolCallDelta, + StreamResult, TokenUsage, ToolCall, }; // ChatMessage 仅单测构造 CompletionRequest 用，避免非 test 构建的 unused import 警告。 #[cfg(test)] @@ -25,204 +27,12 @@ use crate::retry::{ retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable, }; -// ============================================================ -// Anthropic API 请求/响应结构体 -// ============================================================ - -/// Anthropic 请求体 -#[derive(Debug, Clone, Serialize)] -struct AnthropicRequest { - model: String, - messages: Vec, - max_tokens: u32, - #[serde(skip_serializing_if = "Option::is_none")] - system: Option, - #[serde(skip_serializing_if = "Option::is_none")] - temperature: Option, - stream: bool, - #[serde(skip_serializing_if = "Option::is_none")] - tools: Option>, - #[serde(skip_serializing_if = "Option::is_none")] - tool_choice: Option, -} - -/// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters） -#[derive(Debug, Clone, Serialize)] -struct AnthropicToolDef { - name: String, - #[serde(skip_serializing_if = "Option::is_none")] - description: Option, - input_schema: serde_json::Value, -} - -/// Anthropic 同步响应 -#[derive(Debug, Deserialize)] -struct AnthropicResponse { - // SW-260618-25: Anthropic 响应反序列化字段,保留以对齐响应结构(响应 id),标注意图消除 dead_code warning - #[allow(dead_code)] - id: String, - model: String, - content: Vec, - // SW-260618-25: Anthropic 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning - #[allow(dead_code)] - stop_reason: Option, - usage: AnthropicUsage, -} - -/// 响应 content 块（text 或 tool_use） -#[derive(Debug, Deserialize)] -struct AnthropicContentBlock { - #[serde(rename = "type")] - block_type: String, - #[serde(default)] - text: Option, - /// tool_use 块字段 - id: Option, - name: Option, - input: Option, -} - -#[derive(Debug, Deserialize)] -struct AnthropicUsage { - input_tokens: u32, - output_tokens: u32, -} - -// ============================================================ -// SSE 解析纯函数（与 HTTP 解耦，便于单测） -// ============================================================ - -/// 将一条 Anthropic Messages SSE 事件 data 解析为 StreamChunk，并按需更新 usage 累加器。 -/// -/// 按 `type` 字段分发： -/// - `message_start` → 用 `message.usage.input_tokens` 初始化累加器（output 置 0）。 -/// - `message_delta` → **output_tokens 是累计值（非增量）**，直接覆盖 `completion_tokens` 并重算 `total`。 -/// - `content_block_delta` (text_delta/input_json_delta) → 文本/工具入参增量。 -/// - `content_block_start` (tool_use) → 工具块开始，带 id+name。 -/// - `message_stop` → 返回 `finished=true` 终态 chunk，`usage` 取自累加器（`take()`）。 -/// - `error` → 返回 `finished=true` 终态空 chunk。 -/// - 其它（content_block_stop / ping 等）→ 空 chunk。 -/// -/// 等价性：content_block / message_stop / error 等事件分支与原 stream() 闭包逐字一致； -/// usage 透传（message_stop 终态 take() 带出、message_delta 的 output_tokens 按累计值覆盖） -/// 为本次新增能力，对应 StreamChunk 新增的 usage 字段。 -pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option) -> StreamChunk { - // 解析 data 中的 JSON，按 type 字段决定如何转 StreamChunk - let v: serde_json::Value = match serde_json::from_str(data) { - Ok(v) => v, - Err(_) => { - return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } - } - }; - let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or(""); - match ty { - // 消息开始：取 input_tokens 初始化累积器（output 此时未知，置 0） - "message_start" => { - if let Some(inp) = v - .get("message") - .and_then(|m| m.get("usage")) - .and_then(|u| u.get("input_tokens")) - .and_then(|t| t.as_u64()) - { - *usage_accum = Some(TokenUsage { - prompt_tokens: inp as u32, - completion_tokens: 0, - total_tokens: inp as u32, - }); - } - StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } - } - // 消息增量：output_tokens 是累计值（非增量），直接覆盖 completion + 重算 total - "message_delta" => { - if let Some(out) = v.get("usage").and_then(|u| u.get("output_tokens")).and_then(|t| t.as_u64()) { - let acc = usage_accum - .get_or_insert(TokenUsage { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }); - acc.completion_tokens = out as u32; - acc.total_tokens = acc.prompt_tokens + acc.completion_tokens; - } - StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } - } - // 文本增量 - "content_block_delta" => { - if let Some(delta) = v.get("delta") { - if delta.get("type").and_then(|t| t.as_str()) == Some("text_delta") { - let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("").to_string(); - return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }; - } - // 工具入参增量 - if delta.get("type").and_then(|t| t.as_str()) == Some("input_json_delta") { - let partial = delta.get("partial_json").and_then(|t| t.as_str()).unwrap_or("").to_string(); - let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32; - return StreamChunk { - delta: String::new(), - finished: false, - tool_calls: Some(vec![ToolCallDelta { - index: idx, - id: None, - function_name: None, - function_arguments: Some(partial), - }]), - usage: None, - error: None, - reasoning_content: None, - }; - } - } - StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } - } - // 工具块开始：带 id + name - "content_block_start" => { - if let Some(cb) = v.get("content_block") { - if cb.get("type").and_then(|t| t.as_str()) == Some("tool_use") { - let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32; - let name = cb.get("name").and_then(|t| t.as_str()).map(|s| s.to_string()); - // id 缺失时用占位 id 兜底：流式后续 input_json_delta 按 index 累加， - // 中途无法整体跳过；占位 id 保证回传的 tool_use_id 非空，避免 GLM 500。 - let id = match cb.get("id").and_then(|t| t.as_str()).map(|s| s.to_string()) { - Some(id) if !id.is_empty() => Some(id), - _ => { - let placeholder = format!("tool_missing_{}", idx); - warn!(%placeholder, name = ?name, "Anthropic 流式 tool_use 块缺少 id，已填占位 id（原样回传会触发 GLM 500）"); - Some(placeholder) - } - }; - return StreamChunk { - delta: String::new(), - finished: false, - tool_calls: Some(vec![ToolCallDelta { - index: idx, - id, - function_name: name, - function_arguments: None, - }]), - usage: None, - error: None, - reasoning_content: None, - }; - } - } - StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } - } - // 消息结束：带出累积 usage - "message_stop" => StreamChunk { - delta: String::new(), - finished: true, - tool_calls: None, - usage: usage_accum.take(), - error: None, - reasoning_content: None, - }, - // 错误事件：流中途出错。不走 finished 完成路径（避免残缺响应被当正常完成入库）， - // 改由 stream_llm 识别 error 非空 → 发 AiError + 丢弃残缺（与 OpenAI 路径 Err 一致）。 - "error" => { - let msg = v.get("error").and_then(|e| e.get("message")).and_then(|m| m.as_str()).unwrap_or("stream error").to_string(); - error!(%msg, "Anthropic 流式错误事件"); - StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None } - } - // content_block_stop / ping 等不产出 chunk - _ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }, - } -} +// 协议数据结构 + SSE 纯解析（apply_anthropic_event / AnthropicRequest 等）抽至 anthropic_helpers， +// 此处 use 引入以保持本模块内引用路径不变（零行为变更搬迁）。 +use crate::anthropic_helpers::{ + apply_anthropic_event, AnthropicRequest, AnthropicResponse, AnthropicToolDef, + ANTHROPIC_VERSION, DEFAULT_MAX_TOKENS, +}; // ============================================================ // Provider 实现 @@ -236,11 +46,6 @@ pub struct AnthropicCompatProvider { default_model: String, } -/// Anthropic 流式协议版本头 -const ANTHROPIC_VERSION: &str = "2023-06-01"; -/// Anthropic max_tokens 必填，缺省时的兜底值 -const DEFAULT_MAX_TOKENS: u32 = 4096; - impl AnthropicCompatProvider { /// 创建 Provider /// @@ -788,6 +593,7 @@ impl LlmProvider for AnthropicCompatProvider { #[cfg(test)] mod tests { use super::*; + // apply_anthropic_event / TokenUsage 经 super::*（含 anthropic_helpers::apply_anthropic_event 的 use）可见。 /// 辅助：构造 message_start 事件 fn message_start(input_tokens: u32) -> String { diff --git a/crates/df-ai/src/anthropic_helpers.rs b/crates/df-ai/src/anthropic_helpers.rs new file mode 100644 index 0000000..d00299f --- /dev/null +++ b/crates/df-ai/src/anthropic_helpers.rs @@ -0,0 +1,223 @@ +//! Anthropic 协议适配 — 协议数据结构与 SSE 事件解析（纯函数）。 +//! +//! 本模块从 `anthropic_compat.rs` 抽离，承载与 HTTP 无关的纯协议逻辑： +//! - 请求/响应结构体（`AnthropicRequest` 等） +//! - 协议常量（`ANTHROPIC_VERSION` / `DEFAULT_MAX_TOKENS`） +//! - SSE 事件 → `StreamChunk` 转换纯函数（`apply_anthropic_event`） +//! +//! Provider struct + impl（含 HTTP 调用）仍留在 `anthropic_compat.rs`， +//! Rust impl 块不可跨文件，故仅搬迁 impl 块外部的类型/常量/纯函数。 +//! 零行为变更（纯搬迁）。 + +use serde::{Deserialize, Serialize}; +use tracing::{error, warn}; + +use crate::provider::{StreamChunk, TokenUsage, ToolCallDelta}; + +// ============================================================ +// Anthropic API 请求/响应结构体 +// ============================================================ + +/// Anthropic 请求体 +#[derive(Debug, Clone, Serialize)] +pub(crate) struct AnthropicRequest { + pub model: String, + pub messages: Vec, + pub max_tokens: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub system: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub temperature: Option, + pub stream: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_choice: Option, +} + +/// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters） +#[derive(Debug, Clone, Serialize)] +pub(crate) struct AnthropicToolDef { + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub description: Option, + pub input_schema: serde_json::Value, +} + +/// Anthropic 同步响应 +#[derive(Debug, Deserialize)] +pub(crate) struct AnthropicResponse { + // SW-260618-25: Anthropic 响应反序列化字段,保留以对齐响应结构(响应 id),标注意图消除 dead_code warning + #[allow(dead_code)] + id: String, + pub model: String, + pub content: Vec, + // SW-260618-25: Anthropic 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning + #[allow(dead_code)] + stop_reason: Option, + pub usage: AnthropicUsage, +} + +/// 响应 content 块（text 或 tool_use） +#[derive(Debug, Deserialize)] +pub(crate) struct AnthropicContentBlock { + #[serde(rename = "type")] + pub block_type: String, + #[serde(default)] + pub text: Option, + /// tool_use 块字段 + pub id: Option, + pub name: Option, + pub input: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct AnthropicUsage { + pub input_tokens: u32, + pub output_tokens: u32, +} + +// ============================================================ +// 协议常量 +// ============================================================ + +/// Anthropic 流式协议版本头 +pub(crate) const ANTHROPIC_VERSION: &str = "2023-06-01"; +/// Anthropic max_tokens 必填，缺省时的兜底值 +pub(crate) const DEFAULT_MAX_TOKENS: u32 = 4096; + +// ============================================================ +// SSE 解析纯函数（与 HTTP 解耦，便于单测） +// ============================================================ + +/// 将一条 Anthropic Messages SSE 事件 data 解析为 StreamChunk，并按需更新 usage 累加器。 +/// +/// 按 `type` 字段分发： +/// - `message_start` → 用 `message.usage.input_tokens` 初始化累加器（output 置 0）。 +/// - `message_delta` → **output_tokens 是累计值（非增量）**，直接覆盖 `completion_tokens` 并重算 `total`。 +/// - `content_block_delta` (text_delta/input_json_delta) → 文本/工具入参增量。 +/// - `content_block_start` (tool_use) → 工具块开始，带 id+name。 +/// - `message_stop` → 返回 `finished=true` 终态 chunk，`usage` 取自累加器（`take()`）。 +/// - `error` → 返回 `finished=true` 终态空 chunk。 +/// - 其它（content_block_stop / ping 等）→ 空 chunk。 +/// +/// 等价性：content_block / message_stop / error 等事件分支与原 stream() 闭包逐字一致； +/// usage 透传（message_stop 终态 take() 带出、message_delta 的 output_tokens 按累计值覆盖） +/// 为本次新增能力，对应 StreamChunk 新增的 usage 字段。 +pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option) -> StreamChunk { + // 解析 data 中的 JSON，按 type 字段决定如何转 StreamChunk + let v: serde_json::Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(_) => { + return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } + } + }; + let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or(""); + match ty { + // 消息开始：取 input_tokens 初始化累积器（output 此时未知，置 0） + "message_start" => { + if let Some(inp) = v + .get("message") + .and_then(|m| m.get("usage")) + .and_then(|u| u.get("input_tokens")) + .and_then(|t| t.as_u64()) + { + *usage_accum = Some(TokenUsage { + prompt_tokens: inp as u32, + completion_tokens: 0, + total_tokens: inp as u32, + }); + } + StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } + } + // 消息增量：output_tokens 是累计值（非增量），直接覆盖 completion + 重算 total + "message_delta" => { + if let Some(out) = v.get("usage").and_then(|u| u.get("output_tokens")).and_then(|t| t.as_u64()) { + let acc = usage_accum + .get_or_insert(TokenUsage { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }); + acc.completion_tokens = out as u32; + acc.total_tokens = acc.prompt_tokens + acc.completion_tokens; + } + StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } + } + // 文本增量 + "content_block_delta" => { + if let Some(delta) = v.get("delta") { + if delta.get("type").and_then(|t| t.as_str()) == Some("text_delta") { + let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("").to_string(); + return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }; + } + // 工具入参增量 + if delta.get("type").and_then(|t| t.as_str()) == Some("input_json_delta") { + let partial = delta.get("partial_json").and_then(|t| t.as_str()).unwrap_or("").to_string(); + let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32; + return StreamChunk { + delta: String::new(), + finished: false, + tool_calls: Some(vec![ToolCallDelta { + index: idx, + id: None, + function_name: None, + function_arguments: Some(partial), + }]), + usage: None, + error: None, + reasoning_content: None, + }; + } + } + StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } + } + // 工具块开始：带 id + name + "content_block_start" => { + if let Some(cb) = v.get("content_block") { + if cb.get("type").and_then(|t| t.as_str()) == Some("tool_use") { + let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32; + let name = cb.get("name").and_then(|t| t.as_str()).map(|s| s.to_string()); + // id 缺失时用占位 id 兜底：流式后续 input_json_delta 按 index 累加， + // 中途无法整体跳过；占位 id 保证回传的 tool_use_id 非空，避免 GLM 500。 + let id = match cb.get("id").and_then(|t| t.as_str()).map(|s| s.to_string()) { + Some(id) if !id.is_empty() => Some(id), + _ => { + let placeholder = format!("tool_missing_{}", idx); + warn!(%placeholder, name = ?name, "Anthropic 流式 tool_use 块缺少 id，已填占位 id（原样回传会触发 GLM 500）"); + Some(placeholder) + } + }; + return StreamChunk { + delta: String::new(), + finished: false, + tool_calls: Some(vec![ToolCallDelta { + index: idx, + id, + function_name: name, + function_arguments: None, + }]), + usage: None, + error: None, + reasoning_content: None, + }; + } + } + StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None } + } + // 消息结束：带出累积 usage + "message_stop" => StreamChunk { + delta: String::new(), + finished: true, + tool_calls: None, + usage: usage_accum.take(), + error: None, + reasoning_content: None, + }, + // 错误事件：流中途出错。不走 finished 完成路径（避免残缺响应被当正常完成入库）， + // 改由 stream_llm 识别 error 非空 → 发 AiError + 丢弃残缺（与 OpenAI 路径 Err 一致）。 + "error" => { + let msg = v.get("error").and_then(|e| e.get("message")).and_then(|m| m.as_str()).unwrap_or("stream error").to_string(); + error!(%msg, "Anthropic 流式错误事件"); + StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None } + } + // content_block_stop / ping 等不产出 chunk + _ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }, + } +} diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index 37b7ce4..4dd3795 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -2,6 +2,7 @@ pub mod ai_tools; pub mod anthropic_compat; +pub mod anthropic_helpers; pub mod context; pub mod context_helpers; pub mod coordinator; diff --git a/src-tauri/src/commands/ai/audit/cache.rs b/src-tauri/src/commands/ai/audit/cache.rs new file mode 100644 index 0000000..da92c87 --- /dev/null +++ b/src-tauri/src/commands/ai/audit/cache.rs @@ -0,0 +1,151 @@ +//! F-260616-05 高危工具去重缓存(第三批 helper 抽离,行为零变更)。 +//! +//! 从 audit/mod.rs 搬迁:`find_cached_high_risk_result` + `canonical_args_key` +//! + `sort_object_keys` + `PENDING_APPROVAL_PLACEHOLDER`(去重与 process_tool_calls +//! 共用占位常量)。三函数内部闭环(find_cached 调 canonical→sort),无外部 helper 依赖。 + +use df_ai::provider::ChatMessage; +use df_storage::crud::AiToolExecutionRepo; + +use super::super::AiSession; + +/// Med/High 工具挂起审批时回填的占位 tool_result 内容。 +/// +/// 单一常量避免「写入处」(process_tool_calls) 与「去重排查处」 +/// (find_cached_high_risk_result) 各持一份字面量致耦合——若两者漂移, +/// 去重会把 pending 占位误判为已落定结果命中缓存,污染 LLM 上下文。 +pub(crate) const PENDING_APPROVAL_PLACEHOLDER: &str = "需要用户审批，等待确认"; + +/// F-260616-05：高危工具去重（根治 run_command 超时→重试→重新审批循环）。 +/// +/// ⚠ 性能注记(CR-260618-11#5):本函数在 session lock 持有期间对每个 high risk 工具 +/// 串行查 AiToolExecutionRepo::find_by_tool_call_id,工具数多时锁持有线性增长。 +/// 批量预取(改签名传 tool_call_ids 批量查)属架构改暂未做,后续 High risk 工具增多时优先处理。 +/// +/// **根因链**（F-04 batch42 已做超时标注）：LLM 调 run_command 超时 → F-04 把超时标注成 +/// tool_result 回传 LLM → LLM（不可靠）仍重试同命令 → 新 tool_call_id（provider 每轮新 id） +/// → `process_tool_calls` 重新 insert pending(audit.rs:418) → 用户被迫重新审批，循环。 +/// +/// **治本（F-05）**：即使 LLM 仍重试，同 (tool_name, args) 不重复审批。本函数扫描会话历史， +/// 若发现**已落定**（executed/failed/rejected，非 pending）的同类同参高危调用，返回其 +/// tool_result 内容，调用方把缓存结果作为新 tool_call_id 的 tool_result 回传 LLM，跳过审批。 +/// +/// **安全边界**： +/// - 仅 High risk（循环源头）；Low/Med 不去重（Low 直行无审批，Med 重试场景少且去重易误伤）。 +/// - 仅匹配**已落定**结果（pending 的不命中——pending 已有独立流程，不会触发循环； +/// 且避免两个 pending 互相吞掉审批）。LLM 只有在收到 tool_result 后才会重试， +/// 故循环必然是「上一条已落定 → 重试」形态，pending 不命中不影响治本。 +/// - args 走 JSON 规范化比较（键序无关），避免 LLM 两次生成键序不同误判为不同命令。 +/// - 返回 None 表示无缓存命中（走原审批流程）。 +/// +/// `session` 只读扫描 messages（不写），调用方据返回值决定是否跳过 insert pending。 +/// +/// F-260616-09 B 批2:经`conv_id` 索引 per_conv.messages(顶层 messages 批2 后是死字段)。 +/// conv_id 来源:process_tool_calls 入参 → 由 agentic.rs run_agentic_loop 入参透传。 +pub(crate) async fn find_cached_high_risk_result( + session: &AiSession, + conv_id: &str, + audit_repo: &AiToolExecutionRepo, + tool_name: &str, + args: &serde_json::Value, +) -> Option<(String, String)> { + use df_ai::provider::MessageRole; + + // 规范化新调用的 args 为可比字符串（排序键，键序无关） + let new_args_key = canonical_args_key(args); + + // F-260616-09 B 批2:读 per_conv.messages。process_tool_calls 调用前 loop 入口已桥接建立 per_conv, + // 故 conv_read 必命中;防御性 None 时返 None(无缓存命中,走原审批流程)。 + let conv = session.conv_read(conv_id)?; + // ContextManager::iter 返回 impl Iterator（非 DoubleEnded），collect 成 Vec 再反向遍历。 + // 单对话消息量小（百级），collect 开销可忽略。 + let msgs: Vec<&ChatMessage> = conv.messages.iter().collect(); + + // 1) 反向扫描 assistant tool_calls，找最近一条同名同参的 High 工具调用 → 拿到旧 tool_call_id + // 反向：循环是「最近一次超时→重试」，命中通常是末尾附近，反向先停省全扫。 + let mut prev_tool_call_id: Option = None; + for msg in msgs.iter().rev() { + if !matches!(msg.role, MessageRole::Assistant) { + continue; + } + let Some(tcs) = msg.tool_calls.as_ref() else { continue }; + for tc in tcs { + if tc.function.name != tool_name { + continue; + } + // 旧调用的 args 是流式拼接的 JSON 字符串，解析失败跳过（不误判为命中） + let Ok(old_args) = serde_json::from_str::(&tc.function.arguments) else { + continue; + }; + if canonical_args_key(&old_args) == new_args_key { + prev_tool_call_id = Some(tc.id.clone()); + break; + } + } + if prev_tool_call_id.is_some() { + break; + } + } + + // 2) 用旧 tool_call_id 找对应 tool_result。注意：审批拒绝/超时失败也属「已落定」， + // 其 tool_result 内容同样回传（LLM 看到原反馈自行决定，不再逼用户二次审批）。 + // pending 占位（「需要用户审批，等待确认」）不命中——仍在审批中，走原流程。 + let old_id = prev_tool_call_id?; + for msg in msgs.iter().rev() { + if !matches!(msg.role, MessageRole::Tool) { + continue; + } + if msg.tool_call_id.as_deref() != Some(old_id.as_str()) { + continue; + } + // 命中旧 tool_result：排除 pending 占位（内容固定为 PENDING_APPROVAL_PLACEHOLDER） + if msg.content == PENDING_APPROVAL_PLACEHOLDER { + return None; + } + // SW-260618-16: 查审计表拿缓存来源真实 status(completed/rejected/failed),透传给 + // audit_tool_call 而非固定 completed(审计语义与结果内容一致,防"rejected/failed 结果 + // 记 completed"误导安全追溯)。审计记录缺失/查询失败 fallback completed(不阻塞去重,降级原行为)。 + let status = audit_repo + .find_by_tool_call_id(old_id.as_str()) + .await + .ok() + .and_then(|opt| opt.map(|rec| rec.status)) + .unwrap_or_else(|| "completed".to_string()); + return Some((msg.content.clone(), status)); + } + None +} + +/// 把 JSON args 规范化为可比字符串：对象键按字典序排序后序列化， +/// 键序不同的等价参数生成同一 key（防 LLM 两次生成键序不同误判为不同命令）。 +fn canonical_args_key(args: &serde_json::Value) -> String { + let mut v = args.clone(); + sort_object_keys(&mut v); + // 紧凑序列化（无空白），保证稳定可比 + serde_json::to_string(&v).unwrap_or_default() +} + +/// 递归对 JSON 对象的键做字典序排序（就地），数组成员也递归排序。 +fn sort_object_keys(v: &mut serde_json::Value) { + match v { + serde_json::Value::Object(map) => { + // BTreeMap 按键排序，重建 Object + let mut entries: Vec<(String, serde_json::Value)> = map + .iter() + .map(|(k, val)| (k.clone(), val.clone())) + .collect(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + map.clear(); + for (k, mut val) in entries { + sort_object_keys(&mut val); + map.insert(k, val); + } + } + serde_json::Value::Array(arr) => { + for item in arr { + sort_object_keys(item); + } + } + _ => {} + } +} diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 6cff631..f67cbaa 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -18,13 +18,6 @@ use crate::commands::err_str; use super::{AiChatEvent, AiSession, PendingApproval, ToolCallDraft}; use super::tool_registry::generate_diff; -/// Med/High 工具挂起审批时回填的占位 tool_result 内容。 -/// -/// 单一常量避免「写入处」(process_tool_calls) 与「去重排查处」 -/// (find_cached_high_risk_result) 各持一份字面量致耦合——若两者漂移, -/// 去重会把 pending 占位误判为已落定结果命中缓存,污染 LLM 上下文。 -const PENDING_APPROVAL_PLACEHOLDER: &str = "需要用户审批，等待确认"; - /// RiskLevel → 审计记录字符串（low/medium/high） pub(crate) fn risk_str(r: RiskLevel) -> &'static str { match r { @@ -177,139 +170,11 @@ pub(crate) fn emit_data_changed(app_handle: &AppHandle, tool_name: &str) { } } -/// F-260616-05：高危工具去重（根治 run_command 超时→重试→重新审批循环）。 -/// -/// ⚠ 性能注记(CR-260618-11#5):本函数在 session lock 持有期间对每个 high risk 工具 -/// 串行查 AiToolExecutionRepo::find_by_tool_call_id,工具数多时锁持有线性增长。 -/// 批量预取(改签名传 tool_call_ids 批量查)属架构改暂未做,后续 High risk 工具增多时优先处理。 -/// -/// **根因链**（F-04 batch42 已做超时标注）：LLM 调 run_command 超时 → F-04 把超时标注成 -/// tool_result 回传 LLM → LLM（不可靠）仍重试同命令 → 新 tool_call_id（provider 每轮新 id） -/// → `process_tool_calls` 重新 insert pending(audit.rs:418) → 用户被迫重新审批，循环。 -/// -/// **治本（F-05）**：即使 LLM 仍重试，同 (tool_name, args) 不重复审批。本函数扫描会话历史， -/// 若发现**已落定**（executed/failed/rejected，非 pending）的同类同参高危调用，返回其 -/// tool_result 内容，调用方把缓存结果作为新 tool_call_id 的 tool_result 回传 LLM，跳过审批。 -/// -/// **安全边界**： -/// - 仅 High risk（循环源头）；Low/Med 不去重（Low 直行无审批，Med 重试场景少且去重易误伤）。 -/// - 仅匹配**已落定**结果（pending 的不命中——pending 已有独立流程，不会触发循环； -/// 且避免两个 pending 互相吞掉审批）。LLM 只有在收到 tool_result 后才会重试， -/// 故循环必然是「上一条已落定 → 重试」形态，pending 不命中不影响治本。 -/// - args 走 JSON 规范化比较（键序无关），避免 LLM 两次生成键序不同误判为不同命令。 -/// - 返回 None 表示无缓存命中（走原审批流程）。 -/// -/// `session` 只读扫描 messages（不写），调用方据返回值决定是否跳过 insert pending。 -/// -/// F-260616-09 B 批2:经 `conv_id` 索引 per_conv.messages(顶层 messages 批2 后是死字段)。 -/// conv_id 来源:process_tool_calls 入参 → 由 agentic.rs run_agentic_loop 入参透传。 -async fn find_cached_high_risk_result( - session: &AiSession, - conv_id: &str, - audit_repo: &AiToolExecutionRepo, - tool_name: &str, - args: &serde_json::Value, -) -> Option<(String, String)> { - use df_ai::provider::MessageRole; - - // 规范化新调用的 args 为可比字符串（排序键，键序无关） - let new_args_key = canonical_args_key(args); - - // F-260616-09 B 批2:读 per_conv.messages。process_tool_calls 调用前 loop 入口已桥接建立 per_conv, - // 故 conv_read 必命中;防御性 None 时返 None(无缓存命中,走原审批流程)。 - let conv = session.conv_read(conv_id)?; - // ContextManager::iter 返回 impl Iterator（非 DoubleEnded），collect 成 Vec 再反向遍历。 - // 单对话消息量小（百级），collect 开销可忽略。 - let msgs: Vec<&ChatMessage> = conv.messages.iter().collect(); - - // 1) 反向扫描 assistant tool_calls，找最近一条同名同参的 High 工具调用 → 拿到旧 tool_call_id - // 反向：循环是「最近一次超时→重试」，命中通常是末尾附近，反向先停省全扫。 - let mut prev_tool_call_id: Option = None; - for msg in msgs.iter().rev() { - if !matches!(msg.role, MessageRole::Assistant) { - continue; - } - let Some(tcs) = msg.tool_calls.as_ref() else { continue }; - for tc in tcs { - if tc.function.name != tool_name { - continue; - } - // 旧调用的 args 是流式拼接的 JSON 字符串，解析失败跳过（不误判为命中） - let Ok(old_args) = serde_json::from_str::(&tc.function.arguments) else { - continue; - }; - if canonical_args_key(&old_args) == new_args_key { - prev_tool_call_id = Some(tc.id.clone()); - break; - } - } - if prev_tool_call_id.is_some() { - break; - } - } - - // 2) 用旧 tool_call_id 找对应 tool_result。注意：审批拒绝/超时失败也属「已落定」， - // 其 tool_result 内容同样回传（LLM 看到原反馈自行决定，不再逼用户二次审批）。 - // pending 占位（「需要用户审批，等待确认」）不命中——仍在审批中，走原流程。 - let old_id = prev_tool_call_id?; - for msg in msgs.iter().rev() { - if !matches!(msg.role, MessageRole::Tool) { - continue; - } - if msg.tool_call_id.as_deref() != Some(old_id.as_str()) { - continue; - } - // 命中旧 tool_result：排除 pending 占位（内容固定为 PENDING_APPROVAL_PLACEHOLDER） - if msg.content == PENDING_APPROVAL_PLACEHOLDER { - return None; - } - // SW-260618-16: 查审计表拿缓存来源真实 status(completed/rejected/failed),透传给 - // audit_tool_call 而非固定 completed(审计语义与结果内容一致,防"rejected/failed 结果 - // 记 completed"误导安全追溯)。审计记录缺失/查询失败 fallback completed(不阻塞去重,降级原行为)。 - let status = audit_repo - .find_by_tool_call_id(old_id.as_str()) - .await - .ok() - .and_then(|opt| opt.map(|rec| rec.status)) - .unwrap_or_else(|| "completed".to_string()); - return Some((msg.content.clone(), status)); - } - None -} - -/// 把 JSON args 规范化为可比字符串：对象键按字典序排序后序列化， -/// 键序不同的等价参数生成同一 key（防 LLM 两次生成键序不同误判为不同命令）。 -fn canonical_args_key(args: &serde_json::Value) -> String { - let mut v = args.clone(); - sort_object_keys(&mut v); - // 紧凑序列化（无空白），保证稳定可比 - serde_json::to_string(&v).unwrap_or_default() -} - -/// 递归对 JSON 对象的键做字典序排序（就地），数组成员也递归排序。 -fn sort_object_keys(v: &mut serde_json::Value) { - match v { - serde_json::Value::Object(map) => { - // BTreeMap 按键排序，重建 Object - let mut entries: Vec<(String, serde_json::Value)> = map - .iter() - .map(|(k, val)| (k.clone(), val.clone())) - .collect(); - entries.sort_by(|a, b| a.0.cmp(&b.0)); - map.clear(); - for (k, mut val) in entries { - sort_object_keys(&mut val); - map.insert(k, val); - } - } - serde_json::Value::Array(arr) => { - for item in arr { - sort_object_keys(item); - } - } - _ => {} - } -} +// cache(audit/cache.rs):F-260616-05 高危工具去重缓存。 +// 第三批从本文件抽离,行为零变更。pub(super) use 供本文件 process_tool_calls 裸名调用 +// (find_cached_high_risk_result),PENDING_APPROVAL_PLACEHOLDER 供 process_tool_calls 占位回填共享。 +mod cache; +pub(super) use cache::{find_cached_high_risk_result, PENDING_APPROVAL_PLACEHOLDER}; /// AE-2025-03（路径 B）：write_file 审批预览 diff 生成。 ///