lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化: aichat效率剩余(压缩后台化防阻塞/审计批量事务/只读缓存轮内去重/流式增量渲染/AiCommandOutput合批/双渲染合并) + 跨端加固(df-project路径保留大小写/tunnel文档更正supervisor重连/relay固定时间比较与帧上限/启动校验) + 销账

Browse Source
This commit is contained in:
lxy
2026-08-09 21:35:59 +08:00
parent fbd8fae44b
commit 11f4978ec1
15 changed files with 841 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/ai/agentic/context_lifecycle.rs
+210 -100
View File
@@ -1,11 +1,11 @@
//! Agentic 循环上下文生命周期:F-15 自动压缩(智能裁剪)。
//!
//! 从 `mod.rs::run_agentic_loop` 机械抽取(L1017-1193),行为零变更。
//! 从 `mod.rs::run_agentic_loop` 机械抽取,后经 L0-3 后台化(见下)。
//!
//! 触发条件:history_tokens > budget*0.6 且 保护区外有可压缩消息 且 未在压缩中。
//!
//! 流程(对齐 ai_chat_compress_context IPC 的 read-but-don't-mutate 模式):
//! ① 读 active 克隆(不改 status / 不扣 token)→ 喂 LLM 出摘要;
//! ① 读 active 克隆(不改 status / 不扣 token)→ 后台喂 LLM 出摘要;
//! ② LLM 成功 → compress_old_messages(标 compressed + 扣 token)+ insert_at(摘要 system);
//! ③ LLM 失败 → 消息状态完全不变(未改 status / 未扣 token),降级走原 build_for_request 裁剪。
//!
@@ -14,14 +14,21 @@
//! 若先 compress_old_messages(扣 token)再 LLM,失败回滚需精确恢复 history_tokens,
//! 但 ChatMessage 克隆不含 token_count,无法等量加回——回滚 token 不精确。
//! 延迟 mutate 则失败时零副作用(消息状态/token 完全不变),语义最干净。
//! 注:延迟 mutate 的窗口(active_msgs 读出→LLM 出摘要期间)不持锁,但 loop 串行无并发
//! (本函数独占 session_arc,工具执行/审批分支在 stream 之后),故此窗口内 messages 不变。
//! 注:延迟 mutate 的窗口(active_msgs 读出→LLM 出摘要期间)不持锁;L0-3 后台化后窗口拉长
//! (loop 继续推进,可能 push 新消息),但 loop 只 push 增长、protect_start 基于触发时 len,
//! 压缩区间始终落在当前轮之前,不误伤当前轮(见 maybe_auto_compress L0-3 注释)。
//!
//! 安全:复用 loop 顶部已 build+验证 的 provider(不再 build_provider_for 重复 resolve
//! keyring),api_key 经 df_storage::secret 闭环;summary/error payload/日志均不含 api_key。
//! is_compressing 防重入:set_compressing(true/false) 成对(LLM 调用前后均复位)。
//! 安全(L0-3 后台化后):provider 在后台 task 内经 `secret::build_provider_for` 重建
//! (不再复用 loop 顶部已 build 的 provider——后者是 Box 非 Clone 无法移入 task;多一次
//! keyring resolve 在后台线程不阻塞 loop),api_key 经 df_storage::secret 闭环;
//! summary/error payload/日志均不含 api_key。
//! is_compressing 防重入:set_compressing(true) 在触发时置位,后台 task 各出口
//! (成功/失败/conv 已删/provider 重建失败)均复位,防重入不泄漏。
//! 单轮问答(history_tokens 未超 0.6*budget)不触发,零行为变化。
//! messages 操作改 per_conv。
//!
//! L0-3(2026-08-09):压缩 LLM 调用改后台 fire-and-forget(tokio::spawn),不再同步阻塞本轮
//! loop(非流式 LLM 最长 60s)。当前轮用未压缩上下文正常走(不丢回复),压缩结果下一轮生效。
use std::sync::Arc;
@@ -29,7 +36,7 @@ use tauri::{AppHandle, Emitter, Manager};
use tokio::sync::Mutex;
use df_ai::context_helpers::{extract_keyword_summary, CompressedSummary};
use df_ai::provider::{ChatMessage, LlmProvider};
use df_ai::provider::ChatMessage;
use df_storage::models::AiProviderRecord;
use crate::state::{AppState, LlmConcurrency};
@@ -53,19 +60,26 @@ const KEYWORD_FALLBACK_ENABLED: bool = true;
/// 自动压缩(智能裁剪)——在 build_for_request 之前预处理。
///
/// 从 `run_agentic_loop` 抽取,行为零变更。返回值:
/// **L0-3(2026-08-09)后台化**:压缩 LLM 调用不再同步阻塞本轮 loop(非流式 LLM 最长 60s)。
/// 触发时仅做零成本准备(读触发条件 / 读 active 克隆 / 算关键词兜底)后立即 `tokio::spawn`
/// 独立 task fire-and-forget 执行压缩,本函数返回 `false` 让当前轮**用未压缩上下文正常走**
/// (build_for_request 不受影响,当前轮回复不丢);压缩结果由后台 task 在完成时更新
/// per_conv.messages(标 compressed + 插摘要),**下一轮** build_for_request 生效。
/// `is_compressing` 防重入:触发时置 true,后台 task 各出口(成功/失败/conv 已删)均复位。
///
/// 返回值:
/// - `true`:conv 已删除,调用方应立即 `return`(退出整个 loop)。
/// - `false`:正常结束(无论是否触发压缩、压缩是否成功),调用方继续后续流程。
///
/// 参数沿用原内联代码引用的局部:
/// - `provider`:loop 顶部 build 的 provider(F-04b 切换 candidate 后由调用方传入最新)。
/// - `provider_config`:同上,compress 沿用主 candidate(非 fallback 范围)。
/// - `provider_config`:compress 沿用主 candidate(非 fallback 范围)。后台 task 内经
/// `secret::build_provider_for` 重建 provider(复用 loop 顶部同款三步打包,仅多一次
/// keyring resolve,在后台线程不阻塞 loop)。
/// - `iteration`:当前轮次(仅用于日志)。
pub(super) async fn maybe_auto_compress(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
app_handle: &AppHandle,
provider: &Box<dyn LlmProvider>,
provider_config: &AiProviderRecord,
llm_concurrency: &LlmConcurrency,
iteration: usize,
@@ -142,7 +156,7 @@ pub(super) async fn maybe_auto_compress(
}
}
// 改进3 B:在 active_msgs move 进 compress_via_llm 前,先算关键词摘要兜底文本。
// 改进3 B:在 active_msgs move 进压缩前,先算关键词摘要兜底文本。
// LLM 压缩失败时仍想保留用户反复提及的主题词(续接锚点),避免裸裁剪丢主题。
// KEYWORD_FALLBACK_ENABLED=false → 跳过(回退原裸裁剪行为,排障/对比用)。
let keyword_fallback: String = if KEYWORD_FALLBACK_ENABLED {
@@ -151,101 +165,197 @@ pub(super) async fn maybe_auto_compress(
String::new()
};
// 压缩调用(复用 loop 顶部已 build 的 provider,api_key 经 secret 闭环)。
// 成功 → Some(summary);失败 → Err;无 active 可压缩(active_msgs 空)→ 视为 noop。
let compress_outcome: Result<Option<CompressedSummary>, String> = if active_msgs.is_empty() {
Ok(None)
} else {
compress_via_llm(
provider.as_ref(),
provider_config,
active_msgs,
&lang,
conv_id,
llm_concurrency,
).await.map(Some)
};
// L0-3:保护区外无 active 可压缩(active_msgs 空)→ noop,同步复位 is_compressing
// 后返回(不 spawn 空任务)。对齐原 Ok(None) 分支语义。
if active_msgs.is_empty() {
session_arc.lock().await.conv(&conv_id).messages.set_compressing(false);
return false;
}
match compress_outcome {
Ok(Some(summary)) => {
// LLM 成功 → 标 compressed(扣 token)+ 摘要 system 插首位 + set_compressing(false)。
{
let mut session = session_arc.lock().await;
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(protect_start);
// T3: 插入 NL 摘要(向前兼容),JSON 卡片可供后续 WorkingContext(T5)使用
conv.messages.insert_at(0, ChatMessage::system(&summary.nl_summary));
conv.messages.set_compressing(false);
// TODO(T5): 如有 WorkingContext,调用 version.reset_all(summary.json_card, turn)
}
tracing::info!(
conv_id = %conv_id,
iteration,
pre_tokens = pre_compress_tokens,
"[ai] 自动压缩成功,摘要已插首位"
);
// 治 Task#1:loop 自动压缩用 AiAutoCompressed(非手动变体),
// 桌面端静默(仅复位 isCompressing,不弹 toast 不 switchConversation)。
// miniapp 仍插摘要气泡(对端发生压缩告知用户)。
let ev = AiChatEvent::AiAutoCompressed {
conversation_id: Some(conv_id.to_string()),
summary: summary.nl_summary,
};
let _ = app_handle.emit("ai-chat-event", ev.clone());
// L3 emit 双写:tunnel subscriber透传 miniapp
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(ev);
}
Ok(None) => {
// 保护区外无 active 可压缩(已全 compressed/archived)→ noop,仅复位 is_compressing。
session_arc.lock().await.conv(&conv_id).messages.set_compressing(false);
}
Err(e) => {
// LLM 失败 → 改进3 B:仍标 compressed 释放 token + 关键词摘要塞回首条(非裸裁剪)。
//
// 旧行为:消息状态完全不变,降级走 build_for_request 裁剪(丢主题)。
// 新行为(KEYWORD_FALLBACK_ENABLED=true 默认):
// - compress_old_messages 标 [..protect_start] active 为 compressed(释放 token,
// 与成功路径一致,后续 build_for_request 不再把它们进 LLM 上下文);
// - keyword_fallback 非空 → insert_at(0, system 关键词摘要)作续接锚点;
// - keyword_fallback 空(无 user 消息/无可提取词)→ 不插,等价旧裁剪(保底)。
// 持久化语义不变:compressed 仍软删可追溯(DB 全量保留),与成功路径一致。
// KEYWORD_FALLBACK_ENABLED=false → 跳过兜底,等价旧行为(set_compressing(false) +
// 消息状态不变,降级 build_for_request 裁剪)。
session_arc.lock().await.conv(&conv_id).messages.set_compressing(false);
tracing::warn!(
conv_id = %conv_id,
error = %e,
keyword_fallback_len = keyword_fallback.len(),
"[ai] 自动压缩失败,降级走关键词摘要兜底(KEYWORD_FALLBACK_ENABLED={})",
KEYWORD_FALLBACK_ENABLED,
);
if KEYWORD_FALLBACK_ENABLED {
// 标 compressed 释放 token + 关键词摘要塞首位(若非空)。
let inserted = {
// ── L0-3 后台化:压缩 LLM 调用改 fire-and-forget(tokio::spawn 独立 task) ──
//
// 原实现在此 `compress_via_llm(...).await` 同步阻塞整轮(非流式 LLM 最长 60s),
// 期间 loop 无法推进 build_for_request / stream,当前轮回复整体延迟。
//
// 现改为 spawn 独立 task 后台执行压缩,本函数立即返回 false:
// - 当前轮 build_for_request 用**未压缩上下文**,回复不受压缩阻塞影响(关键:不丢当前轮);
// - 后台 task 完成后更新 per_conv.messages(compress_old_messages + insert 摘要),
// 下一轮 build_for_request 生效;
// - is_compressing 防重入:触发时已置 true(上方),后台 task 各出口(成功/失败/conv 已删)
// 均复位,下一轮 maybe_auto_compress 检测到 true 即跳过,不会并发二次压缩。
//
// 选型理由(后台化 vs 降级独立线程):并发安全成立——所有 per_conv 访问经
// session_arc 锁(数据无竞争),且压缩只读「触发时刻快照的 active_msgs」+ 完成后
// 一次性 mutate,与 loop 的 push 均为「锁内短段」,protect_start 基于触发时 len
// (loop 只 push 增长,压缩区间始终落在当前轮之前),故选 tokio::spawn 后台化,
// 无需降级独立线程。provider 在 task 内重建(secret::build_provider_for),避免
// 与 loop 共享 Box<dyn LlmProvider>(后者非 Clone,无法移入 task)。
//
// 无感知降级:provider 重建失败 / LLM 失败 → 仅复位 is_compressing + 关键词兜底,
// 不发 AiError(前端 AiError 会误判生成结束致"压缩后停止"),loop 继续 stream。
let session_arc = session_arc.clone();
let app_handle = app_handle.clone();
let conv_id = conv_id.to_string();
let provider_config = provider_config.clone();
let llm_concurrency = llm_concurrency.clone();
tokio::spawn(async move {
// 后台重建 provider(secret::build_provider_for 三步打包:resolve→ensure→build)。
// 失败 → 同 LLM 失败降级:复位 is_compressing + 关键词摘要兜底(标 compressed 释放
// token + 插关键词),保持「失败→兜底→扣 token」语义——否则 history_tokens 未降,
// 下轮 maybe_auto_compress 会重复触发(每次仅 spawn 后台任务,不阻塞 loop,但浪费)。
let provider = match super::super::secret::build_provider_for(&provider_config) {
Ok(p) => p,
Err(e) => {
tracing::warn!(
conv_id = %conv_id,
error = %e,
keyword_fallback_len = keyword_fallback.len(),
"[ai] 后台压缩 provider 重建失败,降级走关键词摘要兜底(KEYWORD_FALLBACK_ENABLED={})",
KEYWORD_FALLBACK_ENABLED,
);
// 单次锁内 check+mutate:复位 is_compressing(防 conv 两次锁间被删惰性重建)。
{
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(&conv_id) {
session.conv(&conv_id).messages.set_compressing(false);
}
}
if KEYWORD_FALLBACK_ENABLED {
// 同 LLM 失败兜底:标 compressed 释放 token + 关键词摘要塞首位(若非空)。
let inserted = {
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
return;
}
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(protect_start);
if !keyword_fallback.is_empty() {
conv.messages.insert_at(0, ChatMessage::system(&keyword_fallback));
true
} else {
false
}
};
if inserted {
tracing::info!(
conv_id = %conv_id,
pre_tokens = pre_compress_tokens,
"[ai] 压缩 provider 重建失败兜底(后台):关键词摘要已插首位(compressed 标记已扣 token)"
);
}
}
return;
}
};
// 压缩调用(复用主 candidate 的 provider_config,api_key 经 secret 闭环)。
// 成功 → Some(summary);失败 → Err。
let compress_outcome: Result<Option<CompressedSummary>, String> =
compress_via_llm(
provider.as_ref(),
&provider_config,
active_msgs,
&lang,
&conv_id,
&llm_concurrency,
).await.map(Some);
match compress_outcome {
Ok(Some(summary)) => {
// LLM 成功 → 标 compressed(扣 token)+ 摘要 system 插首位 + set_compressing(false)。
// conv 已删(per_conv 无条目)时丢弃压缩结果——惰性重建会污染新 conv,
// is_compressing 标志随 conv 销毁无意义,直接 return。
{
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
return;
}
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(protect_start);
if !keyword_fallback.is_empty() {
conv.messages.insert_at(0, ChatMessage::system(&keyword_fallback));
true
} else {
false
}
// T3: 插入 NL 摘要(向前兼容),JSON 卡片可供后续 WorkingContext(T5)使用
conv.messages.insert_at(0, ChatMessage::system(&summary.nl_summary));
conv.messages.set_compressing(false);
// TODO(T5): 如有 WorkingContext,调用 version.reset_all(summary.json_card, turn)
}
tracing::info!(
conv_id = %conv_id,
iteration,
pre_tokens = pre_compress_tokens,
"[ai] 自动压缩成功(后台),摘要已插首位"
);
// 治 Task#1:loop 自动压缩用 AiAutoCompressed(非手动变体),
// 桌面端静默(仅复位 isCompressing,不弹 toast 不 switchConversation)。
// miniapp 仍插摘要气泡(对端发生压缩告知用户)。
let ev = AiChatEvent::AiAutoCompressed {
conversation_id: Some(conv_id.clone()),
summary: summary.nl_summary,
};
if inserted {
tracing::info!(
conv_id = %conv_id,
pre_tokens = pre_compress_tokens,
"[ai] 压缩失败兜底:关键词摘要已插首位(compressed 标记已扣 token)"
);
let _ = app_handle.emit("ai-chat-event", ev.clone());
// L3 emit 双写:tunnel subscriber透传 miniapp
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(ev);
}
Ok(None) => {
// 保护区外无 active 可压缩(已全 compressed/archived)→ noop,仅复位 is_compressing。
// 单次锁内 check+mutate(防 conv 在两次锁间被删致惰性重建污染新 conv)。
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(&conv_id) {
session.conv(&conv_id).messages.set_compressing(false);
}
}
Err(e) => {
// LLM 失败 → 改进3 B:仍标 compressed 释放 token + 关键词摘要塞回首条(非裸裁剪)。
//
// 旧行为:消息状态完全不变,降级走 build_for_request 裁剪(丢主题)。
// 新行为(KEYWORD_FALLBACK_ENABLED=true 默认):
// - compress_old_messages 标 [..protect_start] active 为 compressed(释放 token,
// 与成功路径一致,后续 build_for_request 不再把它们进 LLM 上下文);
// - keyword_fallback 非空 → insert_at(0, system 关键词摘要)作续接锚点;
// - keyword_fallback 空(无 user 消息/无可提取词)→ 不插,等价旧裁剪(保底)。
// 持久化语义不变:compressed 仍软删可追溯(DB 全量保留),与成功路径一致。
// 压缩失败不发 AiError——前端 AiError case 无条件 setStreaming(false)
// 误判生成结束(后端 loop 实际继续),致"压缩后停止"。降级为 warn + 兜底释放
// token + build_for_request 裁剪,loop 继续 stream_llm,用户自然看到后续回复。
// 单次锁内 check+mutate(防 conv 在两次锁间被删致惰性重建污染新 conv)。
{
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(&conv_id) {
session.conv(&conv_id).messages.set_compressing(false);
}
}
tracing::warn!(
conv_id = %conv_id,
error = %e,
keyword_fallback_len = keyword_fallback.len(),
"[ai] 自动压缩失败(后台),降级走关键词摘要兜底(KEYWORD_FALLBACK_ENABLED={})",
KEYWORD_FALLBACK_ENABLED,
);
if KEYWORD_FALLBACK_ENABLED {
// 标 compressed 释放 token + 关键词摘要塞首位(若非空)。
let inserted = {
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
return;
}
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(protect_start);
if !keyword_fallback.is_empty() {
conv.messages.insert_at(0, ChatMessage::system(&keyword_fallback));
true
} else {
false
}
};
if inserted {
tracing::info!(
conv_id = %conv_id,
pre_tokens = pre_compress_tokens,
"[ai] 压缩失败兜底(后台):关键词摘要已插首位(compressed 标记已扣 token)"
);
}
}
}
// 压缩失败不发 AiError——前端 AiError case 无条件 setStreaming(false)
// 误判生成结束(后端 loop 实际继续),致"压缩后停止"。降级为 warn(上方 line 957 已有)
// + compress_old_messages 兜底释放 token + build_for_request 裁剪,loop 继续 stream_llm,
// 用户自然看到后续回复。压缩降级对用户透明(非致命错误,不该停流)。
}
}
});
}
}
false
src-tauri/src/commands/ai/agentic/mod.rs
+4 -3
View File
@@ -1498,15 +1498,16 @@ pub(crate) async fn run_agentic_loop(
// 自动压缩(智能裁剪)——已抽取至 context_lifecycle::maybe_auto_compress。
//
// 抽取自本函数原内联块(行为零变更)。返回 true = conv 已删除,loop 应立即 return
// 抽取自本函数原内联块。返回 true = conv 已删除,loop 应立即 return
// (对齐原内联块入口的 early-return);返回 false = 正常结束,继续后续流程。
// 实现细节(触发条件 / 延迟 mutate 口径 / 关键词兜底)见
// context_lifecycle.rs 顶部文档与函数体内注释(原样保留)。
// context_lifecycle.rs 顶部文档与函数体内注释。
// L0-3(2026-08-09):压缩 LLM 调用已后台化(fire-and-forget),本调用不再同步阻塞本轮,
// 返回 false 让当前轮用未压缩上下文正常走;压缩结果下一轮生效(见 context_lifecycle.rs)。
if maybe_auto_compress(
&session_arc,
&conv_id,
&app_handle,
&provider,
&provider_config,
&llm_concurrency,
iteration,
src-tauri/src/commands/ai/audit/cache.rs
+4 -1
View File
@@ -215,7 +215,10 @@ fn canonical_args_key(args: &serde_json::Value) -> String {
/// 故 list_directory 的缓存 key 只取 path(忽略其他参数),同目录即命中缓存,
/// 后续命中返回缓存 + 专用警告(cache_hit_warning 目录分支)。
/// 其他只读工具仍走全参 JSON 归一(键序无关,见 canonical_args_key)。
fn readonly_cache_args_key(tool_name: &str, args: &serde_json::Value) -> String {
///
/// pub(crate):供 audit/mod.rs 轮内去重(AC-EFF-T1-2)复用同一 key 口径(同源同 key,
/// 跨轮缓存 + 轮内去重判定一致,防两处 key 漂移致去重误判)。
pub(crate) fn readonly_cache_args_key(tool_name: &str, args: &serde_json::Value) -> String {
if matches!(tool_name, "list_directory" | "list_dir") {
args.get("path")
.and_then(|v| v.as_str())
src-tauri/src/commands/ai/audit/mod.rs
+114 -6
View File
@@ -41,7 +41,7 @@ use approval::{detect_retry_count, handle_approval_tool};
// （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};
pub(crate) use record::{audit_tool_call, build_audit_record, query_audit_history, record_audit};
#[allow(unused_imports)]
pub use record::{
list_tool_executions, tool_failure_stats, ToolExecutionDto, ToolExecutionPage, ToolExecQuery,
@@ -70,7 +70,7 @@ pub(crate) use finalize::audit_finalize;
// cache(audit/cache.rs):高危工具去重缓存 + 只读工具缓存。
// 第三批从本文件抽离,行为零变更。
mod cache;
pub(super) use cache::{cache_hit_warning, detect_listing_bypass, find_cached_readonly_result, insert_listing_bypass_warning, pending_placeholder_for};
pub(super) use cache::{cache_hit_warning, detect_listing_bypass, find_cached_readonly_result, insert_listing_bypass_warning, pending_placeholder_for, readonly_cache_args_key};
// data_change(audit/data_change.rs):AR-11 数据变更联动刷新。
// 第四批从本文件抽离，行为零变更。pub(crate) use 保持 emit_data_changed 对 crate 内可见
@@ -594,16 +594,37 @@ pub(crate) async fn process_tool_calls(
// 同参只读工具成功执行过,命中则直接回填缓存结果跳过真执行,断 LLM 失忆死循环。
// 安全边界见 find_cached_readonly_result 文档(仅白名单只读工具 + 仅 completed 成功结果)。
//
// AC-EFF-T1-2(2026-08-09)轮内去重:同轮重复 (tool, args_key) 只真执行一次。
// LLM 单轮可能发多条同参只读调用(失忆重调),原实现首查未命中(本轮首个尚未落 messages)
// 致重复全部真执行(重复 I/O + 重复 tool_result 回灌)。现进程内暂存首个结果
// (in_round_cache),后续同 key 复用——语义同跨轮缓存,仅因首个在本轮未落库故内存暂存。
// 命中计数递增威慑(cache_hit_warning 同口径);key 与跨轮缓存同源(readonly_cache_args_key),
// 防两处 key 口径漂移。跨轮重复仍走 find_cached_readonly_result(历史扫描,未变)。
//
// find_cached_readonly_result 内部短 lock + 锁外 DB 查,本段不持锁。
// in_round_cache:同轮 (tool, args_key) -> (首个结果内容, 已见同参次数)。首个出现:
// 跨轮缓存命中 → 存 (content, hit_count);未命中 → 真执行,backfill 后存 (content, 1)。
// in_round_dupes:同轮重复的 draft(延后到 backfill 后复用 in_round_cache,防双执行)。
let mut in_round_cache: HashMap<(String, String), (String, u32)> = HashMap::new();
let mut seen_round_keys: HashSet<(String, String)> = HashSet::new();
let mut in_round_dupes: Vec<(ToolCallDraft, RiskLevel, String)> = Vec::new();
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 args_key = readonly_cache_args_key(&draft.name, &args);
let key = (draft.name.clone(), args_key.clone());
if !seen_round_keys.insert(key) {
// 轮内重复:延后复用首个结果(首个已在 seen/uncached/backfill),不重复历史扫描+DB 查。
in_round_dupes.push((draft, risk_level, args_key));
continue;
}
let cached = find_cached_readonly_result(session_arc, conv_id, &audit_repo, &draft.name, &args).await;
if let Some((cached_content, hit_count)) = cached {
// 缓存命中:直接 push tool_result + 审计(decided_by=cache_hit 标记缓存来源),
// 不走真执行 + 不重emit Started/Completed(避免误导前端工具又执行了一次)。
// 缓存命中:存轮内缓存(供后续同 key 复用)+ 直接 push tool_result + 审计
// (decided_by=cache_hit 标记缓存来源),不走真执行 + 不重emit Started/Completed。
// AC-1 根治:弱模型不知道结果来自缓存,仍死循环重调同参工具。此处给回填的
// tool_result 前置「重复调用拦截」警告头(机制化提示,LLM 能看到 tool_result),
// 命中次数递增威慑,告知勿再重复调用相同参数的工具。
in_round_cache.insert((draft.name.clone(), args_key), (cached_content.clone(), hit_count));
let warned_content = format!("{}{}", cache_hit_warning(&draft.name, hit_count), cached_content);
// emit Completed 携带缓存结果供前端折叠卡片展示(与 find_cached_high_risk_result 一致)。
let ev = AiChatEvent::AiToolCallCompleted {
@@ -680,7 +701,12 @@ pub(crate) async fn process_tool_calls(
}
})).await;
// 串行回填 tool_result + 审计（短 lock push + 锁外 audit）
// 串行回填 tool_result + 审计收集(短 lock push + 锁外收集),审计改单事务批量插入。
// AC-EFF-T1-1(2026-08-09):原每结果 audit_tool_call(每次 spawn_blocking + 单行 INSERT,
// 单连接 Mutex 串行)合并为 build_audit_record 收集 + insert_batch 一次事务批量,
// 砍 N 次串行 INSERT 尾巴(治 aichat 效率走查 T1-1)。审计失败不阻断主流程(记日志)。
let mut audit_records: Vec<df_storage::models::AiToolExecutionRecord> =
Vec::with_capacity(results.len());
for (draft, raw_result, risk_level, outcome) in results {
let (status, content) = match outcome {
Ok(c) => ("completed", c),
@@ -717,7 +743,89 @@ pub(crate) async fn process_tool_calls(
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;
// AC-EFF-T1-2 轮内去重回填:真执行结果存入 in_round_cache,供同轮后续同 key 复用
// (首个执行 → count=1,轮内重复命中时递增;key 与跨轮缓存同源)。
{
let args_val = serde_json::from_str(&draft.args).unwrap_or(serde_json::Value::Null);
let exec_key = (draft.name.clone(), readonly_cache_args_key(&draft.name, &args_val));
in_round_cache.entry(exec_key).or_insert_with(|| (content.clone(), 1));
}
audit_records.push(build_audit_record(
conv_id, &draft.id, &draft.name, &draft.args, status, risk_level,
Some(content), Some(decided_by), current_message_id,
));
}
// 单事务批量 INSERT(一次 spawn_blocking,砍 N 次串行尾巴;失败仅记日志不阻断)。
if !audit_records.is_empty() {
let batch_len = audit_records.len();
if let Err(e) = audit_repo.insert_batch(audit_records).await {
tracing::error!(
conv_id = %conv_id,
batch = batch_len,
"[ai] insert_batch 批量写审计记录失败(已回滚,共 {} 条): {}",
batch_len,
e
);
}
}
// AC-EFF-T1-2 轮内重复处理:复用首个结果(已回填 in_round_cache),跳过重复真执行。
// 语义同跨轮缓存命中:emit Completed + push tool_result + 审计(decided_by=cache_hit),
// 命中计数递增威慑(cache_hit_warning 同口径)。首个结果来自跨轮命中(前段已存)或
// 本轮真执行 backfill(上段已存),故必命中;防御性缺失则丢弃(不重复执行,LLM 下轮可重调)。
let mut dupe_audit_records: Vec<df_storage::models::AiToolExecutionRecord> =
Vec::with_capacity(in_round_dupes.len());
for (draft, risk_level, args_key) in in_round_dupes {
let key = (draft.name.clone(), args_key);
let Some((content, count)) = in_round_cache.get(&key).cloned() else {
tracing::warn!(
conv_id = %conv_id,
tool = %draft.name,
tc_id = %draft.id,
"[ai] 轮内去重:首个结果缺失(理论不可达),丢弃重复调用(防御降级)"
);
continue;
};
let new_count = count + 1;
in_round_cache.insert(key, (content.clone(), new_count));
let warned_content = format!("{}{}", cache_hit_warning(&draft.name, new_count), content);
// emit Completed 携带缓存结果供前端折叠卡片展示。
let ev = AiChatEvent::AiToolCallCompleted {
id: draft.id.clone(),
result: serde_json::Value::String(warned_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, &warned_content));
}
dupe_audit_records.push(build_audit_record(
conv_id, &draft.id, &draft.name, &draft.args, "completed", risk_level,
Some(warned_content), Some("cache_hit"), current_message_id,
));
tracing::info!(
conv_id = %conv_id,
tool = %draft.name,
tc_id = %draft.id,
hit_count = new_count,
"[ai] 只读工具轮内去重命中(第 {} 次同参,本轮首个已执行),跳过重复真执行",
new_count
);
}
if !dupe_audit_records.is_empty() {
let dup_len = dupe_audit_records.len();
if let Err(e) = audit_repo.insert_batch(dupe_audit_records).await {
tracing::error!(
conv_id = %conv_id,
batch = dup_len,
"[ai] insert_batch 轮内去重审计批量写失败(已回滚,共 {} 条): {}",
dup_len,
e
);
}
}
}
src-tauri/src/commands/ai/audit/record.rs
+48 -25
View File
@@ -24,10 +24,50 @@ use crate::state::AppState;
use super::risk_str;
/// 写一条工具执行审计记录（insert 失败不阻断主流程，故 `let _ =`）
/// 构造一条工具执行审计记录(纯函数,单条 [`audit_tool_call`] / 批量插入路径共用)。
///
/// `decided_by` 有值（auto/human）= 已决策执行 → 记 executed_at；
/// `None`（pending 待审批）→ executed_at 留空，待 audit_finalize 回填。
/// `decided_by` 有值(auto/human)= 已决策执行 → 记 executed_at;
/// `None`(pending 待审批)→ executed_at 留空,待 audit_finalize 回填。
pub(crate) fn build_audit_record(
conv_id: &str,
tool_call_id: &str,
tool_name: &str,
arguments: &str,
status: &str,
risk_level: RiskLevel,
result: Option<String>,
decided_by: Option<&str>,
message_id: Option<&str>,
) -> AiToolExecutionRecord {
let executed_at = if decided_by.is_some() { Some(now_millis()) } else { None };
AiToolExecutionRecord {
id: new_id(),
conversation_id: Some(conv_id.to_string()),
// P1 消息级溯源:message_id 由调用方(process_tool_calls)从
// ContextManager 取当前 assistant 消息 id 传入(LLM 返回带 tool_calls 的
// assistant 消息已 push 到 per_conv.messages,入口取末条 assistant id)。
// None 表示无 assistant 消息(异常路径/老数据无 id),展示侧兼容。
message_id: message_id.map(|s| s.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()),
}
}
/// 写一条工具执行审计记录(insert 失败不阻断主流程,故 `let _ =`)
///
/// 单条写入路径。批量路径(audit/mod.rs process_tool_calls 回填循环)经
/// [`build_audit_record`] 收集记录后调 `AiToolExecutionRepo::insert_batch`
/// 单事务批量插入(治 aichat 效率 AC-EFF-T1-1,N 次串行 INSERT → 一次事务)。
///
/// `decided_by` 有值(auto/human)= 已决策执行 → 记 executed_at;
/// `None`(pending 待审批)→ executed_at 留空,待 audit_finalize 回填。
pub(crate) async fn audit_tool_call(
repo: &AiToolExecutionRepo,
conv_id: &str,
@@ -40,28 +80,11 @@ pub(crate) async fn audit_tool_call(
decided_by: Option<&str>,
message_id: Option<&str>,
) {
let executed_at = if decided_by.is_some() { Some(now_millis()) } else { None };
if let Err(e) = repo
.insert(AiToolExecutionRecord {
id: new_id(),
conversation_id: Some(conv_id.to_string()),
// P1 消息级溯源:message_id 由调用方(process_tool_calls)从
// ContextManager 取当前 assistant 消息 id 传入(LLM 返回带 tool_calls 的
// assistant 消息已 push 到 per_conv.messages,入口取末条 assistant id)。
// None 表示无 assistant 消息(异常路径/老数据无 id),展示侧兼容。
message_id: message_id.map(|s| s.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
{
let record = build_audit_record(
conv_id, tool_call_id, tool_name, arguments,
status, risk_level, result, decided_by, message_id,
);
if let Err(e) = repo.insert(record).await {
tracing::error!(
"audit_tool_call: 写审计记录失败(conv={}, tool_call_id={}, tool={}): {}",
conv_id,
src-tauri/src/commands/ai/command_stream.rs
+149 -8
View File
@@ -13,38 +13,179 @@
//! 读 task-local(同一 task,因 tools.execute 不 spawn 直接 await handler),
//! 命中则改走 shell `execute_streaming`,每行回调 [`emit_output`] → AiCommandOutput。
//!
//! AC-EFF-S2-5(2026-08-09):AiCommandOutput 合批。run_command 逐行回调不再每行单独 emit +
//! publish_event(大输出风暴时 IPC 事件风暴),改为累积进 [`CommandSink`] 共享缓冲(stdout/stderr
//! 分桶),由「后台 50ms 定时 flush + 体积阈值(4KB)立即 flush + Drop 兜底 flush」三路输出,
//! 每个 stream 字段累积的字符串(多行 '\n' 连接)作为单条 AiCommandOutput 输出(对齐
//! stream_recv DELTA_FLUSH_INTERVAL=50ms 的 delta 合批;消费方按文本渲染,多行输出等价逐行)。
//!
//! 未注入 sink(非 run_command / 调用方未配 scope)时 [`emit_output`] 静默 noop,
//! 兜底不报错不阻断。
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex as StdMutex;
use std::time::Duration;
use tauri::{AppHandle, Emitter, Manager};
use super::AiChatEvent;
use df_execute::shell::StreamKind;
/// AiCommandOutput 合批参数(AC-EFF-S2-5,对齐 stream_recv DELTA_FLUSH_INTERVAL=50ms)。
///
/// 时间维度:后台 flush task 每 50ms 兜底输出一次(与 AR-8 delta 合批同 cadence);
/// 体积维度:单次累积超 [`FLUSH_BYTE_THRESHOLD`] 立即 flush(防大输出风暴 50ms 窗口内积压过多)。
/// 语义:stdout/stderr 各累积的字符串(多行以 '\n' 连接)作为**单条** AiCommandOutput 事件输出,
/// 不再每行单独 emit + publish_event——砍 IPC 事件风暴(与 delta 合批「合并后前端逐条累加,最终一致」
/// 同思路,消费方按文本渲染,多行输出等价于逐行)。
const FLUSH_INTERVAL: Duration = Duration::from_millis(50);
const FLUSH_BYTE_THRESHOLD: usize = 4096;
/// 一次 run_command 调用的输出下沉目标(emit 事件所需上下文)。
#[derive(Clone)]
///
/// AC-EFF-S2-5(2026-08-09):合批。run_command 逐行回调 → 累积进共享缓冲(stdout/stderr 分桶),
/// 由「后台 50ms 定时 flush + 体积阈值立即 flush + Drop 兜底 flush」三路输出,不再每行单独
/// emit(治 aichat 效率 S2-5)。缓冲用 `std::sync::Mutex`(短临界区无 await,emit 是同步回调)。
pub struct CommandSink {
app: AppHandle,
tool_call_id: String,
conversation_id: Option<String>,
/// 合批共享缓冲(emit 线程写 / flush 定时读)。
buffer: Arc<StdMutex<CommandBuffer>>,
/// 后台 flush task 守卫(Drop 时停止 + abort,并兜底 flush 尾部)。
_flusher: FlushGuard,
}
/// 合批缓冲:stdout/stderr 分开累积(按 stream 字段 emit),bytes 供体积阈值判定。
#[derive(Default)]
struct CommandBuffer {
stdout: String,
stderr: String,
bytes: usize,
}
/// 后台 flush task 守卫:stop 标志 + JoinHandle。Drop 时置停 + abort(任务 50ms 循环读到 stop 退出),
/// 尾部输出由 CommandSink::drop 兜底 flush,不丢命令结尾。
struct FlushGuard {
stop: Arc<AtomicBool>,
handle: Option<tokio::task::JoinHandle<()>>,
}
impl CommandSink {
pub fn new(app: AppHandle, tool_call_id: String, conversation_id: Option<String>) -> Self {
Self { app, tool_call_id, conversation_id }
let buffer = Arc::new(StdMutex::new(CommandBuffer::default()));
let stop = Arc::new(AtomicBool::new(false));
// 后台 50ms flush task。仅当处于 tokio runtime 上下文时 spawn(tokio::spawn 需 runtime)。
// CommandSink::new 由 execute_with_heartbeat / ai_approve(async)调用,正常在 runtime 内;
// 非 runtime(如单测直接 new)跳过定时 task,退化为「体积阈值 + Drop 兜底」flush(仍不丢尾部)。
let handle = if tokio::runtime::Handle::try_current().is_ok() {
let buf = buffer.clone();
let stop_f = stop.clone();
let app_f = app.clone();
let id_f = tool_call_id.clone();
let conv_f = conversation_id.clone();
Some(tokio::spawn(async move {
let mut interval = tokio::time::interval(FLUSH_INTERVAL);
interval.tick().await; // 弃首 tick(tokio interval 首 tick 立即返回,对齐心跳弃首模式)
loop {
interval.tick().await;
if stop_f.load(Ordering::SeqCst) {
break;
}
Self::flush_inner(&app_f, &id_f, &conv_f, &buf);
}
}))
} else {
None
};
Self {
app,
tool_call_id,
conversation_id,
buffer,
_flusher: FlushGuard { stop, handle },
}
}
/// emit 一行 stdout/stderr(AiCommandOutput,双写 app.emit + ai_event_bus)。
/// emit 失败静默吞(前端未 listen / 总线无订阅不阻断命令执行)。
/// 累积一行 + 触发 flush 判定(行加入 stdout/stderr 分桶)。
///
/// AC-EFF-S2-5:不立即 emit,先累积入缓冲;体积超阈值立即 flush,否则等 50ms 定时 flush
/// 或 Drop 兜底 flush。emit 失败静默吞(前端未 listen / 总线无订阅不阻断命令执行)。
fn emit(&self, kind: StreamKind, line: &str) {
let over_threshold = {
let mut buf = self.buffer.lock().unwrap();
let target = if kind == StreamKind::Stdout { &mut buf.stdout } else { &mut buf.stderr };
if !target.is_empty() {
target.push('\n');
}
target.push_str(line);
buf.bytes += line.len();
buf.bytes >= FLUSH_BYTE_THRESHOLD
};
if over_threshold {
Self::flush_inner(&self.app, &self.tool_call_id, &self.conversation_id, &self.buffer);
}
}
/// 兜底 flush(把当前缓冲整体取出并 emit)。Drop 时调用,保证尾部输出不丢。
fn flush(&self) {
Self::flush_inner(&self.app, &self.tool_call_id, &self.conversation_id, &self.buffer);
}
/// 内部 flush:单次锁内 take 缓冲(不持锁 emit),stdout/stderr 各输出一条(多行拼接)。
/// 与后台 task / Drop 并发调用安全:take 语义下同一批只被 drain 一次(无重复 emit)。
fn flush_inner(
app: &AppHandle,
tool_call_id: &str,
conversation_id: &Option<String>,
buffer: &Arc<StdMutex<CommandBuffer>>,
) {
let drained = {
let mut buf = buffer.lock().unwrap();
if buf.stdout.is_empty() && buf.stderr.is_empty() {
return;
}
Some(std::mem::take(&mut *buf))
};
if let Some(buf) = drained {
if !buf.stdout.is_empty() {
Self::emit_line(app, tool_call_id, conversation_id, StreamKind::Stdout, &buf.stdout);
}
if !buf.stderr.is_empty() {
Self::emit_line(app, tool_call_id, conversation_id, StreamKind::Stderr, &buf.stderr);
}
}
}
/// emit 一条 AiCommandOutput(stdout/stderr 累积字符串,双写 app.emit + ai_event_bus)。
fn emit_line(
app: &AppHandle,
tool_call_id: &str,
conversation_id: &Option<String>,
kind: StreamKind,
line: &str,
) {
let ev = AiChatEvent::AiCommandOutput {
id: self.tool_call_id.clone(),
id: tool_call_id.to_string(),
stream: kind.as_str().to_string(),
line: line.to_string(),
conversation_id: self.conversation_id.clone(),
conversation_id: conversation_id.clone(),
};
let _ = self.app.emit("ai-chat-event", ev.clone());
let _ = self.app.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
let _ = app.emit("ai-chat-event", ev.clone());
let _ = app.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
}
}
impl Drop for CommandSink {
fn drop(&mut self) {
// 停后台 flush task + abort(任务在 stop 后下次 tick 退出),再兜底 flush 剩余缓冲
// (命令结束尾部输出不丢)。并发安全:flush_inner 的 take 语义保证不重复 emit。
self._flusher.stop.store(true, Ordering::SeqCst);
if let Some(h) = self._flusher.handle.take() {
h.abort();
}
self.flush();
}
}