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

优化: 边界加固(AI loop竞态根治+数据/审批/反馈/并发/错误分类)

Browse Source 
AI loop 竞态(P0):per-conv epoch/owner token + 存活心跳治 force_send 双loop + stop 3s兜底误判;旧loop stale 全跳过(guard/emit/save)

agentic 收尾(A2-B8):Fatal 退出落库user消息(镜像Exhausted)+ 入口早退补save + usage is_estimated 打标 + emit_ai_completed_once 单点收敛清审批残留

聊天清理(A2-B9):clearChat 先停loop→DB单事务→内存清(clear_conversation_atomic)+ 前端错误气泡

循环并发(A2-B11):三态 ProviderAcquire(NotConfigured/Acquired/Exhausted)+ 候选循环非阻塞+防抖3次饱和降级+单测

错误分类(A2-B12):stream error帧接入 classify_status_or_class + 关键词保守降级 + 7单测

数据(G1.2/G1.4):purge_with_descendants 级联补全(11表单事务+存在性守卫)+ move_task_queue 单事务收口(两调用方共用)

git只读(G3.1):run_git_status/diff/log success判定(exit_code差异语义,失败结构化{success:false,error})

安全(G5.2/G5.6):create_project 目录Err+name校验 + module.rs 路径遍历DRY(分段匹配修a..b.rs误伤)

幂等(V2/V32):裸ALTER全守卫化 + v1..v40全链重跑幂等测试(16过)

附:remote_bridge await 临时引用修(E0716)+ agentic emit 收敛 E0716 app_state 绑定修
This commit is contained in:
lxy
2026-08-05 22:10:32 +08:00
parent ec9f0bf1ea
commit 5667da6cf4
16 changed files with 1696 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/ai/agentic/mod.rs
+282 -32
View File
@@ -1,5 +1,6 @@
//! Agentic 循环 — 流式接收 → 工具执行 → 结果回传 LLM → 循环
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
@@ -21,7 +22,7 @@ use df_ai::context_helpers::{
// B 路线 Phase 1:plan_hint 接入主 loop——filter_tool_defs_planned 在 filter_tool_defs
// 收敛的扁平子集之上叠加 plan_hint 编排(并行组同批聚拢/顺序依赖源在前),供 LLM 看到
// 一份按编排意图排序的工具列表。feature flag PLANNING_ENABLED(false 默认关)门控接入。
use df_ai::intent::{filter_tool_defs, filter_tool_defs_planned, suggested_model_tier, IntentRecognizer};
use df_ai::intent::{filter_tool_defs, filter_tool_defs_planned, suggested_model_tier, IntentRecognizer, Intent};
use df_ai::coordinator::{Coordinator, ExecutionResult};
use df_ai::persona::PersonaRegistry;
use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider, MessageRole};
@@ -40,7 +41,7 @@ use df_ai::router::{
use df_storage::db::Database;
use df_storage::models::AiProviderRecord;
use crate::state::{AppState, LlmConcurrency};
use crate::state::{AppState, LlmConcurrency, ProviderAcquire};
use super::audit::process_tool_calls;
// compress_via_llm 已随压缩逻辑迁至 context_lifecycle.rs(maybe_auto_compress 内调用)。
@@ -79,6 +80,16 @@ pub const DEFAULT_MAX_AGENT_ITERATIONS: usize = 10;
/// partial_text）不重试——保文入库 + AiCompleted(incomplete=true) + 系统提示网络中断。
pub const DEFAULT_MAX_AGENT_RETRIES: usize = 3;
/// 防瞬时抖动(G4.1,2026-08-05):主 provider per-provider 信号量连续饱和阈值。
///
/// 候选循环内 `try_acquire_for_provider` 返 `Exhausted` 时该 provider 连续饱和计数 +1
/// (跨迭代累计);`Acquired` 时清零(已恢复)。链首(当前主 provider)连续 N 次饱和
/// → 候选链构建时降级到链尾,让 fallback 先试。
///
/// **动机**:单次瞬时饱和(其他对话标题/压缩短暂占满)不应让主 provider 被长期跳过。
/// N=3 对齐 DEFAULT_MAX_AGENT_RETRIES 量级,3 轮连续饱和才判持续过载,防抖动误降级。
pub const PROVIDER_SATURATION_DEMOTE_THRESHOLD: usize = 3;
/// 改进3 B 常量开关 KEYWORD_FALLBACK_ENABLED 已随压缩逻辑迁至 context_lifecycle.rs
/// (在该模块内部为私有 const,文档见 context_lifecycle.rs 顶部)。
///
@@ -176,6 +187,10 @@ pub const TOPIC_MARKER_GOAL_AWARE: bool = true;
/// 单点回退)。与 CIRCUIT_BREAKER_ENABLED 独立。
pub const STALL_BREAKER_ENABLED: bool = true;
/// 纯问候工具门(默认 true):纯问候/社交短文本消息 → LLM 可见工具置空(tools=None)，
/// 模型只能纯文本回复。false(回退)→ 走既有 intent 过滤(可能回全量),排障/对比用。
pub const PURE_GREETING_TOOL_GATE: bool = true;
/// G2 连续重复阈值(默认 3)。
///
/// 对齐 CIRCUIT_BREAKER_THRESHOLD=3:连续 3 轮判定签名重复足以判漂移。新范式(签名重复)比
@@ -297,7 +312,7 @@ pub mod command_lock;
pub(crate) mod helpers;
pub(crate) use helpers::try_continue_agent_loop;
pub(crate) use helpers::infer_goal_from_tool_calls;
pub(crate) use helpers::{tool_call_signature, is_repetitive_exploration};
pub(crate) use helpers::{tool_call_signature, is_repetitive_exploration, is_pure_greeting};
// ============================================================
// 单 Provider 流式结果 + fallback 辅助
@@ -578,19 +593,55 @@ pub(crate) async fn run_agentic_loop(
max_retries: usize,
start_iteration: usize,
model_override: Option<String>,
// F1 并发 epoch(2026-08-05):调用方 spawn 前 `per_conv.loop_epoch.fetch_add(1)` 后传入,
// loop 捕获自己的 epoch。退出 / Guard 复位 / emit 只认 owner(epoch 匹配),否则是旧 loop。
loop_epoch: u64,
) {
// generating 状态由 RAII guard 收敛复位(正常 exit 显式 reset;panic/异常 Drop 兜底)
let mut guard = GeneratingGuard::new(conv_id.clone(), app_handle.clone());
// F1 并发 epoch + F2 存活心跳:一次锁内取 per_conv 关键 Arc + 建 guard。
// guard.new 同步(transition 经 ConvStateStore 无 await),可在此锁内调。
// conv_read 不惰性建(与旧 stop_flag/notify 取用段一致):conv 已删(per_conv 无条目)
// → 旧 loop 直接退出(ConvStateStore 条目应已被 conversation_delete 同步清)。
let (stop_flag, notify, loop_epoch_arc, last_heartbeat, mut guard) = {
let session = session_arc.lock().await;
let Some(conv) = session.conv_read(&conv_id) else {
tracing::warn!(stale_conv = %conv_id, "[ai] loop 入口 conv 已删,退出");
return;
};
(
conv.stop_flag.clone(),
conv.notify.clone(),
conv.loop_epoch.clone(),
conv.last_heartbeat.clone(),
GeneratingGuard::new(conv_id.clone(), conv.loop_epoch.clone(), loop_epoch, app_handle.clone()),
)
};
let my_epoch = loop_epoch;
// 入口即报活(loop 存活证据)+ 陈旧检查:spawn 后已被新 loop 接管(epoch 已变)→ 立即退出。
last_heartbeat.store(df_types::now_millis() as u64, Ordering::SeqCst);
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
tracing::info!(
conv_id = %conv_id,
my_epoch,
current_epoch = loop_epoch_arc.load(Ordering::SeqCst),
"[ai] loop 入口即被新 loop 接管(epoch 不匹配),退出"
);
guard.disarm();
return;
}
// 治本:LLM 生成间隙向前端发 AiHeartbeat,防前端 watchdog 误断流。
// 后端 emit AiHeartbeat → useAiEvents.ts resetStreamWatchdog(convId) → per-conv timer 重置。
// 解决:LLM 重试循环(~60s)期间无事件到达前端 → watchdog 45s/90s 到期误杀。
// shutdown_tx 在函数作用域结束(Drop)时通知 heartbeat 任务退出。
// F2(2026-08-05):heartbeat_loop 同时更新 per_conv.last_heartbeat(loop 存活证据),
// 供 ai_chat_stop 兜底判断 loop 是否真死(长工具执行/流式/压缩期间心跳持续 → 豁免不误迁 Idle)。
let (_heartbeat_tx, mut heartbeat_rx) = tokio::sync::watch::channel(());
let hb_app = app_handle.clone();
let hb_conv_id = conv_id.clone();
let hb_heartbeat = last_heartbeat.clone();
tokio::spawn(async move {
heartbeat_loop(&mut heartbeat_rx, &hb_app, &hb_conv_id).await;
heartbeat_loop(&mut heartbeat_rx, &hb_app, &hb_conv_id, &hb_heartbeat).await;
});
// 入口桥接:loop 启动前确保 per_conv 存在(已存在则保留累积,不存在则建)。
@@ -685,6 +736,13 @@ pub(crate) async fn run_agentic_loop(
Ok(Ok((pc, cands, prov, kl))) => (pc, cands, prov, kl),
Ok(Err(ProviderResolveError::EnsureKeyFailed(msg))) => {
guard.reset().await;
// F1:旧 loop 不 emit(被新 loop 接管时,错误由 owner loop 负责)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
// G1.1(2026-08-05):入口早退也落库 user 消息(调用方已 push 内存),防重启/切会话丢末条 user。
// resolved_model 此处尚未算出,model 传 None(仅持久化 user 消息,无 assistant 消息可标)。
save_conversation(&session_arc, &db, &conv_id, None, None, true).await;
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: msg.clone(),
// ensure_resolved_key 失败 = key 缺失/钥匙串损坏,归 Auth
@@ -704,6 +762,12 @@ pub(crate) async fn run_agentic_loop(
// L1 补丁:provider 解析 30s 超时(DB list_all / keyring resolve 卡死)。
// 走 AiError 分支复位 generating,对齐 ensure_resolved_key 失败处理口径。
guard.reset().await;
// F1:旧 loop 不 emit(被新 loop 接管时,错误由 owner loop 负责)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
// G1.1(2026-08-05):入口超时早退也落库 user 消息(同上,model None)。
save_conversation(&session_arc, &db, &conv_id, None, None, true).await;
tracing::error!(conv_id = %conv_id, "[ai] provider 解析超时(30s),可能 DB/keyring 卡死");
let err_msg = "Provider 解析超时(30s),请检查数据库/钥匙串状态后重试".to_string();
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
@@ -733,6 +797,13 @@ pub(crate) async fn run_agentic_loop(
"[ai] 发起 LLM 请求"
);
// F1 并发 epoch:provider 解析(30s 窗口)期间被新 loop 接管 → 立即退出,
// 不执行 pre-loop 的 plan/coordinator 等冗余工作。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
// 主对话路由 — TaskRequirements(needs_tool_use=true)。
// 模态当前仅 Text(图像消息类型未实现,后续多模态接入时检测 Part/Image 追加 Vision)。
// select_model_id None(池空/无匹配)→ 兜底 default_model,行为与接入前一致。
@@ -790,6 +861,11 @@ pub(crate) async fn run_agentic_loop(
.unwrap_or_default()
};
let (intent, conf) = IntentRecognizer::recognize(&user_text);
// 纯问候护栏:纯社交短文本 → 工具全隐藏(tools=None),治弱模型把问候当指令擅自调工具
// (会话 b4d6b4e0 实证:"你好"→list_project_modules("DevFlow")+探索源码)。
// 机制非说教:空 tool_defs → provider tools=None → 模型物理上只能纯文本回复。
// 仅 Chat 意图 + is_pure_greeting 双条件(含动作词/实体引用放行,如"你好,看看moyu"正常暴露工具)。
let pure_greeting = PURE_GREETING_TOOL_GATE && matches!(intent, Intent::Chat) && is_pure_greeting(&user_text);
const INTENT_CONF_THRESHOLD: f32 = 0.7;
let all_defs = tools_arc.tool_definitions();
let total = all_defs.len(); // 提前记录全量数(all_defs 将 move 进 tool_defs)
@@ -807,7 +883,9 @@ pub(crate) async fn run_agentic_loop(
// Phase0a 就绪 true)与 PLANNING_ENABLED(planner.rs 主 loop 规划开关,本批仍是 false)
// 分离:即使将来 PLANNING_ENABLED 翻 true,plan_hint 内部 PLAN_HINT_ENABLED 关闭时
// filter_tool_defs_planned 仍退扁平(双层开关,任一关闭均退旧行为)。
let tool_defs = if conf >= INTENT_CONF_THRESHOLD {
let tool_defs: Vec<df_ai::provider::ToolDefinition> = if pure_greeting {
Vec::new() // 纯问候 → 空工具集(tools=None,强制纯文本回复)
} else if conf >= INTENT_CONF_THRESHOLD {
let filtered = if df_ai::planner::PLANNING_ENABLED {
// Phase 1:plan_hint 编排排序。intent_label 供 plan_hint 备用(当前规则纯关键词驱动)。
filter_tool_defs_planned(&all_defs, &intent, intent.as_str(), &user_text)
@@ -832,21 +910,8 @@ pub(crate) async fn run_agentic_loop(
planning_enabled = df_ai::planner::PLANNING_ENABLED,
"[ai] 意图收敛工具"
);
// 停止信号副本：stream_llm 与每轮迭代共享读取，避免重复加锁
// notify 同取一份 Arc 引用:stream_llm select! 监听 notified() 即时唤醒。
// 取 per_conv 的 stop_flag/notify。
// conv_id 来源:run_agentic_loop 入参(loop 启动快照,与 guard 一致)。
let (stop_flag, notify) = {
let session = session_arc.lock().await;
// panic-guard:原裸 .expect 在 conv 已删(并发删除/状态竞态)时 panic,
// guard 不 reset/终态不发/registry 不清致永久卡。改为 None 显式退出(对齐同函数其他 return 点;
// guard Drop 兜底复位 generating)。
let Some(conv) = session.conv_read(&conv_id) else {
tracing::warn!(stale_conv = %conv_id, "[ai] loop 入口 conv 已删,退出");
return;
};
(conv.stop_flag.clone(), conv.notify.clone())
};
// 停止信号副本 / notify / loop_epoch / last_heartbeat 已在 loop 入口(L593)一次锁内取用
// (stop_flag/notify 供 stream_llm 与每轮迭代共享,loop_epoch 供 owner 判定,last_heartbeat 供存活心跳)。
// token 累加器:loop 生命周期内各轮叠加,退出时传 save_conversation(累加模式落库)
let mut tokens = TokenAccumulator::default();
@@ -873,6 +938,12 @@ pub(crate) async fn run_agentic_loop(
// DeepSeek thinking 模式推理内容跨轮透传
let mut last_reasoning_content: Option<String> = None;
// G4.3(2026-08-05):记录本轮 token 用量是否估算值(round_usage.prompt_tokens==0 →
// estimated_prompt 兜底处打标)。loop 内各轮更新,退出路径(MidStream/stop 等在 loop 内
// 直接用 round_usage 判;max_iterations/正常完成在 loop 后用本字段)透传 AiCompleted 仅展示,
// 不参与 TokenAccumulator/save 累加。
let mut last_round_estimated: bool = false;
// G1 目标钉扎:入口把 PerConvState.pinned_goals 拼进 system_prompt 尾部(一次拼好整个 loop 复用)。
//
// 治 R1(目标消息被压缩出局)/R5(prompt 说教无锚点):system_prompt 是 loop 不变量 + build_for_request
@@ -1156,9 +1227,12 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &llm_concurrency,
&mut guard,
&coord_usage,
// Coordinator 路径无 round(未 stream),无估算语义 → false。
false,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
&loop_epoch_arc, my_epoch,
).await;
return;
}
@@ -1176,11 +1250,18 @@ pub(crate) async fn run_agentic_loop(
// retry 同 loop 内,持 per_conv 合理。
let _conv_per_conv_permit = llm_concurrency.acquire_per_conv(&conv_id).await;
// 防瞬时抖动(G4.1):per-provider 连续饱和计数(跨迭代累计)。
// 候选循环内 Exhausted +1 / Acquired 清零;链首(主 provider)连续
// PROVIDER_SATURATION_DEMOTE_THRESHOLD 次饱和 → 候选链构建时降级到链尾(见链构建处)。
// 仅 per-provider 层非阻塞降级,不影响 global/per_conv 阻塞限流。
let mut provider_saturation: HashMap<String, usize> = HashMap::new();
// 0 = 不限:effective_max=usize::MAX,for 到不了上界,靠 stop_flag/收敛/审批退出(下方达上限暂停分支不触发)
let effective_max = if max_iterations == 0 { usize::MAX } else { max_iterations };
for iteration in start_iteration..effective_max {
// 用户请求停止 → 收尾退出（已生成文本已在上一轮入库）
if stop_flag.load(Ordering::SeqCst) {
// 用户请求停止 或 已被新 loop 接管(stale)→ 收尾退出(已生成文本已在上一轮入库)。
// F1:stale 判定(epoch 不匹配)让旧 loop 在新 loop 启动后立即在此退出,不再继续跑。
if stop_flag.load(Ordering::SeqCst) || loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
@@ -1191,6 +1272,7 @@ pub(crate) async fn run_agentic_loop(
};
// 入口 stop:本轮可能尚未 stream(首轮即停),不记 model——避免把未实际生成的 model 写入 models 数组
// 统一走 finish_round_exit:save(Some usage, None model) + spawn_title + emit(None,None,publish=true)
// (stale 时 finish_round_exit 内部仅 disarm 跳过 save/reset/emit,不干扰新 loop)
finish_round_exit(
&session_arc, &db, &conv_id,
Some(&usage), None,
@@ -1198,9 +1280,12 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &llm_concurrency,
&mut guard,
&usage,
// 入口 stop:本轮尚未 stream,last_round_estimated 仍为初始 false(无估算)。
last_round_estimated,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
&loop_epoch_arc, my_epoch,
).await;
return;
}
@@ -1293,6 +1378,13 @@ pub(crate) async fn run_agentic_loop(
// per_conv.iteration_used 唯一真相源。
let conv = session.conv(&conv_id);
conv.iteration_used = iteration + 1;
// F1 并发 epoch:每轮顶部 stale 检查(防旧 loop 在 topic/compress 等长段期间被接管后继续跑)。
// F2 存活心跳:每轮更新(与 heartbeat_loop 20s 心跳互补,报活更即时)。
if conv.loop_epoch.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
conv.last_heartbeat.store(df_types::now_millis() as u64, Ordering::SeqCst);
}
// 新一轮通知前端（第二轮起），前端需新建 assistant 消息
@@ -1463,6 +1555,30 @@ pub(crate) async fn run_agentic_loop(
let mut candidate_chain: Vec<AiProviderRecord> = Vec::with_capacity(1 + candidates.len());
candidate_chain.push(provider_config.clone());
candidate_chain.extend(candidates.iter().cloned());
// 防瞬时抖动(G4.1):连续 PROVIDER_SATURATION_DEMOTE_THRESHOLD 次饱和才把主 provider
// 降级到链尾。计数器在候选循环内维护(Exhausted +1 / Acquired 清零),跨迭代累计。
// 链首(当前主 provider)连续 N 次 Exhausted → 移到链尾让 fallback 先试;
// 恢复后 Acquired 清零 → 下轮回到链首。单次/瞬时饱和(N 次内)不降级,防跳过主 provider。
// first():链构建总是 push 至少 1 个(provider_config),但防御性取首防空链 panic。
if let Some(streak) = candidate_chain
.first()
.and_then(|c| provider_saturation.get(&c.id))
.copied()
.filter(|s| *s >= PROVIDER_SATURATION_DEMOTE_THRESHOLD)
{
let saturated = candidate_chain.remove(0);
candidate_chain.push(saturated.clone());
tracing::info!(
conv_id = %conv_id,
provider = %saturated.name,
streak = streak,
threshold = PROVIDER_SATURATION_DEMOTE_THRESHOLD,
"[ai] 主 provider 连续 {} 次 per-provider 饱和,降级到候选链尾(防抖动跳过)",
streak,
);
}
let (full_text, tool_calls_acc, round_usage, incomplete, round_reasoning_content) = {
// outcome 累积成功结果(Complete/Partial),InitFailed 不写入。
let mut outcome: Option<(String, std::collections::HashMap<u32, super::ToolCallDraft>, df_ai::provider::TokenUsage, bool, Option<String>)> = None;
@@ -1472,10 +1588,33 @@ pub(crate) async fn run_agentic_loop(
let mut last_exhausted_error: Option<String> = None;
'candidate: for candidate in &candidate_chain {
// per-provider permit(可选):set_provider_caps 未配置时返回 None(单 provider 零变化);
// 配置后取额外 permit 防单 provider 被打满(限流 429)。切换 candidate 时上一 permit
// 随 _provider_permit 绑定作用域 Drop 释放。
let _provider_permit = llm_concurrency.acquire_for_provider(&candidate.id).await;
// per-provider permit(非阻塞三态,G4.1):
// NotConfigured(未配置,无限流)→ 无 permit 直接 proceed(单 provider 零变化);
// Acquired(permit)→ 持 permit stream,切换 candidate 时随作用域 Drop 释放;
// Exhausted(占满)→ 连续饱和计数 +1,跳下一 candidate(降级,不阻塞)。
// 旧 acquire_for_provider(阻塞 acquire_owned)在主 provider 信号量占满时整条
// fallback 链卡死(降级失效),已由本三态非阻塞版本取代。
let _provider_permit = match llm_concurrency.try_acquire_for_provider(&candidate.id).await {
ProviderAcquire::NotConfigured => None,
ProviderAcquire::Acquired(permit) => {
// 成功拿到 permit:该 provider 已恢复,连续饱和计数清零(跨迭代防抖)。
provider_saturation.insert(candidate.id.clone(), 0);
Some(permit)
}
ProviderAcquire::Exhausted => {
// 占满:连续饱和计数 +1;本次跳下一 candidate(非阻塞,不等待)。
// 跨迭代防抖:连续 N 次饱和后链首降级(见候选链构建处),防瞬时抖动跳过主 provider。
*provider_saturation.entry(candidate.id.clone()).or_insert(0) += 1;
tracing::debug!(
conv_id = %conv_id,
provider = %candidate.name,
streak = provider_saturation.get(&candidate.id).copied().unwrap_or(0),
"[ai] 候选 {} per-provider 信号量占满,跳过(非阻塞,防单 provider 429)",
candidate.name,
);
continue 'candidate;
}
};
let stream_outcome = stream_one_provider(
candidate,
@@ -1522,7 +1661,15 @@ pub(crate) async fn run_agentic_loop(
// Fatal(4xx 非429/鉴权/参数错):立即放弃整轮 fallback。
// 统一 emit 最终错误气泡(单气泡聚合)。
guard.reset().await;
// F1:旧 loop(被新 loop 接管)不 emit 错误气泡(owner loop 负责呈现)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
emit_fatal_error(&app_handle, &conv_id, &error).await;
// G1.1(2026-08-05):Fatal 退出也落库 user 消息(镜像下方 Exhausted 分支),
// 否则重启/切会话从 DB 恢复时末条 user 丢失(与 Exhausted 同根因不对称)。
// epoch 校验通过后、return 前补 save(owner loop 才写,防 owner 已转移仍写)。
save_conversation(&session_arc, &db, &conv_id, None, Some(&resolved_model), true).await;
return;
}
}
@@ -1538,6 +1685,9 @@ pub(crate) async fn run_agentic_loop(
"[ai] 全 provider 候选流前失败重试耗尽,放弃本轮",
);
guard.reset().await;
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
let err_msg = last_exhausted_error
.unwrap_or_else(|| "AI 调用失败：所有候选 provider 重试耗尽".to_string());
emit_fatal_error(&app_handle, &conv_id, &err_msg).await;
@@ -1549,6 +1699,17 @@ pub(crate) async fn run_agentic_loop(
}
};
// G4.3:本轮 token 用量是否估算值(provider 未报 prompt_tokens → estimated_prompt 兜底),
// 供 loop 内/loop 后各退出路径透传 AiCompleted(is_estimated)仅作展示标注。
last_round_estimated = round_usage.prompt_tokens == 0;
// F1 并发 epoch:流返回后若已被新 loop 接管(force_send 等),旧 loop 不再 push 消息 /
// 保文 / 执行工具,立即退出(guard disarm 跳过复位,防 clobber 新 loop 的 Generating)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
// CR-30-2 / UX-2025-04 / 决策 a1: MidStream 保文路径——partial_text 已接收,
// 入库为正常 assistant 消息 + emit AiCompleted(incomplete=true) + 追加系统提示消息。
// 不走 AiError(非异常中断,已有可用文本),不重试(决策 a1)。
@@ -1580,9 +1741,11 @@ pub(crate) async fn run_agentic_loop(
);
return;
}
// 与 push_assistant_message 一致:trim 归一化,纯空白保文(网络中断只有空白)不落库
let full_text = full_text.trim();
if !full_text.is_empty() {
let conv = session.conv(&conv_id);
let mut msg = ChatMessage::assistant(&full_text);
let mut msg = ChatMessage::assistant(full_text);
msg.model = Some(resolved_model.clone());
// MidStream 保文也回填 reasoning_content
msg.reasoning_content = round_reasoning_content.clone();
@@ -1622,9 +1785,11 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &llm_concurrency,
&mut guard,
&emit_usage,
round_usage.prompt_tokens == 0,
Some(true), None, true,
&pinned_goals_snapshot,
&app_handle,
&loop_epoch_arc, my_epoch,
).await;
return;
}
@@ -1689,8 +1854,9 @@ pub(crate) async fn run_agentic_loop(
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&resolved_model), true).await;
}
// 停止信号：已生成文本入库后退出，不再执行后续工具调用
if stop_flag.load(Ordering::SeqCst) {
// 停止信号 或 已被新 loop 接管(stale):已生成文本入库后退出,不再执行后续工具调用。
// F1:stale 时旧 loop 在此退出,不执行工具(防重复工具执行);finish_round_exit 内部跳过 emit。
if stop_flag.load(Ordering::SeqCst) || loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
@@ -1707,9 +1873,11 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &llm_concurrency,
&mut guard,
&usage,
round_usage.prompt_tokens == 0,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
&loop_epoch_arc, my_epoch,
).await;
return;
}
@@ -1729,6 +1897,13 @@ pub(crate) async fn run_agentic_loop(
"[AI-DIRAUTH-DIAG] agentic loop 收到 pending"
);
// F1 并发 epoch:工具执行期间被新 loop 接管(force_send)后立即退出,
// 不熔断/不停审批/不进下一轮(避免旧 loop 继续污染新 loop 会话)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
// L1 断路器:连续同类工具失败熔断(治 agent 无止损死循环,机制非 prompt 说教)。
// count_recent_failures 读末尾连续 Tool 消息,失败包按结构化信号归一 key 入滚动窗口计数。
if CIRCUIT_BREAKER_ENABLED {
@@ -1742,6 +1917,10 @@ pub(crate) async fn run_agentic_loop(
"[ai] L1 断路器熔断:连续同类失败 {} 次,疑似死循环停止", max_count
);
guard.reset().await;
// F1:旧 loop 不 emit 熔断求助卡(owner loop 负责呈现)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
emit_circuit_breaker_tripped(&app_handle, &conv_id, max_count, &sample_key);
return;
}
@@ -1764,6 +1943,11 @@ pub(crate) async fn run_agentic_loop(
// 有待审批 → 暂停循环，等待用户审批后通过 ai_approve → try_continue_agent_loop 恢复
if pending_count > 0 {
// F1:旧 loop 被接管后不停审批(force_send 已清 pending),直接退出。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
@@ -1813,14 +1997,22 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &llm_concurrency,
&mut guard,
&usage,
last_round_estimated,
Some(true), None, false,
&pinned_goals_snapshot,
&app_handle,
&loop_epoch_arc, my_epoch,
).await;
return;
}
// 正常完成
// F1 并发 epoch:正常完成前若已被新 loop 接管,不 spawn 后台 save/标题/提炼、不 reset/emit
// (owner loop 负责收尾),直接退出。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
@@ -1853,6 +2045,12 @@ pub(crate) async fn run_agentic_loop(
});
}
// F1 并发 epoch:emit 前 final stale 检查(防上方后台 spawn 块期间被 force_send 接管后仍 emit
// 重复 AiCompleted)。guard.reset 本身 epoch-aware(stale 跳过迁移),此处主要防 emit。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
guard.disarm();
return;
}
guard.reset().await;
// generating 复位后再 emit Completed：落库/标题/提炼已在后台，前端立即感知完成
// (正常完成路径 save+extract+title 已在上方 spawn 异步,此处仅 emit,故直接调 emit_ai_completed_once)
@@ -1865,8 +2063,13 @@ pub(crate) async fn run_agentic_loop(
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
// F1 并发 epoch:emit 前最后一次 stale 检查(guard.reset 已幂等,此处防 emit 重复 AiCompleted)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return;
}
emit_ai_completed_once(
&app_handle, &conv_id, &normal_usage,
last_round_estimated,
None, None, true,
&pinned_goals_snapshot,
).await;
@@ -1874,15 +2077,19 @@ pub(crate) async fn run_agentic_loop(
// ── heartbeat_loop: 后台心跳任务(emit AiHeartbeat 防前端 watchdog 误杀) ──
// 退出条件:heartbeat_rx 发送端 Drop(function 作用域结束)。每 20s emit 一次。
// F2(2026-08-05):同时更新 per_conv.last_heartbeat(loop 存活证据)——loop 全程(含流式/工具
// 执行/压缩/provider 解析)每 20s 刷新,供 ai_chat_stop 兜底判断 loop 是否真死。
async fn heartbeat_loop(
heartbeat_rx: &mut tokio::sync::watch::Receiver<()>,
app: &AppHandle,
conv_id: &str,
last_heartbeat: &std::sync::Arc<std::sync::atomic::AtomicU64>,
) {
loop {
tokio::select! {
_ = heartbeat_rx.changed() => break,
_ = tokio::time::sleep(std::time::Duration::from_secs(20)) => {
last_heartbeat.store(df_types::now_millis() as u64, std::sync::atomic::Ordering::SeqCst);
let _ = app.emit(
"ai-chat-event",
AiChatEvent::AiHeartbeat { conversation_id: Some(conv_id.to_string()) },
@@ -2295,16 +2502,33 @@ async fn emit_fatal_error(app_handle: &AppHandle, conv_id: &str, error: &str) {
// usage 字段(对齐 run_agentic_loop 各退出点用法):emit 端的 token 三元组从 `usage` 取;
// prompt/completion 与 usage 一致(各退出点原样从 tokens 或 round-derived usage 传入)。
//
// is_estimated(G4.3, 2026-08-05):本轮 token 用量是否估算值(round_usage.prompt_tokens==0 →
// estimated_prompt 处打标)。透传前端仅作展示标注,不参与 TokenAccumulator/save 累加。
//
// G2.2(2026-08-05):本 helper 是 run_agentic_loop 所有正常出口的统一 emit 点——
// AiCompleted 发出时同步清该 conv 的审批残留(终态化占位 + 移除 pending_approvals),
// 防占位 tool_result 残留 messages 下次发送喂给 LLM。仅清目标 conv(retain 保留其他 conv)。
//
// pinned_goals:前端直接读取刷新(G1 目标钉扎),原样透传快照(不 clone,借用调用方)。
async fn emit_ai_completed_once(
app_handle: &AppHandle,
conv_id: &str,
usage: &df_ai::provider::TokenUsage,
is_estimated: bool,
emit_incomplete: Option<bool>,
publish_incomplete: Option<bool>,
do_publish: bool,
pinned_goals: &[super::GoalEntry],
) {
// G2.2:清审批残留收敛此处(所有正常出口都经本 helper 单点 emit)。锁内纯写改短 lock 段。
{
// app_handle.state() 返回临时引用,直接 .ai_session.lock() 借用临时值 → E0716 悬垂。
// 先绑定 state(守卫整块生命周期),再经 DerefMut 方法调用。
let app_state = app_handle.state::<AppState>();
let mut session_guard = app_state.ai_session.lock().await;
super::commands::chat::finalize_pending_placeholders(&mut *session_guard, conv_id, "生成完成");
session_guard.pending_approvals.retain(|_, a| a.conversation_id.as_deref() != Some(conv_id));
}
// emit 端(前端通道):incomplete 用 emit_incomplete(MidStream 传 Some(true))。
// token 分项(2026-08-02):cache_hit/cache_miss/reasoning 透传前端分计费展示。
let _ = app_handle.emit(
@@ -2316,6 +2540,7 @@ async fn emit_ai_completed_once(
prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens,
reasoning_tokens: usage.reasoning_tokens,
is_estimated,
incomplete: emit_incomplete,
conversation_id: Some(conv_id.to_string()),
pinned_goals: pinned_goals.to_vec(),
@@ -2332,6 +2557,7 @@ async fn emit_ai_completed_once(
prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens,
reasoning_tokens: usage.reasoning_tokens,
is_estimated,
incomplete: publish_incomplete,
conversation_id: Some(conv_id.to_string()),
pinned_goals: pinned_goals.to_vec(),
@@ -2359,8 +2585,11 @@ async fn emit_ai_completed_once(
// 参数口径(对齐各退出点原代码):
// - save_usage/save_model:透传 save_conversation;Coordinator 传 (None, None) 仍会调 save(幂等)
// - emit_usage:emit AiCompleted 的 token 三元组来源(各退出点经 tokens.add/round-derived 后传入)
// - emit_estimated(G4.3, 2026-08-05):本轮 token 用量是否估算值,透传 emit_ai_completed_once 仅展示
// - spawn_title:true → spawn_ensure_title(provider_config 后台生成,失败有 extract_title 兜底)
// - emit_incomplete/publish_incomplete/do_publish:透传 emit_ai_completed_once(见该 helper 文档)
// - loop_epoch_arc/my_epoch:F1 并发 epoch——旧 loop(epoch 不匹配)只 disarm 跳过 save/reset/emit,
// 不干扰新 loop 收尾(防「双完成 + 双 AiCompleted + guard 互踩」)。
async fn finish_round_exit(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
@@ -2372,12 +2601,27 @@ async fn finish_round_exit(
llm_concurrency: &LlmConcurrency,
guard: &mut GeneratingGuard,
emit_usage: &df_ai::provider::TokenUsage,
emit_estimated: bool,
emit_incomplete: Option<bool>,
publish_incomplete: Option<bool>,
do_publish: bool,
pinned_goals: &[super::GoalEntry],
app_handle: &AppHandle,
loop_epoch_arc: &std::sync::Arc<std::sync::atomic::AtomicU64>,
my_epoch: u64,
) {
// F1 并发 epoch 收尾守卫:旧 loop(被 force_send/新 loop 接管,epoch 不匹配)不 save/reset/emit,
// 只 disarm guard 退出——owner loop 负责收尾,防「双完成 + 双 AiCompleted + guard 互踩」。
if loop_epoch_arc.load(std::sync::atomic::Ordering::SeqCst) != my_epoch {
tracing::debug!(
conv_id = %conv_id,
my_epoch,
current_epoch = loop_epoch_arc.load(std::sync::atomic::Ordering::SeqCst),
"[ai] 旧 loop 收尾跳过(已被新 loop 接管,不 save/reset/emit)"
);
guard.disarm();
return;
}
// 落库:save_conversation 幂等(每轮重复覆盖落库),Coordinator 传 (None,None) 也调用(对齐原 :1126)
save_conversation(session_arc, db, conv_id, save_usage, save_model, true).await;
// 标题生成后台化(不阻塞 Completed emit):Coordinator / max_iterations 不 spawn(对齐原行为)
@@ -2388,7 +2632,7 @@ async fn finish_round_exit(
guard.reset().await;
// generating 复位后再 emit Completed(对齐原各路径顺序)
emit_ai_completed_once(
app_handle, conv_id, emit_usage,
app_handle, conv_id, emit_usage, emit_estimated,
emit_incomplete, publish_incomplete, do_publish,
pinned_goals,
).await;
@@ -2415,6 +2659,12 @@ fn push_assistant_message(
cache_miss: u32,
reasoning: u32,
) {
// 根治「空工具轮 assistant 消息落库」:LLM 仅返回 tool_calls 无文本时 full_text 可能为
// 空/纯空白(assistant("")/assistant("\n") 均合法落库),前端渲染空气泡。入口统一 trim:
// - 纯空白 → content 归并为 ""(非 whitespace),前端 !content 判断可命中;有 tool_calls 仍 push(工具卡有效)
// - 纯空白且无 tool_calls → 下方 !full_text.is_empty() 为 false,跳过 push(空轮无任何信息不落库)
// - 真实文本仅去首尾空白,不影响内容
let full_text = full_text.trim();
if has_tool_calls {
let mut order: Vec<u32> = tool_calls_acc.keys().copied().collect();
order.sort_unstable();