Files
DevFlow/src-tauri/src/commands/ai/agentic/mod.rs
T
lxy a031521776 优化: token分项显示(in/cache/out/reasoning)+ 详情面板 + base前置
token分项(各计费不同,不显 total):df-ai 解析 provider cache/reasoning(openai_compat prompt_cache_hit/miss/reasoning_tokens + anthropic cache_read/creation)+ TokenUsage 加字段(全构造点)+ AiMessage/AiCompleted/DB V39(ai_messages 加 cache_hit/miss/reasoning 列)+ message_repo 映射(持久化)+ 前端 MessageList 显 in·cache·out·reason(in=cache_miss 全价,reasoning 有才显)+ 点击 token 弹详情面板(完整 usage+缓存命中率+model)+ df-miniapp 同步

base前置(提升 prompt cache 命中率):chat.rs aug 拼 base 后(4处)+ knowledge_inject 知识拼 base 后(固定 base 前缀,cache 命中)

附修:replace_conversation 原 13 列 INSERT 丢消息级 token → 改 18 列
2026-08-03 01:22:30 +08:00

2562 lines
135 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! Agentic 循环 — 流式接收 → 工具执行 → 结果回传 LLM → 循环
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tauri::{AppHandle, Emitter, Manager};
use tokio::sync::Mutex;
use df_ai::ai_tools::AiToolRegistry;
use df_ai::context::TokenEstimator;
// 占位配对完整性:agentic 出口第二道防线断言(深度防御)。
use df_ai::context::ContextManager;
// 改进3 B: 压缩失败兜底关键词摘要(纯函数 extract_keyword_summary)。
// 改进4: 工具结果 view-only 摘要(should_summarize_tool_result / extract_key_info)。
// extract_keyword_summary 随压缩逻辑迁至 context_lifecycle.rs(改进3 B 关键词兑底)。
use df_ai::context_helpers::{
extract_key_info, should_summarize_tool_result,
PLACEHOLDER_INTEGRITY_ENABLED,
};
// 改进2 B:意图收敛工具(LLM 可见 tool_defs 按 intent 过滤,执行路径仍走完整 registry)
// 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::coordinator::{Coordinator, ExecutionResult};
use df_ai::persona::PersonaRegistry;
use df_ai::provider::{ChatMessage, CompletionRequest, LlmProvider, MessageRole};
// 复用 retry::backoff_delay(jitter 1s→2s→4s) + is_status_retryable(Fatal 分类)
// 实现流前失败重试退避对齐,避免重写退避逻辑。
use df_ai::retry;
// 智能路由 helper + TaskRequirements + 维度枚举。
// 调用点构造 TaskRequirements(主对话:needs_tool_use=true,
// 当前仅 Text 模态;后续多模态接入时检测消息内 Part/Image 追加 Vision),
// 经 select_model_id 在 provider.model_configs 池中选最优;池空/无匹配兜底 default_model。
// 注:路由已解耦,cost_tier/intelligence 不再参与硬路由。
use df_ai::router::{
select_model_id, Modality, TaskRequirements,
};
use df_storage::db::Database;
use df_storage::models::AiProviderRecord;
use crate::state::{AppState, LlmConcurrency};
use super::audit::process_tool_calls;
// compress_via_llm 已随压缩逻辑迁至 context_lifecycle.rs(maybe_auto_compress 内调用)。
use super::conversation::{save_conversation, TokenAccumulator};
use super::knowledge_inject::{maybe_spawn_extraction};
use super::stream_recv::{stream_llm, StreamResult};
use super::title::{ensure_conversation_title, spawn_ensure_title};
use super::{AiChatEvent, AiSession, ErrorType};
// ConvState 经本文件内 `pub mod conv_state;` 同 crate 直接访问(conv_state::ConvState)。
/// L1 补丁:run_agentic_loop 入口 provider 解析超时保护的内部错误类型。
///
/// 用于把 provider 解析块(list_all + select + resolve + ensure + build)包入
/// `tokio::time::timeout` 后的内部 Err 路由——区分 ensure_resolved_key 失败(走 Auth
/// 错误)与整体超时(走 Unknown 错误)。超时由外层 timeout 的 Err(Elapsed) 单独匹配。
enum ProviderResolveError {
/// ensure_resolved_key 失败(key 缺失/钥匙串损坏):走 Auth 错误分支(对齐原 :412 处理)。
EnsureKeyFailed(String),
}
/// Agentic 循环默认最大迭代次数(可配置项的默认值)
///
/// 默认 10 轮。已接入配置:AppState.agent_max_iterations(Arc<AtomicUsize>) +
/// ai_set_agent_max_iterations command + Settings.vue 数字配置。调用方在 loop 入口
/// load AtomicUsize 快照后透传 `max_iterations: usize` 形参,当前 loop 锁定边界,
/// 热改下次发消息生效(与 llm_concurrency 传 Arc 实时反映的区别)。
pub const DEFAULT_MAX_AGENT_ITERATIONS: usize = 10;
/// 压缩保护区条数 + 压缩关键词兑底开关已随压缩逻辑迁至 context_lifecycle.rs
/// (PROTECT_COUNT / KEYWORD_FALLBACK_ENABLED,在该模块内部为私有 const)。
///
///
/// 默认 3 次(初次 + 2 次重试)。复用 retry::backoff_delay 退避(1s→2s→4s+jitter) +
/// retry::is_status_retryable Fatal 分类(4xx 非429 立即放弃) + 30s 总预算。
/// 只重试流前失败(Init Err:未输出任何 token);流中途失败(MidStream:已输出
/// partial_text)不重试——保文入库 + AiCompleted(incomplete=true) + 系统提示网络中断。
pub const DEFAULT_MAX_AGENT_RETRIES: usize = 3;
/// 改进3 B 常量开关 KEYWORD_FALLBACK_ENABLED 已随压缩逻辑迁至 context_lifecycle.rs
/// (在该模块内部为私有 const,文档见 context_lifecycle.rs 顶部)。
///
/// 改进4 常量开关:tool_result 大输出是否做 view-only 摘要压缩(默认 true)。
///
/// true(默认):build_for_request 后送 stream 前,遍历 history 中 tool_result,
/// 超阈值(content >2KB 或占历史 token >40%)的 content 应用 extract_key_info
/// 替换(保留错误行 + 首尾各 5 行)。**仅 messages clone 视图,不改 ContextManager
/// 持久化**(对齐 sanitize_messages:DB 原始 tool_result 完整保留)。
/// false(关闭):tool_result 原样送 LLM(旧行为)。排障/对比用。
pub const TOOL_RESULT_COMPRESS_ENABLED: bool = true;
/// 改进5 常量开关:是否启用主题切换系统标记(默认 true)。
///
/// true(默认):push 时末两条 user 消息 topic 都非 None 且不同(双高置信)→
/// pending_topic_marker 置位 → agentic loop 顶部读取并 insert 一条 system 软提示
/// `"── 用户已切换话题(从「{old}」到「{new}」),请以新话题为准 ──"`(软提示,不强制 LLM)。
/// false(关闭):loop 顶部跳过读取/insert(置 false 即可观察无主题标记效果)。
/// 保守:双高置信才标(任一 topic None 不标),不强制 LLM(软提示非硬约束)。
pub const TOPIC_MARKER_ENABLED: bool = true;
// ============================================================
// G1 目标钉扎(治 R1 目标消息被压缩出局 + R3 重锚定 + R5 prompt 无锚点,2026-06-26)
//
// 机制:run_agentic_loop 每轮从 LLM 工具调用推理目标(infer_goal_from_tool_calls),
// 存 PerConvState.pinned_goals(Vec<GoalEntry>,内容态字段,绕过 sanitize step0 is_active 过滤),
// run_agentic_loop 入口拼 active 目标进 system_prompt 尾部。
// system_prompt 是 loop 不变量 + build_for_request 不裁剪,故目标天然免疫压缩/裁剪,
// 彻底治 R1(目标消息物理出局)/R5(prompt 说教无锚点),无 insert_at(0) 的连续 System
// 1214/首位锚点稀释/小预算被裁三重风险。
// ============================================================
/// G1 总开关:目标钉扎是否启用(默认 true)。
///
/// true(默认):run_agentic_loop 每轮从 LLM 工具调用推理目标(infer_goal_from_tool_calls),
/// 存入 PerConvState.pinned_goals(Vec<GoalEntry>) + loop 入口拼 active 目标进 system_prompt。
/// false(回退):提取跳过 + loop 入口拼接跳过,pinned_goals 永远空 Vec,system_prompt 零变化,
/// 完全退回改动前行为(目标靠压缩摘要 + topic marker + 聚焦准则 prompt 续命)。单点回退,
/// 不影响 G2/G4/G5(对齐 KEYWORD_FALLBACK_ENABLED 模式:每改配开关 + 兜底降级旧行为)。
/// 2026-07-03 改:提取从 chat.rs 用户消息规则 → run_agentic_loop 工具调用推理(helpers.rs)。
pub const GOAL_PIN_ENABLED: bool = true;
/// G1 banner 开关:目标拼进 system_prompt 时是否加「## 当前目标」分隔标题(默认 true)。
///
/// true(默认):目标文本前置 `## 当前目标(全程锚定,所有动作须服务于它)` 标题 + 目标内容,
/// 显式分隔防与行为准则混淆,提升 LLM 注意力分配。false(裸拼):目标文本直接 append 到
/// system_prompt,无标题分隔(紧凑,排障/对比用)。
pub const GOAL_INJECT_BANNER: bool = true;
/// G1 截断长度:每条目标文本截断上限(默认 500 字符)。
///
/// 防 R1 反向风险:长 user 消息(粘贴需求文档/长 bug 描述)每轮占 system_prompt 预算。
/// system_prompt 虽不被裁剪但仍计 sys_tokens 占预算,故截断防长目标撑爆。500 保守(首版,
/// 可调),足够覆盖正常一句话目标。截断后追加「…」省略号标识。
pub const GOAL_MAX_CHARS: usize = 500;
/// G1 多目标上限:最多累积目标数(默认 5)。
///
/// 防无限膨胀:每次发消息提取的目标追加到 pinned_goals vec,超上限时淘汰最早目标。
pub const MAX_GOALS: usize = 5;
/// G4 目标感知降级:话题标记 insert 当 pinned_goals 存在时跳过(默认 true)。
///
/// 治 R2(话题标记反向误导):G1 目标钉扎生效后每轮 system_prompt 已含目标,topic marker 的
/// 「请以新话题为准」软提示成冗余且与目标矛盾(诊断 §三双锚点稀释)。true(默认)= 当
/// pinned_goal 存在时 take_topic_marker 后丢弃跳过 insert(防 marker 累积 + 消双锚点稀释);
/// false(回退)= 无视 pinned_goal 照常 insert,完全退回原双锚点行为(排障/对比用)。
///
/// 与 TOPIC_MARKER_ENABLED 独立:TOPIC_MARKER_ENABLED 是整段总开关(false=整段跳过),
/// TOPIC_MARKER_GOAL_AWARE 是 goal 感知前置 guard(仅 GOAL_AWARE=true 且 pinned_goal
/// 存在时跳过 insert)。双关默认 true,各自单点回退不耦合(G1 关闭时 GOAL_AWARE 仍生效,
/// 但 pinned_goal 永远 None → 不跳过 → 退原 topic 行为,双层兜底)。
pub const TOPIC_MARKER_GOAL_AWARE: bool = true;
// ============================================================
// G2 探索熔断(治 R4 探索无进展死循环 + token 失控)
//
// 判定范式(2026-08-01 根本性重构):
// - 旧(废弃):连续 N 轮工具结果全空(is_empty_tool_result 关键词判空成功)。错误代理指标,
// grep 无匹配是有效排除信号非漂移,误熔断正常排除式搜索(实证 ac448296)。
// - 新(当前):连续 N 轮工具调用签名重复(is_repetitive_exploration 判 tool_call_signature)。
// 漂移本质 = AI 卡住反复同样调用;正常探索(换词/换路径/换工具)签名不同,真死循环签名重复。
//
// 机制(非 prompt 说教,骨架沿用):连续 N 轮判定签名重复 → 两段式(WARN_FIRST 先警示再熔断):
// 警示注入下轮 prompt 给 LLM 自纠机会,熔断(guard.reset + AiHelpRequired)逼用户换思路/人工介入。
// 与 L1 断路器(CIRCUIT_BREAKER_*)互补:L1 治反复同类失败(失败信号归一 key),
// G2 治反复同类调用(调用签名归一),两者独立计数互不干扰。
// ============================================================
/// G2 总开关:探索熔断是否启用(默认 true)。
///
/// true(默认):process_tool_calls 后取最近 N 个 assistant tool_calls 签名,喂
/// is_repetitive_exploration 判重复,重复累计 stall_count,达 STALL_BREAKER_THRESHOLD → 警示/熔断。
/// false(回退):整段跳过,stall_count 永远 0,退 max_iterations 旧行为(可能继续游荡但行为不变,
/// 单点回退)。与 CIRCUIT_BREAKER_ENABLED 独立。
pub const STALL_BREAKER_ENABLED: bool = true;
/// G2 连续重复阈值(默认 3)。
///
/// 对齐 CIRCUIT_BREAKER_THRESHOLD=3:连续 3 轮判定签名重复足以判漂移。新范式(签名重复)比
/// 旧范式(结果空)更精准——签名重复是漂移的直接证据,无关键词漏判风险。阈值 3 在
/// 两段式 WARN_FIRST(== THRESHOLD-1 即 2 轮先警示)容错下,误杀面收敛到「连续 3 轮同一组调用」
/// 的真实死循环场景,最坏退 max_iterations 兜底。
pub const STALL_BREAKER_THRESHOLD: u32 = 3;
/// G2 两段式开关:熔断前是否先警示一轮(默认 true)。
///
/// true(默认):stall_count == THRESHOLD-1 时注入「⚠ 检测到重复的工具调用 ...」到下轮
/// system_prompt(给 LLM 自纠机会);stall_count >= THRESHOLD 才熔断。
/// 降误杀:偶发的相似调用(如分段读后回读同一处确认)先警示再熔断。false(激进):直接熔断不警示,
/// 确认无误杀场景用。
pub const STALL_BREAKER_WARN_FIRST: bool = true;
/// G2 警示是否回顾目标(默认 true,需 G1 goal 字段)。
///
/// true(默认):警示文本引用 pinned_goal(若存在)提示「回顾目标: {goal}」,精准;
/// false:警示泛化(不引 goal 文本)。G2 不强依赖 G1(熔断不读 goal 也能工作,仅警示泛化)。
pub const STALL_BREAKER_GOAL_REMIND: bool = true;
/// G2 重复检测样本量:check_stall_breaker 扫历史 assistant tool_calls 取最近多少个签名喂判定。
///
/// 12 = 覆盖 2-3 轮多工具调用(单轮可并行多个 grep/read),既能让 is_repetitive_exploration 的
/// 「最小样本 6」+「单点 3 次」阈值有意义,又不至于扫太长历史稀释近期漂移信号。
pub const STALL_BREAKER_SAMPLE_SIZE: usize = 12;
/// L1 断路器:连续同类工具失败熔断阈值(治 kms 会话 53 轮 0 产出死循环)。
///
/// 背景:agent 无止损,某工具反复同类失败(权限拒绝/路径错误等)仍每轮重试,
/// 耗尽 max_iterations 前 0 产出。机制(非 prompt 教 AI):每轮 process_tool_calls
/// 后取末尾连续 Tool 消息,失败包按结构化信号(tool_name + 错误类别)归一为 key,
/// 入滚动窗口(CIRCUIT_BREAKER_WINDOW)统计,同一 key 在窗口内累计达此阈值 →
/// guard.reset + emit AiError + return 强制熔断,逼用户换思路或人工介入。
///
/// 阈值 3:同类失败 3 次足以判死循环(去重后仍累加,不同错误各自计数互不干扰)。
pub const CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
/// L1 断路器滑动窗口大小:仅保留最近 N 条失败记录参与计数。
///
/// 治"长任务偶发失败误熔断":全 loop 累加会让早期偶发失败与后期同类失败叠加触发。
/// 滚动窗口让计数只反映最近的失败密度——长任务中途偶发 1~2 次同类失败不触发,
/// 真正的连续死循环(N 条全同 key)才触发。N=20:对齐 max_iterations 量级,既覆盖
/// 单轮并行失败爆发,又足够长以容忍偶发抖动。
pub const CIRCUIT_BREAKER_WINDOW: usize = 20;
/// L1 断路器总开关(默认 true)。false → 跳过断路器检查,降级为纯 max_iterations
/// 旧行为(排障/对比/临时关闭用)。机制优先 prompt 说教,每改配开关 + 兜底(关降级旧行为)。
pub const CIRCUIT_BREAKER_ENABLED: bool = true;
/// L1 断路器熔断时是否发结构化求助(aichat 体验与 agent 能力系统化重构 §2.3,2026-06-21)。
///
/// true(默认):熔断 emit AiHelpRequired(结构化求助卡:reason + context + options),
/// 引导用户换思路/授权路径/人工接管(机制优先 prompt 说教,非教 AI 自己止损)。
/// false(兜底回退):熔断仍 emit AiError(旧行为,前端错误气泡),用于求助卡未就绪/
/// 排障/对比。两路保留 guard.reset + return 强制熔断语义不变,仅换前端呈现形态。
/// 配合 CIRCUIT_BREAKER_ENABLED:CIRCUIT_BREAKER_ENABLED=false 时断路器整段跳过,
/// 本开关无意义;CIRCUIT_BREAKER_ENABLED=true 时本开关决定呈现形态。
pub const CIRCUIT_BREAKER_HELP_EVENT: bool = true;
// 占位配对完整性开关(解 400 orphan)。
//
// 单一真相源:`df_ai::context_helpers::PLACEHOLDER_INTEGRITY_ENABLED`(本模块顶部已 use)。
// 删除本地副本(B 路线改进:避免与 df-ai 同名 const 双源,改一处易漏改另一处)。
//
// 根因:审批挂起占位 tool_result(内容 audit/cache.rs:pending_placeholder_for,带
// `__PENDING__:tc_id` 标记)与其 tool_call 头经 sanitize/裁剪后可能丢配对头 → orphan
// tool_result → deepseek-v4-pro 等端点 400。df-ai 的 sanitize_messages step3.5(反向
// orphan)已豁免保留占位,build_for_request 出口已用 assert_placeholder_pairing 自愈补头。
//
// 该开关控制 agentic loop 末尾(build_for_request + tool_result 压缩后)的**第二道防线**
// 出口断言:在 messages 送 stream 前再过一次 assert_placeholder_pairing(depth-defense,
// 防 build_for_request 到送 stream 之间的转换引入新 orphan)。
//
// true(默认):agentic 出口再断言一次占位配对,失败自愈补头。
// false(回退):agentic 出口不断言(仅依赖 df-ai build_for_request 内部一次自愈,旧行为)。
// 兜底:flag 关→等价改动前(仅 df-ai 内部自愈);view-only 不改持久化。
// ============================================================
// 重构第一批(2026-06-19):GeneratingGuard 抽离到 guard.rs(纯结构搬迁,行为零变更)。
// run_agentic_loop 内仍 `GeneratingGuard::new(...)`,路径从本模块改 super::guard。
// ============================================================
mod guard;
use guard::GeneratingGuard;
/// 审批超时取消(由 try_continue_agent_loop 入口调用)
mod approval_timeout;
// ============================================================
// 重构:上下文/标题/知识生命周期抽取(行为零变更)。
//
// - context_lifecycle:F-15 自动压缩逻辑,从 run_agentic_loop 内联块抽取为
// maybe_auto_compress 函数(mod.rs 调用,返回 bool 表达原 early-return)。
// - title_lifecycle / knowledge_lifecycle:标题/知识逻辑早已模块化(title.rs /
// knowledge_inject.rs),mod.rs 仅单行调用,无可抽取的内联代码,文件作文档占位。
// ============================================================
mod context_lifecycle;
use context_lifecycle::maybe_auto_compress;
#[allow(unused_imports)]
mod title_lifecycle;
#[allow(unused_imports)]
mod knowledge_lifecycle;
mod workflow_context;
// ============================================================
// L2 统一状态机(ConvState enum + 转换守卫,单一真相源)。
// 设计:generating 状态机加固 + aichat 体验与 agent 能力系统化重构。
//
// 当前架构:
// - conv_state.rs:纯逻辑 enum + 守卫 + 单测。
// - run_agentic_loop 入口桥接:迁移 ConvState(Idle→Generating / Error→Generating)。
// - guard.reset/drop 同步 ConvState→Idle(正常退出 + 异常兜底)。
// - ConvState 是生成态唯一真相源(无条件迁移 + emit)。
// - 兜底:状态机层迁移失败(非法转换)记 warn 不 panic,核心生成态复位仍经 enum 迁移收敛。
// ============================================================
pub mod conv_state;
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};
// ============================================================
// 单 Provider 流式结果 + fallback 辅助
// ============================================================
/// 单 candidate provider 一次迭代的流式调用结果。
///
/// 显式区分三态,驱动外层 `for candidate` fallback 循环:
/// - `Success`:`Complete` 或 `Partial`(MidStream 保文)。携带该 provider 路由出的
/// `resolved_model`(调用方据此更新迭代级 resolved_model,供 push/save)。
/// - `InitFailedExhausted`:流前 `InitFailed{retryable=true}` 在本 provider 上重试
/// 耗尽(或预算 30s 耗尽)。retryable=true 即「瞬态错误,换 provider 可能成功」→
/// 外层 `continue` 切下一 candidate(重建 build_provider + resolved_model 重算)。
/// 携带 `error`(最后一条诊断文本),供外层「全 candidate 耗尽」时 emit 最终 AiError。
/// - `Fatal`:流前 `InitFailed{retryable=false}`(4xx 非429/鉴权/参数错)。立即放弃
/// 整个 fallback(对齐 retry.rs Fatal 分类 + provider_pool 文档「不可重试错误
/// 立即放弃不浪费备用 provider」)。携带 `error`,stream_one_provider 内 Fatal 分支
/// emit AiError(终态不切候选,无残留气泡风险),调用方仅做 guard.reset + return。
/// 空 key(build_provider_for Err)同样归 Fatal 语义——key 缺失/钥匙串损坏非瞬态,
/// 换 provider 无意义(但实际场景:外层已对 primary 做了启动 Auth 早失败,故此分支
/// 多见于 secondary 配置不一致,保守 Fatal 收敛)。
enum StreamOutcome {
Success {
text: String,
tool_calls: std::collections::HashMap<u32, super::ToolCallDraft>,
usage: df_ai::provider::TokenUsage,
incomplete: bool,
resolved_model: String,
/// DeepSeek thinking 模式推理内容(需回传到下一轮请求)
reasoning_content: Option<String>,
},
/// error 字段供外层全 candidate 耗尽时 emit 最终 AiError(单气泡聚合)。
InitFailedExhausted { error: String },
/// error 字段供 stream_one_provider 内 Fatal 分支 emit AiError。
Fatal { error: String },
}
/// 单 provider 流式调用 + 重试。
///
/// 封装一个 candidate 的:resolve_secret → ensure_resolved_key → build_provider
/// → select_model_id(resolved_model 在该 candidate.model_configs 上重算)
/// → 流式 stream_recv + 流前 InitFailed retryable 重试循环。
///
/// 与主链共享的退避/分类:`retry::backoff_delay`(1s→2s→4s+jitter) +
/// `StreamResult::InitFailed{retryable}`(镜像 retry::is_status_retryable) +
/// 30s 总挂钟预算(`retry_deadline` 由调用方传入,本轮 fallback 各 candidate 共享一个预算)。
///
/// **permit 责任**:本函数不取并发 permit——permit 持有期间多 provider 限流不重叠,
/// 由调用方在进 candidate 前取 global/per_conv(整轮迭代共享) + 本 candidate 的
/// per-provider permit(切换 candidate 时释放旧取新)。
///
/// 返回 [`StreamOutcome`]:Fatal 含已 emit AiError;Success 含保文或正常完成;
/// InitFailedExhausted 让调用方切下一 candidate。
async fn stream_one_provider(
candidate: &AiProviderRecord,
messages: &[ChatMessage],
tool_defs: &[df_ai::provider::ToolDefinition],
app_handle: &AppHandle,
stop_flag: &std::sync::atomic::AtomicBool,
notify: &tokio::sync::Notify,
conv_id: &str,
max_retries: usize,
retry_deadline: tokio::time::Instant,
model_override: &Option<String>,
agentic_req: &TaskRequirements,
last_reasoning_content: &Option<String>,
) -> StreamOutcome {
// resolve→ensure→build 三步(复用 secret::build_provider_for,DRY)。
// key 缺失/损坏 → Err → 归 Fatal(stream_llm Fatal 也是 key 类错,语义一致)。
// 注:primary 候选的启动 Auth 早失败已在外层 run_agentic_loop 顶部处理(emit + return),
// 本函数到达时 primary 的 key 已验证过;此处 Err 多见于 secondary 配置不一致,
// 保守归 Fatal 立即放弃(不浪费预算试下一 provider,因 key 错非瞬态)。
let provider: Arc<dyn LlmProvider> = match super::secret::build_provider_for(candidate) {
Ok(p) => Arc::from(p),
Err(msg) => return StreamOutcome::Fatal { error: msg },
};
// resolved_model 在本 candidate.model_configs 上重算(provider 切换后
// 模型池不同,必须重选;否则拿主 provider 的 model_id 去打次 provider 会吃 400/404)。
// model_override 穿透:override 非空且在本 candidate 池中 → 用 override;
// 否则用 resolved_model(绝不因 override 致无模型)。
let resolved_model = resolve_model_with_override(agentic_req, candidate, model_override);
// 流前 InitFailed 重试循环。
// 重试逻辑抽至 stream_with_retry helper(扁平重构),本函数仅负责 provider 准备 + model 决议 + 委派。
stream_with_retry(
&provider, candidate, messages, tool_defs, app_handle, stop_flag, notify, conv_id,
max_retries, retry_deadline, &resolved_model, last_reasoning_content,
).await
}
// ── resolve_model_with_override: 在 candidate.model_configs 上重算 resolved_model,穿透 override ──
// override 非空且在本 candidate 池中(enabled) → 用 override;否则用 select_model_id 结果。
// 绝不因 override 致无模型(兜底用 default_model)。
fn resolve_model_with_override(
agentic_req: &TaskRequirements,
candidate: &AiProviderRecord,
model_override: &Option<String>,
) -> String {
let resolved = select_model_id(agentic_req, &candidate.model_configs)
.unwrap_or_else(|| candidate.default_model.clone());
match model_override.as_deref() {
Some(id) if !id.is_empty()
&& candidate.model_configs.iter().any(|m| m.model_id == id && m.enabled) =>
{
id.to_string()
}
_ => resolved,
}
}
// ── stream_with_retry: 流式 + 流前 InitFailed retryable 重试循环(扁平抽自原嵌套 5 层 match) ──
// 返回 StreamOutcome。内部含 retry_attempt 循环 + match StreamResult 三分支:
// Complete/Partial → Success;InitFailed Fatal → Fatal;InitFailed Retryable → 重试/耗尽。
// 嵌套 ≤ 3 层:for 内 match 内单层 if(早返回避免嵌套加深)。
async fn stream_with_retry(
provider: &Arc<dyn LlmProvider>,
candidate: &AiProviderRecord,
messages: &[ChatMessage],
tool_defs: &[df_ai::provider::ToolDefinition],
app_handle: &AppHandle,
stop_flag: &std::sync::atomic::AtomicBool,
notify: &tokio::sync::Notify,
conv_id: &str,
max_retries: usize,
retry_deadline: tokio::time::Instant,
resolved_model: &str,
last_reasoning_content: &Option<String>,
) -> StreamOutcome {
for retry_attempt in 0..=max_retries {
let retry_request = CompletionRequest {
model: resolved_model.to_string(),
messages: messages.to_vec(),
temperature: Some(0.7),
max_tokens: Some(8192),
stream: true,
tools: if tool_defs.is_empty() { None } else { Some(tool_defs.to_vec()) },
tool_choice: None,
reasoning_content: last_reasoning_content.clone(),
};
let result = stream_llm(
Arc::clone(provider), retry_request, app_handle, stop_flag, notify, conv_id,
).await;
match result {
StreamResult::Complete { text, tool_calls, usage, reasoning_content } => {
return StreamOutcome::Success {
text, tool_calls, usage, incomplete: false,
resolved_model: resolved_model.to_string(), reasoning_content,
};
}
StreamResult::Partial { text, tool_calls, usage, reasoning_content } => {
// MidStream 保文不重试。携带 incomplete=true 交调用方走保文路径。
tracing::warn!(
conv_id = %conv_id,
text_len = text.len(),
"[ai] 流中途失败(候选 {}),保文不重试(incomplete=true)",
candidate.name,
);
return StreamOutcome::Success {
text, tool_calls, usage, incomplete: true,
resolved_model: resolved_model.to_string(), reasoning_content,
};
}
StreamResult::InitFailed { retryable, error } => {
// Fatal(4xx 非429/鉴权/参数错)立即放弃整个 fallback。
if !retryable {
tracing::warn!(
conv_id = %conv_id,
provider = %candidate.name,
attempt = retry_attempt + 1,
"[ai] 候选 {} 流前失败 Fatal(4xx/鉴权),立即放弃 fallback",
candidate.name,
);
return StreamOutcome::Fatal { error };
}
// Retryable:判断是否重试预算耗尽
if let Some(exhausted_error) = check_retry_exhausted(
conv_id, candidate, retry_attempt, max_retries, retry_deadline, &error,
) {
return StreamOutcome::InitFailedExhausted { error: exhausted_error };
}
// 未耗尽:退避 + emit AiStreamRetry + sleep + continue
emit_retry_attempt(
app_handle, conv_id, retry_attempt, max_retries, retry_deadline,
).await;
}
}
}
// 仅 max_retries=0(无重试配置)且首试 InitFailed retryable=true 时到达此处:
// 0..=0 循环首趟即 is_last → 已在循环内 return InitFailedExhausted。此为防御兜底。
StreamOutcome::InitFailedExhausted {
error: "AI 调用失败:重试预算耗尽(防御兜底)".to_string(),
}
}
// ── check_retry_exhausted: 判断 InitFailed Retryable 是否已耗尽重试预算 ──
// 返回 Some(error) 表示已耗尽(调用方应返回 InitFailedExhausted);None 表示可继续重试。
fn check_retry_exhausted(
conv_id: &str,
candidate: &AiProviderRecord,
retry_attempt: usize,
max_retries: usize,
retry_deadline: tokio::time::Instant,
error: &str,
) -> Option<String> {
let is_last = retry_attempt >= max_retries;
let now = tokio::time::Instant::now();
let budget_exhausted = now >= retry_deadline;
if !is_last && !budget_exhausted {
return None;
}
// 本 candidate 重试耗尽 / 预算耗尽 → 交外层切下一 candidate。
// 不在此 emit AiError(可能切下一 candidate 成功,emit 会留残留气泡);
// 携带 error 交外层「全 candidate 耗尽」时 emit 最后一条 AiError。
let reason = if budget_exhausted { "预算耗尽" } else { "耗尽" };
tracing::warn!(
conv_id = %conv_id,
provider = %candidate.name,
attempt = retry_attempt + 1,
budget_exhausted = budget_exhausted,
"[ai] 候选 {} 流前失败重试{}({}次),切下一 provider",
candidate.name, reason, max_retries + 1,
);
Some(error.to_string())
}
// ── emit_retry_attempt: 退避日志 + emit AiStreamRetry + sleep ──
// 前置:未耗尽重试预算(check_retry_exhausted 已返 None)。本函数完成退避后返回。
async fn emit_retry_attempt(
app_handle: &AppHandle,
conv_id: &str,
retry_attempt: usize,
max_retries: usize,
retry_deadline: tokio::time::Instant,
) {
let now = tokio::time::Instant::now();
let delay = retry::backoff_delay((retry_attempt + 1) as u32)
.min(retry_deadline.saturating_duration_since(now));
let total_retry = retry_attempt + 1;
tracing::warn!(
conv_id = %conv_id,
attempt = total_retry,
max_attempts = max_retries + 1,
delay_ms = delay.as_millis() as u64,
"[ai] 流前失败(Retryable),{}ms 后重试 ({}/{})",
delay.as_millis(), total_retry, max_retries + 1,
);
let ev = AiChatEvent::AiStreamRetry {
attempt: total_retry as u32,
max_attempts: (max_retries + 1) as u32,
conversation_id: Some(conv_id.to_string()),
};
let _ = app_handle.emit("ai-chat-event", ev.clone());
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(ev);
tokio::time::sleep(delay).await;
}
/// Agentic 循环:流式接收 → 工具执行 → 结果回传 LLM → 循环
///
/// 退出条件:
/// - LLM 只返回文本(无 tool_calls)→ 正常结束
/// - 有工具需要审批 → 暂停循环(generating 保持 true),等 ai_approve 恢复
/// - 达到最大迭代次数 → 正常结束
pub(crate) async fn run_agentic_loop(
session_arc: Arc<Mutex<AiSession>>,
tools_arc: Arc<AiToolRegistry>,
db: Arc<Database>,
app_handle: AppHandle,
provider_config: AiProviderRecord,
system_prompt: String,
conv_id: String,
knowledge_config: crate::state::KnowledgeConfig,
llm_concurrency: LlmConcurrency,
max_iterations: usize,
max_retries: usize,
start_iteration: usize,
model_override: Option<String>,
) {
// generating 状态由 RAII guard 收敛复位(正常 exit 显式 reset;panic/异常 Drop 兜底)
let mut guard = GeneratingGuard::new(conv_id.clone(), app_handle.clone());
// 治本:LLM 生成间隙向前端发 AiHeartbeat,防前端 watchdog 误断流。
// 后端 emit AiHeartbeat → useAiEvents.ts resetStreamWatchdog(convId) → per-conv timer 重置。
// 解决:LLM 重试循环(~60s)期间无事件到达前端 → watchdog 45s/90s 到期误杀。
// shutdown_tx 在函数作用域结束(Drop)时通知 heartbeat 任务退出。
let (_heartbeat_tx, mut heartbeat_rx) = tokio::sync::watch::channel(());
let hb_app = app_handle.clone();
let hb_conv_id = conv_id.clone();
tokio::spawn(async move {
heartbeat_loop(&mut heartbeat_rx, &hb_app, &hb_conv_id).await;
});
// 入口桥接:loop 启动前确保 per_conv 存在(已存在则保留累积,不存在则建)。
//
// 所有调用方(IPC commands.rs 写路径 + agentic/mod.rs loop + audit.rs process_tool_calls +
// conversation.rs save + title.rs + knowledge_inject.rs)统一以 per_conv 为真相源。IPC 在 spawn
// loop 前已通过 `session.conv(active_conversation_id).*` 建立 per_conv 并写入初始状态
// (messages push user / generating=true / stop_flag=false / iteration_used=0 等),故 loop
// 入口只需确保 per_conv 存在(防御性 conv() 惰性建,正常路径下已存在)。
//
// 已存在的 per_conv(同 conv 上一轮 loop 留下,如审批暂停后续跑)**保留累积状态**:loop 上一轮
// 在 per_conv.messages 累积的 assistant tool_calls / tool_result 不会因桥接抹掉。审批拒绝/通过
// 时 IPC 直接写 per_conv.messages.replace_tool_result_content,续跑 loop 读 per_conv 拿到正确状态。
//
// guard 语义:loop 启动占用生成态,per_conv.generating=true(IPC spawn 前已置 true,此处幂等确认)。
//
// 单 loop 安全性:单 active_conversation_id 阶段无并发 loop 抢 per_conv 覆盖。多 loop 并发时,
// 每 conv 各自 per_conv 条目,互不干扰。
//
// 入口:ConvState 迁移由 guard.new 在无锁 ConvStateStore 上完成。
// 多 Provider 负载均衡池 — 选主 + fallback 候选列表。
//
// 流程:list_all → ProviderPool::select(按 模型亲和 > weight > is_default 排序)→ 有序 Vec。
// 单 provider 场景:池仅 1 enabled provider → select 返回单元素 Vec → 首位 = 唯一 provider,
// 行为同 F-01 前(零变化)。空池(0 enabled)→ fallback 入参 provider_config(保启动行为)。
//
// F-04b:select 返回的**完整有序 Vec** 即 fallback 序(主→备用)。本轮接入真实切换:
// 流式重试块外层包 `for candidate in &candidates`,主 candidate InitFailed{retryable=true}
// 耗尽 → continue 切下一 candidate(stream_one_provider 内重建 build_provider + resolved_model
// 在新 candidate.model_configs 上重算 select_model_id);Fatal(4xx 非429)立即放弃整轮。
//
// 主候选的 model_configs 用于路由(F-01),其 provider_config 用于 build_provider。
// compress_via_llm / 标题 / 后台 spawn 沿用主 candidate(非 fallback 范围,见各调用点注释)。
//
// AiProviderRepo::new 仅 clone Arc<Database>(廉价),不复用 AppState.ai_providers
// (run_agentic_loop 签名只传 Arc<Database>,改签名会牵动 3 调用点 + try_continue)。
let provider_repo = df_storage::crud::AiProviderRepo::new(&db);
// L1 补丁:provider 解析(list_all + select + resolve + ensure + build)整体包 30s timeout。
// 原实现无超时,数据库/keyring 卡死时 run_agentic_loop 入口卡住,generating 永真 + 前端看门狗
// 超时静默吞消息。超时走 AiError 分支(对齐 :412 ensure_resolved_key 失败处理),guard.reset 复位。
// 内部 list_all 失败仍容忍(空池兜底,行为不变);仅整体超时(如 DB 挂死无响应)才走 Err 分支。
let provider_resolve = tokio::time::timeout(
std::time::Duration::from_secs(30),
async {
let pool_providers: Vec<AiProviderRecord> = match provider_repo.list_all().await {
Ok(v) => v,
Err(e) => {
tracing::warn!(error = %e, "[ai] list_all providers 失败,负载均衡池退化为入参默认 provider(空池兜底)");
Vec::new()
}
};
let ranked_candidates: Vec<AiProviderRecord> = super::provider_pool::ProviderPool::select(
&pool_providers,
// specify 模式(用户指定 model):传 override 作亲和键,ProviderPool 优先选
// 「池中含该 model 的 provider」作 primary,打破下方「router 选模型需 provider_config」
// 的鸡生蛋——override 此时已知(入参 ← session.model_override),无须等 router。
// auto 模式(override=None)→ 全亲和纯权重排序,行为不变(向后兼容)。
model_override.as_deref(),
);
let (primary_provider, candidates): (AiProviderRecord, Vec<AiProviderRecord>) =
match ranked_candidates.split_first() {
Some((first, rest)) => (first.clone(), rest.to_vec()),
None => (provider_config.clone(), Vec::new()),
};
// resolve→ensure_resolved_key(空 key 早失败)→build_provider 三步统一走工厂
// 空 key 早失败(逻辑见 secret::ensure_resolved_key 单测):避免空 key 发请求吃 401,错误伪装成"API Key 无效"
//
// resolve 一次复用——原实现 build_provider_for 成功后又独立调 resolve_provider_secret
// 取 key_len(重复 keyring resolve)。现 resolve 一次:既供 key_len 诊断日志,又供 build_provider,
// 去重复 keyring resolve 调用。逻辑等价于 secret::build_provider_for(resolve→ensure→build 三步),
// 仅因 build_provider_for 隐藏 resolved key 无法复用而在此内联(未改 secret.rs 锁边界)。
let resolved_key = super::secret::resolve_provider_secret(&primary_provider);
let normalized_key = match super::secret::ensure_resolved_key(
&primary_provider.name, &resolved_key,
) {
Ok(k) => k,
Err(msg) => return Err(ProviderResolveError::EnsureKeyFailed(msg)),
};
// key_len 用归一化后长度(剥引号/trim 后),与实际发往 provider 的 key 一致
let key_len = normalized_key.len();
let provider: Box<dyn LlmProvider> = df_ai::build_provider(
&primary_provider.provider_type,
&primary_provider.base_url,
&normalized_key,
&primary_provider.default_model,
);
Ok::<_, ProviderResolveError>((primary_provider, candidates, provider, key_len))
},
).await;
let (primary_provider, candidates, provider, key_len) = match provider_resolve {
Ok(Ok((pc, cands, prov, kl))) => (pc, cands, prov, kl),
Ok(Err(ProviderResolveError::EnsureKeyFailed(msg))) => {
guard.reset().await;
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: msg.clone(),
// ensure_resolved_key 失败 = key 缺失/钥匙串损坏,归 Auth
error_type: Some(ErrorType::Auth),
conversation_id: Some(conv_id.clone()),
});
// L3 emit 双写(2026-06-22):关键 AiError publish 到事件总线,供跨模块订阅。
// EVENT_BUS_ENABLED 门控在 publish 内部(false 静默丢弃),无消费者时空转不报错。
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: msg,
error_type: None,
conversation_id: Some(conv_id.clone()),
});
return;
}
Err(_elapsed) => {
// L1 补丁:provider 解析 30s 超时(DB list_all / keyring resolve 卡死)。
// 走 AiError 分支复位 generating,对齐 ensure_resolved_key 失败处理口径。
guard.reset().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 {
error: err_msg.clone(),
error_type: Some(ErrorType::Unknown),
conversation_id: Some(conv_id.clone()),
});
// L3 emit 双写:超时 AiError publish 到事件总线(同上,门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: err_msg,
error_type: None,
conversation_id: Some(conv_id.clone()),
});
return;
}
};
// 用主候选覆盖入参 provider_config(下游 build_provider / 路由 / 日志均用此)。
// mut:F-04b 切换 candidate 后更新为实际成功所用 provider(供后续 push/save/标题 spawn)。
let mut provider_config = primary_provider;
// 诊断日志:401/错误时据此定位是 url/type/model/key 哪项问题(只记长度不记明文)
tracing::info!(
provider = %provider_config.name,
provider_type = %provider_config.provider_type,
base_url = %provider_config.base_url,
model = %provider_config.default_model,
key_len = key_len,
"[ai] 发起 LLM 请求"
);
// 主对话路由 — TaskRequirements(needs_tool_use=true)。
// 模态当前仅 Text(图像消息类型未实现,后续多模态接入时检测 Part/Image 追加 Vision)。
// select_model_id None(池空/无匹配)→ 兜底 default_model,行为与接入前一致。
//
// **子项 1+2 根因修复**:此处是 pre-loop 初始路由(算 resolved_model 兜底用),estimated_context=0
// (loop 内 messages 此处尚未构建,无法估值)+ tier=None(intent 在 739 行之后才识别)。
// 真实 estimated_context + tier 在 loop 内(line ~1395,estimated_prompt 算出后)重建 agentic_req
// shadow 此绑定,candidate chain 用 loop 内的真实估值版本(主路由路径)。
let agentic_req = TaskRequirements {
modalities: vec![Modality::Text],
needs_tool_use: true,
estimated_context: 0,
tier: None,
};
let resolved_model = select_model_id(&agentic_req, &provider_config.model_configs)
.unwrap_or_else(|| provider_config.default_model.clone());
// 用户指定模型 override 穿透(仅主对话生效,标题/扫描/灵感仍走路由)。
// 兜底原则:override 非空且在该 provider model_configs 池中 → 用 override;否则用 resolved_model。
// 绝不让 override 导致无模型(空/不在池 → 落回路由结果,行为不变)。
// mut:F-04b 切换 candidate 后由 stream_one_provider 返回的 Success.resolved_model 覆盖
// (新 candidate 模型池不同,必须重选);此后 push/save 均用迭代最新值。
let mut resolved_model = match model_override.as_deref() {
Some(id) if !id.is_empty()
&& provider_config.model_configs.iter().any(|m| m.model_id == id && m.enabled) =>
{
id.to_string()
}
_ => resolved_model,
};
// 改进2 B:意图收敛工具(LLM 可见 tool_defs 按 intent 过滤,执行路径仍走完整 registry)
//
// 机制化收敛跑题:取末条 active user 消息 → IntentRecognizer 识别 → 置信 ≥ 阈值
// 则按 intent domain 过滤工具子集(减少 LLM 在无关工具上分心/误用)。
// 三重 fallback(改进2 B §可靠):
// 1. 置信 < INTENT_CONF_THRESHOLD(0.7)→ 全量
// 2. subset 空(Chat/Unknown/Conversation)→ filter_tool_defs 内部回全量
// 3. 过滤后 < 3 条(疑似漂移/误收敛)→ 回全量
// 关键安全:filter 仅改 LLM 可见 tool_defs,**不改执行**——audit 走 tools_arc.get/execute
// 完整 registry,LLM 即使幻觉一个被滤掉的工具名,audit 也能查到/拒绝。
//
// INTENT_CONF_THRESHOLD(常量开关):阈值,低于此值不收敛(回全量)。
// 注意:conf 截断到 1.0(intent.rs),单关键词 weight=1.0 即达 conf=1.0,故阈值=1.0
// 时 conf>=1.0 仍过滤(非关闭)。真关闭收敛:置 >1.0(如 1.1)。调低 = 更激进收敛。
let user_text: String = {
let session = session_arc.lock().await;
session
.conv_read(&conv_id)
.and_then(|c| {
let msgs = c.messages.all_messages_clone();
msgs.iter()
.rev()
.find(|m| matches!(m.role, df_ai::provider::MessageRole::User))
.map(|m| m.content.clone())
})
.unwrap_or_default()
};
let (intent, conf) = IntentRecognizer::recognize(&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)
// B 路线 Phase 1 接入:PLANNING_ENABLED(false 默认关)门控 plan_hint 编排。
//
// **零行为变更(关时)**:flag 关时走 filter_tool_defs(intent 收敛扁平子集),
// 与 Phase 1 接入前完全一致——现有 intent/agentic 测试全绿无回归。
//
// **开启时**:调 filter_tool_defs_planned,它内部先 filter_tool_defs 收敛再叠加
// plan_hint 编排(并行组同批聚拢/顺序依赖源在前)。三重 fallback 与 filter_tool_defs
// 同语义(plan_hint 空/非法/registry 漂移均退 filter_tool_defs 扁平结果)。
//
// 关键安全(与 filter_tool_defs 同):本段只改 LLM 可见 tool_defs 的可见性/顺序,
// 不改执行(audit 走 tools_arc 完整 registry)。PLAN_HINT_ENABLED(plan_hint 函数开关,
// 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 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)
} else {
// 旧行为:intent 收敛扁平子集(零行为变更,flag 默认关走此路)。
filter_tool_defs(&all_defs, &intent)
};
if filtered.len() < 3 {
all_defs // 兜底:过滤<3(漂移/误收敛)回全量
} else {
filtered
}
} else {
all_defs // 低置信 fallback 全量
};
tracing::info!(
conv_id = %conv_id,
intent = intent.as_str(),
conf,
filtered = tool_defs.len(),
total,
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())
};
// token 累加器:loop 生命周期内各轮叠加,退出时传 save_conversation(累加模式落库)
let mut tokens = TokenAccumulator::default();
// 收敛标志:仅当 LLM 末轮无 tool_calls 自行 break(正常收敛)时置 true;
// 区分"正常收敛退出"与"达 MAX 被截断退出"——后者末轮 tool_calls 仍非空(tool_result 不再回传 LLM),属异常
let mut converged = false;
// L1 断路器:连续同类工具失败滚动窗口(VecDeque<key>,长度封顶 CIRCUIT_BREAKER_WINDOW)。
// key=结构化信号(tool_name + 错误类别)。每轮 process_tool_calls 后追加末尾失败 key,
// 超窗自动淘汰最旧。窗口内同 key 计数达 CIRCUIT_BREAKER_THRESHOLD → 熔断退出。
// 治"全 loop 累加不衰减":早期偶发失败不会与后期叠加误熔断。
let mut fail_window: std::collections::VecDeque<String> = std::collections::VecDeque::new();
// G2 探索熔断:重复探索计数器(loop 生命周期累计,与 fail_window 同生命周期)。
// 每轮 process_tool_calls 后 check_stall_breaker 取最近 N 个 assistant tool_calls 签名,
// 喂 is_repetitive_exploration 判重复,重复 stall_count+=1,非重复重置 0。
// 达 STALL_BREAKER_THRESHOLD → 警示/熔断(治 R4 游荡死循环 + token 失控)。
// 注:2026-08-01 从「末尾 Tool 结果是否空(is_empty_tool_result)」改为「调用签名重复」,
// 治旧范式误熔断正常排除式搜索(实证会话 ac448296)。
let mut stall_count: u32 = 0;
let mut stall_warned: bool = false;
// DeepSeek thinking 模式推理内容跨轮透传
let mut last_reasoning_content: Option<String> = None;
// G1 目标钉扎:入口把 PerConvState.pinned_goals 拼进 system_prompt 尾部(一次拼好整个 loop 复用)。
//
// 治 R1(目标消息被压缩出局)/R5(prompt 说教无锚点):system_prompt 是 loop 不变量 + build_for_request
// 从不裁剪它,故目标天然免疫压缩/裁剪/sanitize。本块是治 R1 的结构性根因(目标进 prompt 字符串非
// messages 流,无 insert_at(0) 的连续 System 1214/首位锚点稀释/小预算被裁三重风险)。
//
// 单次 lock 读 pinned_goals clone(复用 stop_flag 取用模式,同一 lock 块);非空 → 逐条截断到
// GOAL_MAX_CHARS,按 GOAL_INJECT_BANNER 拼 banner+编号列表。GOAL_PIN_ENABLED=false → 整块跳过,
// pinned_goals 永远空 Vec(单点回退等价改动前)。拼接在 sys_tokens 估算前。
//
// 注:仅 run_agentic_loop 入口注入。手动压缩/标题/提炼等路径不注入目标。
//
// 目标提取方式(2026-07-03 改):从 LLM 本轮工具调用推理目标描述(工具名+路径),
// 不再从用户消息规则提取,也不依赖 LLM 输出结构化标记。
// 推理结果存入 pinned_goals: Vec<GoalEntry>,含 text+status 状态追踪。
// 新目标加入时自动标记之前的 active 为 completed。
let mut system_prompt = system_prompt;
if GOAL_PIN_ENABLED {
let goals: Vec<super::GoalEntry> = {
let session = session_arc.lock().await;
session
.conv_read(&conv_id)
.map(|c| c.pinned_goals.clone())
.unwrap_or_default()
};
// 只把 active 目标注入 system_prompt(completed 不干扰 LLM 注意力)
let active_goals: Vec<&super::GoalEntry> = goals.iter().filter(|g| matches!(g.status, super::GoalStatus::Active)).collect();
if !active_goals.is_empty() {
let goal_lines: Vec<String> = active_goals.iter().enumerate().map(|(i, g)| {
let trimmed = g.text.trim();
let truncated: String = trimmed.chars().take(GOAL_MAX_CHARS).collect();
let truncated = if truncated.chars().count() >= GOAL_MAX_CHARS {
format!("{}…", truncated)
} else {
truncated
};
format!("{}. {}", i + 1, truncated)
}).collect();
let goals_text = goal_lines.join("\n");
system_prompt = if GOAL_INJECT_BANNER {
format!(
"{}\n\n## 当前目标(全程锚定,所有动作须服务于它们)\n{}",
system_prompt, goals_text
)
} else {
format!("{}\n\n{}", system_prompt, goals_text)
};
tracing::info!(
conv_id = %conv_id,
count = goals.len(),
first_goal = %goals.first().map(|g| &g.text[..std::cmp::min(120, g.text.len())]).unwrap_or(""),
"[ai] G1 目标钉扎:已把 {} 个 pinned_goals 拼进 system_prompt",
goals.len()
);
}
}
// T5: WorkingContext L1 注入(L1a 常驻 + L1b 条件注入)
// 读取工作上下文,构建 L1 注入文本并拼入 system_prompt
{
let session = session_arc.lock().await;
if let Some(conv) = session.conv_read(&conv_id) {
let injection = conv.working_context.build_injection(3); // dirty_window=3
if !injection.is_empty() {
system_prompt = format!("{}\n\n## 工作上下文\n{}", system_prompt, injection);
}
}
}
// EnvSnapshot 环境感知:入口把当前平台的真实环境(OS/shell/工具版本)拼进 system_prompt 尾部。
//
// 治「LLM 跨平台命令幻觉」:LLM 训练数据 Unix 多,易生成 macOS/Linux 语法命令(PowerShell 5
// 不支持 `&&`、Windows 路径分隔符 `\`、GBK 终端中文乱码),把真实环境塞 prompt 即可锚定
// 输出平台一致性。detect() 是 OnceLock 全局缓存(启动时探一次,后续零开销),与 G1 一样
// 是 loop 不变量(整个会话不重探),与目标钉扎拼接次序无强约束(放其后,语义自然)。
let env_prompt = df_execute::EnvSnapshot::detect().await.to_prompt();
let behavior_prompt = concat!(
"\n## AI 定位\n",
"你是 DevFlow 的 AI 助手,拥有完整的工具链。用户只负责提需求和审批,",
"所有执行由你完成——读写文件、运行命令、创建项目、搜索代码等都是你直接调用工具完成的。",
"**绝不输出请在终端执行以下命令这类指令——你自己用 run_command 工具执行即可。**\n",
"\n## 行为准则\n",
"- 所有操作都通过工具完成,用户不参与执行\n",
"- 优先使用开发工具 IPC,非必要不写独立脚本\n",
"- 脚本需要审批通过才执行,会拖慢工作流\n",
"- 已有 40+ 工具覆盖绝大多数场景,先查工具列表再决定\n",
"- 如果现有工具无法完成任务,告知用户缺少什么能力,建议向 DevFlow 反馈以开发新工具\n",
"- **直接回答用户问题,不要以评论开头**——严禁用好问题等无信息量的开场白,直接输出答案或执行操作\n",
"- **及时收敛**:工具执行完目标达成后立即输出结果,不要继续调工具做无关操作",
);
system_prompt = format!("{}\n\n{}\n\n{}", system_prompt, env_prompt, behavior_prompt);
// T4: 工作流 DAG 注入 — 当会话关联工作流时,将活跃路径注入 system prompt
{
let session = session_arc.lock().await;
if let Some(conv) = session.conv_read(&conv_id) {
if let Some(ref dag_summary) = conv.workflow_dag_summary {
system_prompt = format!("{}[工作流]\n{}\n", system_prompt, dag_summary);
tracing::debug!(
conv_id = %conv_id,
workflow_id = ?conv.workflow_id,
"[ai] T4: 已注入工作流 DAG 上下文"
);
}
}
}
// ── 多 Agent 并行执行:Coordinator 分解(plan_execution_enabled 时) ──
// 对话透明化 L1:拍快照供 AiCompleted 事件携带(coordinator 路径出口也用)
// 只取 text(前端不需要状态信息)
let pinned_goals_snapshot: Vec<super::GoalEntry> = {
let session = session_arc.lock().await;
session
.conv_read(&conv_id)
.map(|c| c.pinned_goals.clone())
.unwrap_or_default()
};
// ── Plan-driven Phase 1:LLM 规划端(AICHAT_PLAN_ENABLED 门控) ──
//
// 对齐 docs/02-架构设计/专项设计/aichat-plan-driven-设计-2026-08-01.md Phase 1:
// 开启时由 LLM 生成 Plan JSON(替代 coordinator.decompose 关键词匹配),成功后 emit
// AiPlanCreated 事件(步骤 + 状态 pending)供前端展示(前端卡片 Phase 3 后续,本步只 emit)。
//
// **职责边界(对齐设计文档)**:
// - Phase 1 只治「Plan 从哪来」(LLM 出 Plan)+ 事件可见性,**不改执行逻辑**;
// - 执行调度(Plan→to_layers→并行/串行 dispatch)是 Phase 2 范围,留 plan_executor 接通后做;
// - 故本块 emit 后**继续走 ReAct 主链**(不 early return),Plan 仅作规划可见性,执行仍单链 ReAct。
//
// **开关默认关 + 兜底回退**(memory ai-improvement-principles):
// - AICHAT_PLAN_ENABLED=false(默认)→ 整块跳过,零行为变更(现有 ReAct 不受影响);
// - LLM 调用失败/JSON 解析失败/validate 失败 → decompose_with_llm 返 None → 整块跳过,
// 继续走 ReAct(不阻断,三重兜底);
// - 与 plan_executor::PLAN_EXECUTION_ENABLED 正交:执行开关独立治 dispatch,本块不碰。
//
// LLM 调用复用主对话 provider + resolved_model(已解析,无需二次 build);available_tools
// 喂 LLM 限定 tool_hint 取值域防幻觉(取 tool_defs 名清单,即 LLM 可见工具子集)。
if df_ai::coordinator::aichat_plan_enabled() && conf >= INTENT_CONF_THRESHOLD {
let coord = Coordinator::new(PersonaRegistry::new());
// 可用工具名清单(供 LLM 填 tool_hint,防幻觉不存在的工具名)
let available_tools: Vec<String> = tool_defs
.iter()
.map(|d| d.function.name.clone())
.collect();
// LLM 调用包 30s 超时(对齐 compress_via_llm 60s / title 20s 量级):规划属独立 LLM
// 调用,超时返 None 回退 ReAct,不拖垮 agentic loop。
let plan_result = tokio::time::timeout(
std::time::Duration::from_secs(30),
coord.decompose_with_llm(
&*provider,
&resolved_model,
intent.as_str(),
&user_text,
&available_tools,
),
)
.await;
match plan_result {
Err(_) => {
tracing::warn!(
conv_id = %conv_id,
"[PLAN-LLM] LLM 规划调用超时(30s),回退纯 ReAct"
);
}
Ok(Some(decompose_result)) => {
// emit AiPlanCreated(步骤 + 状态 pending)供前端展示(Phase 3 前端卡片后续)
let layers_payload = build_plan_layers_payload(&decompose_result.plan, &coord);
let plan_id = format!("plan-{}", ulid_like_id());
tracing::info!(
conv_id = %conv_id,
plan_id = %plan_id,
layer_count = layers_payload.len(),
task_count = decompose_result.subtasks.len(),
"[PLAN-LLM] Plan 创建,emit AiPlanCreated"
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiPlanCreated {
plan_id: plan_id.clone(),
layers: layers_payload,
conversation_id: Some(conv_id.clone()),
});
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(
AiChatEvent::AiPlanCreated {
plan_id,
// layers 已 move,复用 decompose_result 重建等价载荷(reborrow 避免 move)
layers: build_plan_layers_payload(&decompose_result.plan, &coord),
conversation_id: Some(conv_id.clone()),
}
);
// Phase 1:emit 后继续走 ReAct 主链(执行调度是 Phase 2)。
// Plan 仅作规划可见性 + 前端展示;执行仍单链 ReAct,不改 dispatch 逻辑。
}
Ok(None) => {
// 三重兜底已触发(LLM 错/解析错/validate 错),decompose_with_llm 内已记 warn。
// 静默继续 ReAct(不阻断)。
tracing::info!(
conv_id = %conv_id,
"[PLAN-LLM] LLM 规划未产出有效 Plan,继续走 ReAct"
);
}
}
}
if df_ai::plan_executor::plan_execution_enabled() && conf >= INTENT_CONF_THRESHOLD {
let coord = Coordinator::new(PersonaRegistry::new());
let decompose_result = coord.decompose(intent.as_str(), &user_text);
tracing::info!(
conv_id = %conv_id,
intent = intent.as_str(),
subtask_count = decompose_result.subtasks.len(),
has_plan = !decompose_result.plan.is_empty(),
"[COORDINATOR] 意图分解完成"
);
if !decompose_result.plan.is_empty() {
// 串行执行子任务(Phase 1 简单路径,后续可升级并行 dispatch_with_budget)
let results = coord.dispatch(&decompose_result.plan, |subtask, persona| async move {
ExecutionResult {
subtask_id: subtask.id.clone(),
persona_id: persona.id.clone(),
output: format!(
"### 子任务: {}{}\n\n意图: {}",
subtask.intent, persona.name, subtask.intent
),
success: true,
}
}).await;
// 合并子任务结果
let merge_result = coord.merge(&results);
// 将合并产出推回主对话(单条可展开 assistant 消息)
// 消息级 token:Coordinator 路径无本轮 LLM(子任务已各自计入其会话/或未计入),
// 此处为合成汇总非直接 LLM 输出,与下方 emit coord_usage(空 tokens 快照)对齐 → 0。
{
let mut session = session_arc.lock().await;
let mut merge_msg = ChatMessage::assistant(&format!(
"## 多 Agent 执行完成\n\n共执行 {} 个子任务,成功 {} 个。\n\n{}",
results.len(),
results.iter().filter(|r| r.success).count(),
merge_result.merged_output
));
merge_msg.prompt_tokens = Some(0);
merge_msg.completion_tokens = Some(0);
merge_msg.prompt_cache_hit_tokens = Some(0);
merge_msg.prompt_cache_miss_tokens = Some(0);
merge_msg.reasoning_tokens = Some(0);
session.conv(&conv_id).messages.push(merge_msg);
if !merge_result.conflicts.is_empty() {
let mut conflict_msg = ChatMessage::assistant(&format!(
"### 冲突检测\n\n检测到 {} 个文件冲突:\n{}",
merge_result.conflicts.len(),
merge_result.conflicts.iter()
.map(|c| format!("- {}: {}", c.file, c.description))
.collect::<Vec<_>>()
.join("\n")
));
conflict_msg.prompt_tokens = Some(0);
conflict_msg.completion_tokens = Some(0);
conflict_msg.prompt_cache_hit_tokens = Some(0);
conflict_msg.prompt_cache_miss_tokens = Some(0);
conflict_msg.reasoning_tokens = Some(0);
session.conv(&conv_id).messages.push(conflict_msg);
}
}
// 落库 + generating 复位 + emit AiCompleted(统一走 finish_round_exit,行为零变更)
// Coordinator 路径:save(None,None) + 不 spawn_title + emit(None,None,publish=true)
// emit_usage=tokens 快照(本路径无 round, tokens 为空, total=0 对齐原 tokens.total())
let coord_usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
finish_round_exit(
&session_arc, &db, &conv_id,
None, None,
false,
&provider_config, &llm_concurrency,
&mut guard,
&coord_usage,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
).await;
return;
}
}
// system_prompt 是 run_agentic_loop 的不变参数(整个 loop 期间文本不变),
// 其 token 估算在 loop 外算一次缓存复用,避免每轮/每次重试重复 estimate_text(低收益优化,行为不变)。
let sys_tokens = TokenEstimator::default().estimate_text(&system_prompt);
// 并发限流策略(用户决策「不设并发会话上限」):
// - global 不再由 loop 入口持整 loop(原会话数上限 N=3 已废),多对话 loop 并发不限。
// - per_conv 由 loop 入口(_conv_per_conv_permit)整 loop 持有(含工具执行/审批等待/重试),
// permits=2 防单对话内并发 LLM 调用失控。
// - permit 绑 guard(函数返回)Drop 自动释放——各 return 点退出即释放槽位。
// retry 同 loop 内,持 per_conv 合理。
let _conv_per_conv_permit = llm_concurrency.acquire_per_conv(&conv_id).await;
// 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) {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
// 入口 stop:本轮可能尚未 stream(首轮即停),不记 model——避免把未实际生成的 model 写入 models 数组
// 统一走 finish_round_exit:save(Some usage, None model) + spawn_title + emit(None,None,publish=true)
finish_round_exit(
&session_arc, &db, &conv_id,
Some(&usage), None,
true,
&provider_config, &llm_concurrency,
&mut guard,
&usage,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
).await;
return;
}
// 改进5: 主题切换系统标记(保守,双高置信才标,软提示非强制)。
//
// push 时若末两条 user 消息的 topic 都非 None 且不同(Intent 高置信推断的双 topic),
// ContextManager 已置位 pending_topic_marker("old|new")。loop 顶部读取并消费
// (take 一次性清空,防重复 insert),insert 一条 system 软提示告知 LLM 用户已切换话题。
//
// 保守设计:
// - 双高置信:两条 topic 都非 None(都达 0.7 阈值)才标,任一 None(低置信未标)不标。
// - 软提示:仅 insert 一条 system 消息,不强制 LLM 行为(LLM 仍可按自己理解响应)。
// - TOPIC_MARKER_ENABLED(常量开关)false → 跳过(排障/对比用)。
// topic 不参与裁剪/压缩(只供检测),insert_at(0) 同压缩摘要定位(首位 system)。
if TOPIC_MARKER_ENABLED {
let topic_marker_raw: Option<String> = {
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
tracing::warn!(
stale_conv = %conv_id,
"[ai] conv 已删除,loop 退出(主题标记段入口)"
);
return;
}
let conv = session.conv(&conv_id);
// G4 目标感知降级:总是 take_topic_marker(防 marker 累积),但若 pinned_goals 非空
// 且 TOPIC_MARKER_GOAL_AWARE → 丢弃 take 结果(返 None 跳过 insert)。
// 一次 lock 同读 pinned_goals(避免额外加锁)。take 后丢弃不影响下一轮(marker 每 push
// user 重检测生成,丢弃一次不残留)。GOAL_AWARE=false → 原样返回 marker(退旧行为)。
let goal_active = TOPIC_MARKER_GOAL_AWARE && !conv.pinned_goals.is_empty();
let marker = conv.messages.take_topic_marker();
if goal_active && marker.is_some() {
tracing::info!(
conv_id = %conv_id,
iteration,
"[ai] G4 目标钉扎已启用,跳过话题标记 insert 避免双锚点稀释"
);
None
} else {
marker
}
};
if let Some((old_topic, new_topic)) = topic_marker_raw.and_then(|s| {
// 解析 "old|new" 格式;splitn 防 topic 名内含 '|' 误切(仅切首 '|' 一次)。
let mut parts = s.splitn(2, '|');
let old = parts.next()?.to_string();
let new = parts.next()?.to_string();
Some((old, new))
}) {
let marker_text = format!(
"── 用户已切换话题(从「{}」到「{}」),请以新话题为准 ──",
old_topic, new_topic
);
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(&conv_id) {
let conv = session.conv(&conv_id);
conv.messages.insert_at(0, ChatMessage::system(&marker_text));
tracing::info!(
conv_id = %conv_id,
iteration,
old_topic = %old_topic,
new_topic = %new_topic,
"[ai] 主题切换标记已 insert(软提示,保守双高置信)"
);
}
}
}
// 旧 loop 污染防护——每轮开始校验对话一致性。
// 用户新建/切换对话后 active_conversation_id 变更,本 loop(conv_id 快照)成陈旧,
// 继续跑会往新对话 push 消息/pending 造成污染。检测到即退出(guard Drop 复位 generating)。
//
// 退出判据:!per_conv.contains_key(conv_id)(conv 存在性)。旧 loop 跑自己的 conv
// 不污染他人,active_conversation_id 切换不应让旧 loop 退出;仅当 conv 被删(clear/delete)
// 才退出。单 loop 阶段 conv 不会被删,此校验主要保留语义对齐(真并发场景生效)。
// conv_id 来源:run_agentic_loop 入参(与 guard/stop_flag 取用同源)。
{
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
tracing::warn!(
stale_conv = %conv_id,
"[ai] conv 已删除,旧 loop 退出"
);
return;
}
// 累计 iteration 计数(「下一轮起算值」= 当前轮+1)。
// 审批等待/达 max 暂停退出时,本字段即「当前轮+1」;审批续跑 ai_approve 读此值作
// start_iteration 透传,实现 iteration 累计不重置(防多次审批反复跑满 max 致 token 失控)。
// per_conv.iteration_used 唯一真相源。
let conv = session.conv(&conv_id);
conv.iteration_used = iteration + 1;
}
// 新一轮通知前端(第二轮起),前端需新建 assistant 消息
if iteration > 0 {
let round_n = (iteration + 1) as u32;
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiAgentRound {
round: round_n,
conversation_id: Some(conv_id.clone()),
});
// L3 emit 双写:AiAgentRound publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiAgentRound {
round: round_n,
conversation_id: Some(conv_id.clone()),
});
}
// 自动压缩(智能裁剪)——已抽取至 context_lifecycle::maybe_auto_compress。
//
// 抽取自本函数原内联块(行为零变更)。返回 true = conv 已删除,loop 应立即 return
// (对齐原内联块入口的 early-return);返回 false = 正常结束,继续后续流程。
// 实现细节(触发条件 / 延迟 mutate 口径 / 关键词兜底)见
// context_lifecycle.rs 顶部文档与函数体内注释(原样保留)。
if maybe_auto_compress(
&session_arc,
&conv_id,
&app_handle,
&provider,
&provider_config,
&llm_concurrency,
iteration,
sys_tokens,
).await {
return;
}
// 构建请求消息(超预算时自动裁剪旧消息,保护工具调用三元组 + 最近 6 条)
// sys_tokens 已在 loop 外缓存;本块构建的 messages 在本轮重试循环中复用
// (本轮 stream_llm 不持 session_arc、不改 messages,重试无 push 发生,重建等价于复用)。
// build_for_request 读 per_conv.messages。
// conv_id 来源:run_agentic_loop 入参。
let messages = {
let session = session_arc.lock().await;
let conv = match session.conv_read(&conv_id) {
Some(c) => c,
None => {
tracing::warn!(stale_conv = %conv_id, "[ai] conv 已删除,loop 退出(build_for_request 入口)");
return;
}
};
let (mut history_msgs, _trimmed) = conv.messages.build_for_request(sys_tokens);
// T2-修复(方案A):发 LLM 前展开 namespace 引用为真实内容(view-only,仅本轮请求视图)。
// push 时(audit/mod.rs)read_file/list_directory/grep/diff_files 等大结果被
// namespace_store 引用化进 messages;持久化(conversation.rs)已展开写 DB,但发 LLM 的
// 实时请求此前未展开 → LLM 每轮只看到 "namespace://..." 占位符,AI 实际拿不到工具结果。
// 此处展开,namespace 退化为内存/持久化层优化,LLM 永远看真实内容。
// read_only None(条目被 LRU 淘汰 / 重启后 namespace 清空 / 跨会话残留引用)→
// 替换为 EVICTED_PLACEHOLDER 提示文案(非保留无意义 namespace:// URI)+ warn 日志。
for m in &mut history_msgs {
if df_ai::namespace_store::is_namespace_ref(&m.content) {
let path = m.content.clone();
if let Some(original) = session.namespace_store.read_only(&path) {
m.content = original.to_string();
} else {
tracing::warn!(
conv_id = %conv_id,
path = %path,
"[namespace] 引用已淘汰,替换为提示文案"
);
m.content = df_ai::namespace_store::EVICTED_PLACEHOLDER.to_string();
}
}
}
let mut msgs = vec![ChatMessage::system(&system_prompt)];
msgs.extend(history_msgs);
msgs
};
// 改进4: tool_result view-only 摘要压缩(build_for_request 后,送 stream 前)。
//
// 遍历 messages(history clone,已含 system prompt + sanitize 后历史)中的 role=Tool 消息,
// 超阈值(content >2KB 或占历史 token >40%)的 content 应用 extract_key_info 替换:
// 保留错误行(error/panic/失败/.rs:N)+ 首/尾各 5 行,中间省略。
//
// **view-only**:messages 是 build_for_request 返回的 clone,改它只影响本轮 LLM 请求视图,
// 不改 ContextManager 持久化(对齐 sanitize_messages:DB 原始 tool_result 完整保留)。
// 故即使摘要有误/过度压缩,下次 build_for_request 仍从 DB 全量重建,可自愈。
//
// TOOL_RESULT_COMPRESS_ENABLED=false(常量开关)→ 跳过(原样送 LLM,排障/对比用)。
let messages: Vec<ChatMessage> = if TOOL_RESULT_COMPRESS_ENABLED {
let history_tokens_snapshot: u32 = {
let session = session_arc.lock().await;
session
.conv_read(&conv_id)
.map(|c| c.messages.history_tokens())
.unwrap_or(0)
};
summarize_tool_results(messages, history_tokens_snapshot, &conv_id, iteration)
} else {
messages
};
// 占位配对完整性(第二道防线,depth-defense):build_for_request 已在 df-ai 内部
// 自愈一次,此处 tool_result 压缩/系统提示插入后再断言一次,防转换引入新 orphan。
// view-only:messages 是 clone,assert_placeholder_pairing 仅改本 Vec,不改 ContextManager 持久化。
let messages = ContextManager::assert_placeholder_pairing(messages, PLACEHOLDER_INTEGRITY_ENABLED);
// 预估输入 token(兜底:部分 provider 如 GLM 流式 usage 不报 prompt_tokens,后段用它补)
// 注:stream_one_provider 内每次重试重建 request(因 provider.stream 消费 body),
// 此处不再预构建 request(旧 request 变量已废弃),仅保留 messages 供 estimated_prompt。
let estimated_prompt: u32 = {
let est = TokenEstimator::default();
messages.iter().map(|m| est.estimate_message(m)).sum()
};
// 子项 1+2 根因修复:loop 内重建 agentic_req,真实 estimated_context + 意图 tier。
//
// **子项 1(estimated_context 死代码)**:原 pre-loop agentic_req 传 0,
// 上下文窗口过滤维度(context_window >= estimated_context)恒过,失效。此处用本轮
// estimated_prompt(system+history 全量 token 估值)作 estimated_context → 窗口过滤生效
// (小窗口模型如 8K 被大上下文任务正确滤掉,不再误选)。
//
// **子项 2(weight tier tiebreak)**:tier 接 intent→ModelTier(Code/Debug→Heavy,
// Chat→Fast 等),同 weight 候选间按 tier tiebreak(满足档位下限的候选胜,见 router.rs)。
// intent 在 loop 外(786 行)已识别,loop 内每轮复用同一 intent(用户末条 active 消息
// 在单轮 LLM 调用内不变;多轮对话 intent 演化由用户后续消息触发,下轮 recognize 更新)。
//
// **变量 shadow**:此绑定覆盖 pre-loop 的 agentic_req(line ~744 算初始 resolved_model
// 用过),candidate chain(下方 stream_one_provider)取此 loop 内版本。Rust shadow 安全:
// pre-loop 版本在 line 744 用完即弃,resolved_model 已落到 mut 变量。
let agentic_req = TaskRequirements {
modalities: vec![Modality::Text],
needs_tool_use: true,
estimated_context: estimated_prompt as usize,
tier: suggested_model_tier(&intent),
};
// LLM 并发限流:
// per_conv 由 loop 入口(_conv_per_conv_permit)整 loop 持有(含工具执行/审批等待/重试),
// 防单对话内并发 LLM 调用失控(单对话内 permits=2,非会话数限制)。
// global 不再由 loop 入口持有,回归原义「LLM 调用并发限流」:
// 由各单次 LLM 调用点(stream_llm 重试循环内/标题/压缩/提炼/项目分析)各自 acquire/drop 防 429。
//
// 流前失败(Init Err)重试,流中途失败(MidStream Partial)不重试保文。重试退避复用
// retry::backoff_delay(1s→2s→4s±20% jitter) + retry::is_status_retryable Fatal 分类
// (stream_recv classify_status_or_class 镜像, 4xx 非429 立即放弃) + 30s 总挂钟预算。
// 重试期间持有 per_conv permit 不释放(防新请求挤占)。
//
// per-provider permit 仍在 candidate 循环内取(切换 candidate 时释放旧取新,避免占用未用 provider 的槽);
// per_conv 由 loop 入口持有。
// 重试总预算(挂钟,含 sleep + 各次请求耗时),对齐 retry::MAX_TOTAL_BUDGET 30s。
// 本轮各 candidate 共享一个 30s 预算(切换 provider 不重置预算,
// 防多 provider 串行重试累加超过单轮总预算)。超预算直接放弃重试交最终错误/保文路径。
let retry_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
// 外层 for candidate 包裹 stream_one_provider。
// 顺序:[primary, ...candidates](空池兜底时仅 primary)。
// - Success → 用结果覆盖迭代级 resolved_model/provider_config(供 push/save/标题);
// 成功即 break。
// - InitFailedExhausted(retryable 耗尽)→ continue 切下一 candidate。
// - Fatal(4xx 非429/鉴权)→ stream_llm 已 emit AiError,guard.reset + return
// 放弃整轮(对齐 retry.rs Fatal + provider_pool 文档)。
// 全 candidate 耗尽 → 沿用单 provider 失败语义(guard.reset + return,AiError 已在
// 各 candidate 的 stream_llm 内 emit 最后一条)。
// 候选链(拥有 Vec):[primary, ...candidates]。预先 clone primary 入链,
// 避免借用 provider_config(切换成功后需写回 provider_config = candidate.clone())。
// candidates 空(单 provider)→ candidate_chain 仅 primary,循环跑一次,耗尽即 return。
let mut candidate_chain: Vec<AiProviderRecord> = Vec::with_capacity(1 + candidates.len());
candidate_chain.push(provider_config.clone());
candidate_chain.extend(candidates.iter().cloned());
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;
// 追踪最后一个 Exhausted candidate 的诊断文本,供「全 candidate 耗尽」时 emit 最终 AiError。
// Fatal 分支即时 emit(终态不切候选),Exhausted 分支仅记录 error 不 emit(可能切下一 candidate 成功,
// emit 会留残留气泡)。
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;
let stream_outcome = stream_one_provider(
candidate,
&messages,
&tool_defs,
&app_handle,
&stop_flag,
&notify,
&conv_id,
max_retries,
retry_deadline,
&model_override,
&agentic_req,
&last_reasoning_content,
).await;
match stream_outcome {
StreamOutcome::Success { text, tool_calls, usage, incomplete, resolved_model: m, reasoning_content: rc } => {
// 成功:更新迭代级 resolved_model + provider_config(供后续 push/save/标题)。
// mut 解构(已在顶部声明 mut):本轮后续 save/push 用「实际成功所用 provider」
// 而非主 candidate。compress 已在本轮 stream 之前用过 primary,不受影响。
resolved_model = m;
provider_config = candidate.clone();
last_reasoning_content = rc;
outcome = Some((text, tool_calls, usage, incomplete, last_reasoning_content.clone()));
break 'candidate;
}
StreamOutcome::InitFailedExhausted { error } => {
// 本 candidate 重试耗尽(retryable)。切下一 candidate 继续尝试。
// candidates 空(单 provider)时此即「耗尽 return」语义——
// 跳出循环后 outcome 仍 None,落入下方「全耗尽 return」兜底。
// 不在此 emit AiError(可能切下一 candidate 成功留残留气泡),
// 仅记录 error 供全耗尽兜底 emit 最后一条。
tracing::warn!(
conv_id = %conv_id,
provider = %candidate.name,
"[ai] 候选 {} 流前失败重试耗尽,切换下一 provider",
candidate.name,
);
last_exhausted_error = Some(error);
continue 'candidate;
}
StreamOutcome::Fatal { error } => {
// Fatal(4xx 非429/鉴权/参数错):立即放弃整轮 fallback。
// 统一 emit 最终错误气泡(单气泡聚合)。
guard.reset().await;
emit_fatal_error(&app_handle, &conv_id, &error).await;
return;
}
}
}
match outcome {
Some(r) => r,
None => {
// 全 candidate 耗尽(含单 provider 场景):沿用原单 provider 失败语义。
tracing::warn!(
conv_id = %conv_id,
candidates_tried = candidate_chain.len(),
"[ai] 全 provider 候选流前失败重试耗尽,放弃本轮",
);
guard.reset().await;
let err_msg = last_exhausted_error
.unwrap_or_else(|| "AI 调用失败:所有候选 provider 重试耗尽".to_string());
emit_fatal_error(&app_handle, &conv_id, &err_msg).await;
// 全 provider 失败也需落库 user 消息(已 push 内存),
// 否则切换/重载从 DB 恢复时末条 user 丢失(实测压缩触发失败场景:DB 末条 tool 无后续 user)。
save_conversation(&session_arc, &db, &conv_id, None, Some(&resolved_model), true).await;
return;
}
}
};
// CR-30-2 / UX-2025-04 / 决策 a1: MidStream 保文路径——partial_text 已接收,
// 入库为正常 assistant 消息 + emit AiCompleted(incomplete=true) + 追加系统提示消息。
// 不走 AiError(非异常中断,已有可用文本),不重试(决策 a1)。
// 并发时用户停优先:网络断同时用户点停 → stop_flag true 时不走保文路径,
// 落入下方 ~1508 stop_flag 检查走停止路径(用户意图优先)。
// partial 文本仍由下方 push_assistant_message(~1485)保文不丢。
if incomplete && !stop_flag.load(Ordering::SeqCst) {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: if round_usage.prompt_tokens == 0 { estimated_prompt } else { round_usage.prompt_tokens },
completion_tokens: round_usage.completion_tokens,
total_tokens: if round_usage.prompt_tokens == 0 { estimated_prompt + round_usage.completion_tokens } else { round_usage.total_tokens },
// 分项 token(2026-08-02):cache/reasoning 透传自 round_usage,落库 + 累加器都需
prompt_cache_hit_tokens: round_usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: round_usage.prompt_cache_miss_tokens,
reasoning_tokens: round_usage.reasoning_tokens,
};
// 累加本轮全量 usage(含 cache/reasoning 分项)到 tokens 累加器
tokens.add_usage(&usage);
// 追加 partial assistant 消息(若无 tool_calls 且有文本)
// 退出校验改 conv 存在性 + push 改 per_conv.messages。
// conv_id 来源:run_agentic_loop 入参。
{
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
tracing::warn!(
stale_conv = %conv_id,
"[ai] MidStream 保文后 conv 已删除,丢弃本轮 push"
);
return;
}
if !full_text.is_empty() {
let conv = session.conv(&conv_id);
let mut msg = ChatMessage::assistant(&full_text);
msg.model = Some(resolved_model.clone());
// MidStream 保文也回填 reasoning_content
msg.reasoning_content = round_reasoning_content.clone();
// 消息级 token(对齐 push_assistant_message 双轨持久化):本轮 partial usage
// (prompt=round 或 estimated 兜底,completion=round)。系统提示消息无 token,不设。
// 分项 token(2026-08-02):cache/reasoning 透传自 round_usage。
msg.prompt_tokens = Some(usage.prompt_tokens);
msg.completion_tokens = Some(usage.completion_tokens);
msg.prompt_cache_hit_tokens = Some(usage.prompt_cache_hit_tokens);
msg.prompt_cache_miss_tokens = Some(usage.prompt_cache_miss_tokens);
msg.reasoning_tokens = Some(usage.reasoning_tokens);
conv.messages.push(msg);
// 追加系统提示消息:响应因网络中断不完整(对齐决策 a1 系统提示机制)
let mut notice = ChatMessage::system("⚠ 响应因网络中断不完整,以上为已接收的部分内容。可重新发送以获取完整回复。");
notice.model = Some(resolved_model.clone());
conv.messages.push(notice);
}
}
// 统一走 finish_round_exit 收尾(save + spawn_title + reset + emit)。
// 注意:partial 文本+系统提示已先 push(上方 block),此 save 落库含本轮 partial,幂等覆盖。
// emit_usage 用 tokens 快照(tokens.add 已累加本轮):total_tokens/prompt/completion 对齐原 emit 三元组。
// MidStream 分叉:emit_incomplete=Some(true)(前端标不完整),publish_incomplete=None(总线消费方),
// do_publish=true(publish 走总线)。spawn_title=true(后台标题,失败 extract 兜底)。
let emit_usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: usage.total_tokens,
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
finish_round_exit(
&session_arc, &db, &conv_id,
Some(&usage), Some(&resolved_model),
true,
&provider_config, &llm_concurrency,
&mut guard,
&emit_usage,
Some(true), None, true,
&pinned_goals_snapshot,
&app_handle,
).await;
return;
}
// global/per_conv permit 已上移 loop 入口整 loop 持有,
// stream 后不再立即释放(会话级并发语义:工具执行期间也占槽)。per-provider permit
// (_provider_permit)仍在 candidate 循环内随作用域 Drop 自动释放。
// 累加本轮 token:用 provider 真实 usage。GLM 等流式不报 completion_tokens 时为 0,如实反映(不预估)。
// 分项 token(2026-08-02):构造本轮完整 usage(prompt 用 estimated 兜底,cache/reasoning 透传 round),
// add_usage 一次性累加 prompt/completion/cache/reasoning 到 tokens 累加器。
let round_prompt = if round_usage.prompt_tokens == 0 { estimated_prompt } else { round_usage.prompt_tokens };
let round_usage_full = df_ai::provider::TokenUsage {
prompt_tokens: round_prompt,
completion_tokens: round_usage.completion_tokens,
total_tokens: if round_usage.prompt_tokens == 0 {
estimated_prompt.saturating_add(round_usage.completion_tokens)
} else {
round_usage.total_tokens
},
prompt_cache_hit_tokens: round_usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: round_usage.prompt_cache_miss_tokens,
reasoning_tokens: round_usage.reasoning_tokens,
};
tokens.add_usage(&round_usage_full);
// 追加 assistant 消息到历史 + G1 目标提取(扁平重构,原嵌套 8 层 → 3 层)
let has_tool_calls = !tool_calls_acc.is_empty();
{
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
tracing::warn!(
stale_conv = %conv_id,
"[ai] stream 后 conv 已删除,丢弃本轮 push"
);
return;
}
push_assistant_message(
&mut session, &conv_id, has_tool_calls, &tool_calls_acc,
&full_text, &resolved_model, &last_reasoning_content,
round_prompt, round_usage.completion_tokens,
round_usage.prompt_cache_hit_tokens, round_usage.prompt_cache_miss_tokens,
round_usage.reasoning_tokens,
);
if GOAL_PIN_ENABLED {
update_pinned_goals(&mut session, &conv_id, &tool_calls_acc);
}
}
// P0-2 根治:LLM 回复 push 后、工具执行前立即落库。
// 背景:save_conversation 原仅在 loop 出口调用,process_tool_calls 卡住时(60s timeout
// 或 session lock 竞争)永远到不了出口,用户重启后上轮回复丢失。此处出 push 锁作用域后
// 立即 save,幂等(每轮重复覆盖落库),即使后续工具卡住本轮消息已持久化。
{
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&resolved_model), true).await;
}
// 停止信号:已生成文本入库后退出,不再执行后续工具调用
if stop_flag.load(Ordering::SeqCst) {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
// 统一走 finish_round_exit:save(Some usage, Some model) + spawn_title + emit(None,None,publish=true)
finish_round_exit(
&session_arc, &db, &conv_id,
Some(&usage), Some(&resolved_model),
true,
&provider_config, &llm_concurrency,
&mut guard,
&usage,
None, None, true,
&pinned_goals_snapshot,
&app_handle,
).await;
return;
}
// 无工具调用 → 最终文本响应,正常收敛退出
#[allow(unused_assignments)]
if !has_tool_calls { converged = true; break; }
let _ = converged;
// 处理工具调用(Low 自动执行 / Medium+High 待审批)
let pending_count = process_tool_calls(&session_arc, tool_calls_acc, &tools_arc, &db, &app_handle, &conv_id).await;
// DIRAUTH 审批链已闭环。原 eprintln 诊断降级为 tracing::debug,
// 避免污染 stderr(用户可见),保留排障能力(RUST_LOG=debug 可见)。
tracing::debug!(
conv_id = %conv_id,
pending_count,
"[AI-DIRAUTH-DIAG] agentic loop 收到 pending"
);
// L1 断路器:连续同类工具失败熔断(治 agent 无止损死循环,机制非 prompt 说教)。
// count_recent_failures 读末尾连续 Tool 消息,失败包按结构化信号归一 key 入滚动窗口计数。
if CIRCUIT_BREAKER_ENABLED {
let (max_count, sample_key) = count_recent_failures(&session_arc, &conv_id, &mut fail_window).await;
// 锁已随作用域 drop,可安全 await/emit(避免持锁 await 死锁)。
if max_count >= CIRCUIT_BREAKER_THRESHOLD {
tracing::warn!(
conv_id = %conv_id,
max_count,
sample_key = %sample_key,
"[ai] L1 断路器熔断:连续同类失败 {} 次,疑似死循环停止", max_count
);
guard.reset().await;
emit_circuit_breaker_tripped(&app_handle, &conv_id, max_count, &sample_key);
return;
}
}
// G2 探索熔断:连续空结果无进展检查。抽至 check_stall_breaker helper(扁平重构)。
// 返回 true=已熔断(loop 应 return),false=继续。stall_count/stall_warned 传入传出。
if STALL_BREAKER_ENABLED {
let (tripped, new_count, new_warned) = check_stall_breaker(
&session_arc, &app_handle, &conv_id, iteration,
stall_count, stall_warned,
).await;
stall_count = new_count;
stall_warned = new_warned;
if tripped {
guard.reset().await;
return;
}
}
// 有待审批 → 暂停循环,等待用户审批后通过 ai_approve → try_continue_agent_loop 恢复
if pending_count > 0 {
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&resolved_model), true).await;
// 审批等待 return 前 disarm guard——保持 generating=true 留 try_continue 续生成,
// 同时 Drop 因 done=true 跳过复位 spawn(避免误复位审批态 generating 致 ai_approve→try_continue 不续)
guard.disarm();
return; // generating 保持 true
}
// 全部自动执行完成 → 继续下一轮
}
// 达 MAX 未收敛(LLM 末轮仍想调工具被截断,末轮 tool_result 不再回传 LLM):转入暂停态询问用户
// emit AiMaxRoundsReached + 保持 generating=true(仿审批等待),
// 等用户点继续(ai_continue_loop → try_continue_agent_loop 再跑 max_iterations 轮)
// 或点停止(ai_stop_loop → 走完成流程)。try_continue 续跑 iteration 由调用方传 start_iteration 决定:
// 审批续跑累计(防多次审批反复跑满 max 致 token 失控,传 session.iteration_used),
// 达 max 续跑重计(用户点继续=授权重来,传 0 + 重置 iteration_used)。
// 注:max_iterations=0(不限)时 effective_max=usize::MAX,for 不会正常结束至此,故不触发暂停(靠 stop/收敛/审批退出)
if !converged {
tracing::warn!(
conv_id = %conv_id,
max_iter = max_iterations,
"[ai] agentic 循环达最大轮次仍未收敛,自动完成(incomplete=true)",
);
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
// 统一走 finish_round_exit:save(Some usage, Some model) + 不 spawn_title(对齐原无 title) +
// emit(Some(true), publish=false)。**do_publish=false 保留原 max_iterations 不 publish 行为**
// (与其他 5 路径不一致是历史现状,本次仅收敛重复代码不改 publish 策略,语义零变更)。
finish_round_exit(
&session_arc, &db, &conv_id,
Some(&usage), Some(&resolved_model),
false,
&provider_config, &llm_concurrency,
&mut guard,
&usage,
Some(true), None, false,
&pinned_goals_snapshot,
&app_handle,
).await;
return;
}
// 正常完成
let usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: tokens.total(),
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
// 落库 + 标题 + 知识提炼打包后台化:不阻塞 generating 复位与 Completed 事件
// save 先行(extract/title 都读已落库消息);extract 内部 fire-and-forget,与 title 可能并发
// (均受 per_conv 信号量约束,读写不同字段互不干扰)
// 并发取舍:与新对话新 loop 的 save 存在低概率并发 upsert,最多丢少量 token 累加(非功能错误,可接受)
let usage_total = usage.total_tokens;
{
let session_arc = session_arc.clone();
let db = db.clone();
let conv_id = conv_id.clone();
let provider_config = provider_config.clone();
let knowledge_config = knowledge_config.clone();
let app_handle = app_handle.clone();
let llm_concurrency = llm_concurrency.clone();
let resolved_model = resolved_model.clone();
tauri::async_runtime::spawn(async move {
save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&resolved_model), true).await;
// 知识提炼:需读已落库的对话消息,故在 save 之后
if let Err(e) = maybe_spawn_extraction(&session_arc, &db, &conv_id, &provider_config, &knowledge_config, llm_concurrency.clone()).await {
tracing::warn!("知识提炼触发失败(非阻断): {}", e);
}
ensure_conversation_title(&provider_config, &db, &conv_id, &app_handle, &session_arc, llm_concurrency).await;
});
}
guard.reset().await;
// generating 复位后再 emit Completed:落库/标题/提炼已在后台,前端立即感知完成
// (正常完成路径 save+extract+title 已在上方 spawn 异步,此处仅 emit,故直接调 emit_ai_completed_once)
// emit_usage=tokens 快照(prompt/completion/total 全从 tokens 取,对齐原 usage_total=tokens.total())
let normal_usage = df_ai::provider::TokenUsage {
prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(),
total_tokens: usage_total,
prompt_cache_hit_tokens: tokens.cache_hit(),
prompt_cache_miss_tokens: tokens.cache_miss(),
reasoning_tokens: tokens.reasoning(),
};
emit_ai_completed_once(
&app_handle, &conv_id, &normal_usage,
None, None, true,
&pinned_goals_snapshot,
).await;
}
// ── heartbeat_loop: 后台心跳任务(emit AiHeartbeat 防前端 watchdog 误杀) ──
// 退出条件:heartbeat_rx 发送端 Drop(function 作用域结束)。每 20s emit 一次。
async fn heartbeat_loop(
heartbeat_rx: &mut tokio::sync::watch::Receiver<()>,
app: &AppHandle,
conv_id: &str,
) {
loop {
tokio::select! {
_ = heartbeat_rx.changed() => break,
_ = tokio::time::sleep(std::time::Duration::from_secs(20)) => {
let _ = app.emit(
"ai-chat-event",
AiChatEvent::AiHeartbeat { conversation_id: Some(conv_id.to_string()) },
);
}
}
}
}
// ── count_recent_failures: L1 断路器失败计数(扁平抽自原嵌套 5 层块) ──
// 读末尾连续 Tool 消息,失败包按结构化信号(tool_name + 错误类别)归一 key,
// 追加进滚动窗口 fail_window(超窗 CIRCUIT_BREAKER_WINDOW 淘汰最旧)。
// 返回 (max_count, sample_key):窗口内同 key 最高频次 + 该 key;max_count=0 表示无失败。
//
// 两处相对原实现的关键修正(原 key=内容前 40 字符 + 全 loop 累加不衰减):
// 1. key 归一:失败 envelope 内容(不同路径/参数)前 40 字符易把同类失败拆成多 key
// (治不住)或异类失败误聚成一类(误熔断)。改读结构化信号——
// is_failure_content 已结构化判成败,本函数按 tool_name + 错误类别(exit_code/错误关键词
// 哈希)归一 key,同一工具的同类错误稳定聚到同一 key。
// 2. 滚动窗口:全 loop 累加会让早期偶发失败与后期叠加触发误熔断。VecDeque 仅保留最近 N 条,
// 计数只反映最近失败密度——长任务中途偶发 1~2 次不触发,真正连续死循环才触发。
async fn count_recent_failures(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
fail_window: &mut std::collections::VecDeque<String>,
) -> (u32, String) {
let messages = {
let session = session_arc.lock().await;
match session.conv_read(conv_id) {
Some(conv) => conv.messages.all_messages_clone(),
None => Vec::new(),
}
};
// 末尾连续 Tool 消息(本轮工具回填结果)。
let recent_tool_results: Vec<&ChatMessage> = messages
.iter().rev()
.take_while(|m| matches!(m.role, MessageRole::Tool))
.collect();
if recent_tool_results.is_empty() {
return (0u32, String::new());
}
// tool_call_id → tool_name 反查表:遍历 Assistant 消息的 tool_calls(开销与消息数线性)。
// tool_result 消息只带 tool_call_id,工具名在其前驱 Assistant 头里。
let mut id_to_name: std::collections::HashMap<&str, &str> = std::collections::HashMap::new();
for m in &messages {
if let MessageRole::Assistant = m.role {
if let Some(calls) = m.tool_calls.as_ref() {
for c in calls {
id_to_name.insert(c.id.as_str(), c.function.name.as_str());
}
}
}
}
// 本轮失败的归一 key 追加进窗口(顺序:消息正序,即工具执行顺序)。
for m in recent_tool_results.into_iter().rev() {
if is_failure_content(&m.content) {
let tool_name = m
.tool_call_id
.as_deref()
.and_then(|id| id_to_name.get(id).copied())
.unwrap_or("unknown_tool");
let key = failure_key(tool_name, &m.content);
fail_window.push_back(key);
while fail_window.len() > CIRCUIT_BREAKER_WINDOW {
fail_window.pop_front();
}
}
}
// 窗口内同 key 频次最高者 = 熔断候选。
let mut freq: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
for k in fail_window.iter() {
*freq.entry(k.as_str()).or_insert(0) += 1;
}
freq.into_iter()
.max_by_key(|(_, v)| *v)
.map(|(k, v)| (v, k.to_string()))
.unwrap_or((0u32, String::new()))
}
/// 归一失败 key:tool_name + 错误类别。保证同一工具的同类错误稳定映射到同一 key。
///
/// - run_command 等带 exit_code → `tool_name::exit=N`(非零退出码即失败类别)。
/// - status=failed/error envelope → `tool_name::error:<错误文本稳定哈希>`:
/// 用 fxhash 风格简单 FNV-1a 哈希避免裸截内容前缀(前 40 字符易把同类失败拆多 key
/// 或异类误聚)。哈希值非敏感信息(仅断路器分组用),无碰撞顾虑(N=20 窗口内几无碰撞)。
/// - 内容解析失败(非 JSON)→ is_failure_content 已返回 false 不会进本函数,
/// 此处兜底返回 `tool_name::unknown` 防御。
fn failure_key(tool_name: &str, content: &str) -> String {
let Ok(v) = serde_json::from_str::<serde_json::Value>(content) else {
return format!("{}::unknown", tool_name);
};
if let Some(exit) = v.get("exit_code").and_then(|x| x.as_i64()) {
return format!("{}::exit={}", tool_name, exit);
}
// 取 error 字段文本做稳定哈希;无 error 字段则按 status 兜底。
let bucket = match v.get("status").and_then(|s| s.as_str()) {
Some("failed") => "failed",
Some("error") => "error",
_ => "other",
};
let err_text = v
.get("error")
.and_then(|e| e.as_str())
.unwrap_or("")
.trim();
if err_text.is_empty() {
return format!("{}::{}", tool_name, bucket);
}
let h = fnv1a_32(err_text.as_bytes());
format!("{}::{}:{:08x}", tool_name, bucket, h)
}
/// FNV-1a 32-bit 哈希(零依赖,纯函数,断路器 key 归一专用;非密码学用途)。
fn fnv1a_32(bytes: &[u8]) -> u32 {
let mut hash: u32 = 0x811c9dc5;
for &b in bytes {
hash ^= b as u32;
hash = hash.wrapping_mul(0x0100_0193);
}
hash
}
// ── is_failure_content: 纯结构化判定(只读字段,绝不解析内容文本) ──
// 原理:工具成败是执行层的结构化事实(exit_code / status),断路器只读字段。内容文本(无论含
// error/失败/任何词)绝不参与判定 —— 这样读含 error 字样的代码、搜"失败"的结果等成功工具内容
// 永远不会被判失败(根除 conv 09e7abfa 的误熔断)。
// 失败 content 由 audit/mod.rs 包成 JSON envelope {"status":"failed","error":...};
// 成功 content 是 result JSON(run_command 含 exit_code,其他工具无 status 字段即成功)或 namespace:// 占位符。
// 非 JSON(namespace 占位符 / 重试 guard skip 纯文本)→ 无结构化失败信号 → 不判(false)。
fn is_failure_content(content: &str) -> bool {
let Ok(v) = serde_json::from_str::<serde_json::Value>(content) else {
return false; // namespace:// 占位符 / 重试 guard skip 等 → 无结构化失败信号,不判
};
if let Some(exit) = v.get("exit_code").and_then(|x| x.as_i64()) {
return exit != 0; // run_command 等:非零退出 = 真失败
}
matches!(v.get("status").and_then(|s| s.as_str()), Some("failed") | Some("error"))
|| v.get("error").is_some()
}
// ── summarize_tool_results: tool_result view-only 摘要压缩(扁平抽自原嵌套 map 块) ──
// 超阈值的 Tool 消息用 extract_key_info 压缩 content;非 Tool / 未超阈值 原样返回。
// 仅改入参 Vec(clone),不改 ContextManager 持久化(下次 build_for_request 从 DB 全量重建)。
fn summarize_tool_results(
messages: Vec<ChatMessage>,
history_tokens_snapshot: u32,
conv_id: &str,
iteration: usize,
) -> Vec<ChatMessage> {
let est = TokenEstimator::default();
let mut compressed_bytes: usize = 0;
let mut original_bytes: usize = 0;
let mut summarized_count: usize = 0;
// 构建 tool_call_id -> tool_name 映射:role=Tool 消息只带 tool_call_id(调用 ID),
// 真实工具名(read_file/run_command/...)在前序 role=Assistant 消息的 tool_calls[].function.name。
// 旧实现误把 tool_call_id(如 call_f892)当 tool_name 传给 extract_key_info,导致:
// 1) 摘要头注释显示 call_f892 而非 read_file(显示 bug);
// 2) extract_key_info 的 read_file 豁免永远命中不了(传入的不是真实名)。
// 用 owned String(非 &str 借用)避开与 messages.into_iter() 的所有权冲突;
// tool_call 数量极少(单轮几次),克隆开销可忽略。
let mut id_to_name: std::collections::HashMap<String, String> = std::collections::HashMap::new();
for m in &messages {
if let Some(calls) = m.tool_calls.as_ref() {
for c in calls {
id_to_name.insert(c.id.clone(), c.function.name.clone());
}
}
}
let msgs: Vec<ChatMessage> = messages
.into_iter()
.map(|mut m| {
if !matches!(m.role, df_ai::provider::MessageRole::Tool) {
return m;
}
let content_len = m.content.len();
let content_tokens = est.estimate_text(&m.content);
if !should_summarize_tool_result(content_len, history_tokens_snapshot, content_tokens) {
return m;
}
// 解析真实工具名:tool_call_id 查表;查不到(异常/老数据)退化为 "tool"。
let resolved_name = m
.tool_call_id
.as_ref()
.and_then(|id| id_to_name.get(id))
.map(|s| s.as_str())
.unwrap_or("tool");
// read_file 豁免压缩:AI 定向读代码,压缩 content 阉割代码分析能力。
// read_file handler 自带 limit 硬上限 2000 行,无爆 prompt 风险。
if resolved_name == "read_file" {
tracing::debug!(
conv_id = %conv_id, iteration,
content_len,
"[ai] read_file 工具结果豁免压缩(AI 定向读代码)"
);
return m;
}
original_bytes += content_len;
let compressed = extract_key_info(&m.content, resolved_name);
compressed_bytes += compressed.len();
summarized_count += 1;
m.content = compressed;
m
})
.collect();
if summarized_count > 0 {
tracing::info!(
conv_id = %conv_id, iteration, summarized_count, original_bytes, compressed_bytes,
"[ai] tool_result view-only 摘要压缩(不改持久化)"
);
}
msgs
}
// ── emit_circuit_breaker_tripped: L1 断路器熔断统一 emit(扁平抽自原嵌套块) ──
// 开关 CIRCUIT_BREAKER_HELP_EVENT: true=求助卡(默认), false=AiError 错误气泡(兜底回退)。
fn emit_circuit_breaker_tripped(app_handle: &AppHandle, conv_id: &str, max_count: u32, sample_key: &str) {
let ev = if CIRCUIT_BREAKER_HELP_EVENT {
AiChatEvent::AiHelpRequired {
reason: format!("连续同类失败 {} 次,疑似死循环已停止", max_count),
context: format!("最近错误: {}", sample_key),
options: vec!["换思路".into(), "授权路径".into(), "人工接管".into()],
conversation_id: Some(conv_id.to_string()),
}
} else {
AiChatEvent::AiError {
error: format!(
"连续同类失败 {} 次,疑似死循环已停止。请换思路或人工介入。最近错误: {}",
max_count, sample_key
),
error_type: Some(ErrorType::Unknown),
conversation_id: Some(conv_id.to_string()),
}
};
let _ = app_handle.emit("ai-chat-event", ev);
}
// ── check_stall_breaker: G2 探索熔断检查(扁平抽自原嵌套 6 层块) ──
// 返回 (tripped, new_stall_count, new_stall_warned):
// - tripped=true → 已 emit AiHelpRequired 且 loop 应 return(调用方需先 guard.reset 再 return)
// - tripped=false → 正常继续,新 stall_count/warned 供调用方回写。
//
// 2026-08-01 根本性重构:判定从「末尾 Tool 结果是否空(is_empty_tool_result)」改为
// 「最近 N 个 assistant tool_calls 签名是否重复(tool_call_signature + is_repetitive_exploration)」。
// 治旧范式误熔断正常排除式搜索(grep 无匹配是有效排除信号,非漂移)。熔断骨架(计数/警示/
// 阈值/emit/return)不变,只换判定输入。
//
// 嵌套 ≤ 3 层:取签名→计数/警示→达阈值 emit,各段早返回。
async fn check_stall_breaker(
session_arc: &Arc<Mutex<AiSession>>,
app_handle: &AppHandle,
conv_id: &str,
iteration: usize,
stall_count: u32,
stall_warned: bool,
) -> (bool, u32, bool) {
// 1) 取最近 N 个 assistant tool_calls 签名:倒序扫 messages,遇 Assistant 且有 tool_calls
// 就抽取每工具签名(name + arguments JSON parse),累积到 STALL_BREAKER_SAMPLE_SIZE 个。
// arguments 非合法 JSON 时 to_string 兜底(签名仍可比,只是兜底全量)。
let signatures: Vec<String> = {
let session = session_arc.lock().await;
let messages = match session.conv_read(conv_id) {
Some(conv) => conv.messages.all_messages_clone(),
None => Vec::new(),
};
let mut sigs: Vec<String> = Vec::new();
for m in messages.iter().rev() {
if sigs.len() >= STALL_BREAKER_SAMPLE_SIZE {
break;
}
if !matches!(m.role, MessageRole::Assistant) {
continue;
}
if let Some(tool_calls) = m.tool_calls.as_ref() {
// 单 assistant 消息内多个 tool_call 按声明顺序取,倒序累积(近期在前)。
for tc in tool_calls.iter().rev() {
if sigs.len() >= STALL_BREAKER_SAMPLE_SIZE {
break;
}
let args: serde_json::Value =
serde_json::from_str(&tc.function.arguments).unwrap_or(serde_json::Value::Null);
sigs.push(tool_call_signature(&tc.function.name, &args));
}
}
}
// 倒序收集的签名需反转为时间正序,保持 is_repetitive_exploration 的语义稳定
// (虽纯函数对顺序不敏感,但反转后日志/调试更直观)。
sigs.reverse();
sigs
};
// 本轮无工具调用(纯文本轮或首轮)→ 不计入也不重置,保持 stall_count。
if signatures.is_empty() {
return (false, stall_count, stall_warned);
}
// 2) 签名重复判定 → 更新 stall_count / stall_warned(沿用原 reset 语义)
let (mut new_count, mut new_warned) = (stall_count, stall_warned);
if is_repetitive_exploration(&signatures) {
new_count += 1;
} else {
new_count = 0;
new_warned = false;
}
// 3) 两段式警示:达 THRESHOLD-1 首次警示(给 LLM 自纠机会)
if STALL_BREAKER_WARN_FIRST
&& new_count == STALL_BREAKER_THRESHOLD.saturating_sub(1)
&& !new_warned
{
new_warned = true;
insert_stall_warning(session_arc, conv_id, iteration, new_count).await;
}
// 4) 熔断:达 THRESHOLD → emit AiHelpRequired + 返回 tripped=true
if new_count < STALL_BREAKER_THRESHOLD {
return (false, new_count, new_warned);
}
tracing::warn!(
conv_id = %conv_id,
stall_count = new_count,
sample_size = signatures.len(),
"[ai] G2 探索熔断:连续 {} 次检测到重复工具调用,疑似卡住停止", new_count
);
let _ = app_handle.emit(
"ai-chat-event",
AiChatEvent::AiHelpRequired {
reason: format!("连续 {} 次检测到重复的工具调用(同样的 grep/读文件反复),疑似卡住已停止", new_count),
context: "工具调用签名高度重复,可能在反复做同一件事而无进展。".into(),
options: vec!["回顾目标".into(), "换思路".into(), "停止".into()],
conversation_id: Some(conv_id.to_string()),
},
);
(true, new_count, new_warned)
}
// ── insert_stall_warning: G2 软提示 insert system 消息(给 LLM 自纠机会) ──
async fn insert_stall_warning(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
iteration: usize,
stall_count: u32,
) {
let goal_text = if STALL_BREAKER_GOAL_REMIND {
fetch_goal_summary(session_arc, conv_id).await
} else {
String::new()
};
let warn_text = format!(
"⚠ 检测到重复的工具调用(同样的 grep/读文件反复),可能卡住{},请回顾目标换思路或停止探索。",
goal_text
);
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(conv_id) {
let conv = session.conv(conv_id);
conv.messages.insert_at(0, ChatMessage::system(&warn_text));
tracing::info!(
conv_id = %conv_id, iteration, stall_count,
"[ai] G2 探索熔断:重复调用警示已 insert(软提示,给 LLM 自纠机会)"
);
}
}
// ── fetch_goal_summary: 读 pinned_goals 拼接为 "(当前目标: a; b)" 或空串 ──
async fn fetch_goal_summary(session_arc: &Arc<Mutex<AiSession>>, conv_id: &str) -> String {
let goals = session_arc
.lock().await
.conv_read(conv_id)
.map(|c| c.pinned_goals.clone())
.unwrap_or_default();
if goals.is_empty() {
return String::new();
}
let goal_summary = goals.iter()
.map(|g| g.text.trim())
.filter(|g| !g.is_empty())
.collect::<Vec<_>>()
.join("; ");
format!("(当前目标: {})", goal_summary)
}
// ── emit_fatal_error: 致命错误统一 emit AiError(扁平抽自原 Fatal/Exhausted 双分支重复代码) ──
// emit+publish 双写 Tauri 事件总线。error_type=Network 对齐原 Fatal 分支(鉴权/4xx归类)。
async fn emit_fatal_error(app_handle: &AppHandle, conv_id: &str, error: &str) {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: error.to_string(),
error_type: Some(ErrorType::Network),
conversation_id: Some(conv_id.to_string()),
});
// L3 emit 双写:Fatal AiError publish 到事件总线(门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: error.to_string(),
error_type: None,
conversation_id: Some(conv_id.to_string()),
});
}
// ── emit_ai_completed_once: AiCompleted 事件 emit+publish 双写 helper(对齐 emit_fatal_error 模式) ──
//
// 抽自 6 处退出路径(stop_flag 入口 / Coordinator merge / MidStream 保文 / push 后 stop /
// max_iterations / 正常完成)中「emit AiCompleted + publish_event AiCompleted」近重复代码。
//
// **incomplete 语义分叉**(memory: 语义不变):
// MidStream 保文路径 emit 端 incomplete=Some(true)(前端据 标记不完整),publish 端 incomplete=None
// (事件总线消费方无需此标记);max_iterations 路径仅 emit 不 publish(原实现即无 publish,保留)。
// 其余 4 路径 emit/publish 的 incomplete 一致(均 None 或均 Some)。
// 故本 helper 拆 emit_incomplete / publish_incomplete / do_publish 三参,精确镜像原各路径差异。
//
// usage 字段(对齐 run_agentic_loop 各退出点用法):emit 端的 token 三元组从 `usage` 取;
// prompt/completion 与 usage 一致(各退出点原样从 tokens 或 round-derived usage 传入)。
//
// pinned_goals:前端直接读取刷新(G1 目标钉扎),原样透传快照(不 clone,借用调用方)。
async fn emit_ai_completed_once(
app_handle: &AppHandle,
conv_id: &str,
usage: &df_ai::provider::TokenUsage,
emit_incomplete: Option<bool>,
publish_incomplete: Option<bool>,
do_publish: bool,
pinned_goals: &[super::GoalEntry],
) {
// emit 端(前端通道):incomplete 用 emit_incomplete(MidStream 传 Some(true))。
// token 分项(2026-08-02):cache_hit/cache_miss/reasoning 透传前端分计费展示。
let _ = app_handle.emit(
"ai-chat-event",
AiChatEvent::AiCompleted {
total_tokens: usage.total_tokens,
prompt_tokens: usage.prompt_tokens,
completion_tokens: usage.completion_tokens,
prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens,
reasoning_tokens: usage.reasoning_tokens,
incomplete: emit_incomplete,
conversation_id: Some(conv_id.to_string()),
pinned_goals: pinned_goals.to_vec(),
},
);
// publish 端(事件总线):EVENT_BUS_ENABLED 门控在 publish 内;incomplete 用 publish_incomplete
// (MidStream 传 None);do_publish=false(max_iterations)跳过整段 publish(保留原无 publish 行为)。
if do_publish {
let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(
AiChatEvent::AiCompleted {
total_tokens: usage.total_tokens,
prompt_tokens: usage.prompt_tokens,
completion_tokens: usage.completion_tokens,
prompt_cache_hit_tokens: usage.prompt_cache_hit_tokens,
prompt_cache_miss_tokens: usage.prompt_cache_miss_tokens,
reasoning_tokens: usage.reasoning_tokens,
incomplete: publish_incomplete,
conversation_id: Some(conv_id.to_string()),
pinned_goals: pinned_goals.to_vec(),
},
);
}
}
// ── finish_round_exit: run_agentic_loop 收尾统一入口(抽自 5 处退出路径重复代码) ──
//
// 收敛各退出点的「save_conversation + spawn_ensure_title + guard.reset + emit AiCompleted」序列。
// 退出点(行为零变更):
// - stop_flag 入口:save(Some usage, None model) + spawn_title + reset + emit(None,None,publish=true)
// - Coordinator merge:save(None, None) + no title + reset + emit(None,None,publish=true)
// - MidStream 保文:save(Some usage, Some model) + spawn_title + reset + emit(Some(true),None,publish=true)
// - push 后 stop:save(Some usage, Some model) + spawn_title + reset + emit(None,None,publish=true)
// - max_iterations:save(Some usage, Some model) + no title + reset + emit(Some(true),-,publish=false)
//
// **不在范围**(模式异构,保留原状):
// - 正常完成(spawn 异步 save+extract+ensure_title 阻塞式,与 spawn_ensure_title 后台式不同):
// 仅复用 emit_ai_completed_once,不走本 helper。
// - Fatal / Exhausted:emit AiError(已由 emit_fatal_error 统一),不走本 helper。
// - stale_conv return:无 emit/guard 显式 reset(依赖 guard Drop),不走本 helper。
//
// 参数口径(对齐各退出点原代码):
// - save_usage/save_model:透传 save_conversation;Coordinator 传 (None, None) 仍会调 save(幂等)
// - emit_usage:emit AiCompleted 的 token 三元组来源(各退出点经 tokens.add/round-derived 后传入)
// - spawn_title:true → spawn_ensure_title(provider_config 后台生成,失败有 extract_title 兜底)
// - emit_incomplete/publish_incomplete/do_publish:透传 emit_ai_completed_once(见该 helper 文档)
async fn finish_round_exit(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
conv_id: &str,
save_usage: Option<&df_ai::provider::TokenUsage>,
save_model: Option<&str>,
spawn_title: bool,
provider_config: &AiProviderRecord,
llm_concurrency: &LlmConcurrency,
guard: &mut GeneratingGuard,
emit_usage: &df_ai::provider::TokenUsage,
emit_incomplete: Option<bool>,
publish_incomplete: Option<bool>,
do_publish: bool,
pinned_goals: &[super::GoalEntry],
app_handle: &AppHandle,
) {
// 落库: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(对齐原行为)
if spawn_title {
spawn_ensure_title(provider_config, db, conv_id, app_handle, session_arc, llm_concurrency);
}
// generating 复位(guard.reset 幂等):保证前端收 AiCompleted 时后端已 Idle(发送队列续发不被「正在生成中」拒绝)
guard.reset().await;
// generating 复位后再 emit Completed(对齐原各路径顺序)
emit_ai_completed_once(
app_handle, conv_id, emit_usage,
emit_incomplete, publish_incomplete, do_publish,
pinned_goals,
).await;
}
// ── push_assistant_message: 轮结束后向 session.messages 推入 assistant 消息(扁平抽自原嵌套块) ──
// has_tool_calls=true: assistant_with_tools(文本+工具调用占位), false 且文本非空: assistant(纯文本)。
// 两分支都不为空且都不需 push 时(no tool + empty text),返回(no-op)。
// 入参 session 需外层调用方持锁;本函数只做纯内存 mutate,无 await/emit,不会死锁。
// prompt_tokens/completion_tokens: 本轮 LLM 调用 token 用量(消息级持久化,解 reload/压缩/切会话后
// 历史 assistant 消息 token 不显)。两构造分支都设。
// 分项 token(2026-08-02):cache_hit/cache_miss/reasoning 透传自 round_usage,前端分计费展示。
fn push_assistant_message(
session: &mut AiSession,
conv_id: &str,
has_tool_calls: bool,
tool_calls_acc: &std::collections::HashMap<u32, super::ToolCallDraft>,
full_text: &str,
resolved_model: &str,
last_reasoning_content: &Option<String>,
prompt_tokens: u32,
completion_tokens: u32,
cache_hit: u32,
cache_miss: u32,
reasoning: u32,
) {
if has_tool_calls {
let mut order: Vec<u32> = tool_calls_acc.keys().copied().collect();
order.sort_unstable();
let ai_tool_calls: Vec<df_ai::provider::ToolCall> = order.iter()
.enumerate()
.map(|(pos, i)| {
let draft = &tool_calls_acc[i];
// CR-空 id 最终防御:流式 chunk 解析点(openai_helpers)已对 Some("") 生成
// fallback,但若 provider 流式完全不发 id 字段(chunk id=Noneaccumulate
// 不覆盖),draft.id 残留空串。此处按 sorted order 位置兜底
// `gen_stream_{pos}`,保证 push 到 messages 的 assistant tool_call.id 非空唯一
// (下游 audit/mod.rs:203 seen_ids 去重 + 工具结果按 tool_call_id 路由依赖此)。
// 非空 id 原样(与 chunk 解析点 fallback 共用 helperDRY)。
let id = df_ai::provider::tool_call_id_or_fallback(&draft.id, pos, "gen_stream");
df_ai::provider::ToolCall::new(id, &draft.name, &draft.args)
})
.collect();
let mut msg = ChatMessage::assistant_with_tools(full_text, ai_tool_calls);
msg.model = Some(resolved_model.to_string());
msg.reasoning_content = last_reasoning_content.clone();
msg.prompt_tokens = Some(prompt_tokens);
msg.completion_tokens = Some(completion_tokens);
msg.prompt_cache_hit_tokens = Some(cache_hit);
msg.prompt_cache_miss_tokens = Some(cache_miss);
msg.reasoning_tokens = Some(reasoning);
session.conv(conv_id).messages.push(msg);
} else if !full_text.is_empty() {
let mut msg = ChatMessage::assistant(full_text);
msg.model = Some(resolved_model.to_string());
msg.reasoning_content = last_reasoning_content.clone();
msg.prompt_tokens = Some(prompt_tokens);
msg.completion_tokens = Some(completion_tokens);
msg.prompt_cache_hit_tokens = Some(cache_hit);
msg.prompt_cache_miss_tokens = Some(cache_miss);
msg.reasoning_tokens = Some(reasoning);
session.conv(conv_id).messages.push(msg);
}
}
// ── update_pinned_goals: G1 目标钉扎提取+裁剪(扁平抽自原嵌套块,原 8 层 → 3 层) ──
// 从工具调用推理新目标 → 去重 add → 标记旧 active 为 completed → 超上限时淘汰最早 completed。
// 入参 session 需外层调用方持锁;纯内存 mutate,无 await。
fn update_pinned_goals(
session: &mut AiSession,
conv_id: &str,
tool_calls_acc: &std::collections::HashMap<u32, super::ToolCallDraft>,
) {
let new_goals = infer_goal_from_tool_calls(tool_calls_acc);
if new_goals.is_empty() {
return;
}
let goal_conv = session.conv(conv_id);
let mut added = false;
for g in &new_goals {
let is_dup = goal_conv.pinned_goals.iter().any(|e|
e.text.contains(g.as_str()) || g.contains(&e.text)
);
if !is_dup {
goal_conv.pinned_goals.push(super::GoalEntry::new(g.clone()));
added = true;
}
}
// 有新目标 → 标记之前的 active 为 completed(LLM 进入下一任务)
if added && goal_conv.pinned_goals.len() > new_goals.len() {
let active_until = goal_conv.pinned_goals.len() - new_goals.len();
for entry in goal_conv.pinned_goals.iter_mut().take(active_until) {
if matches!(entry.status, super::GoalStatus::Active) {
entry.status = super::GoalStatus::Completed;
}
}
}
// 超上限时保留 active,淘汰最早 completed
if goal_conv.pinned_goals.len() > MAX_GOALS {
let active: Vec<super::GoalEntry> = goal_conv.pinned_goals.iter()
.filter(|g| matches!(g.status, super::GoalStatus::Active))
.cloned().collect();
if active.len() <= MAX_GOALS {
let keep = MAX_GOALS - active.len();
let completed: Vec<super::GoalEntry> = goal_conv.pinned_goals.iter()
.filter(|g| matches!(g.status, super::GoalStatus::Completed))
.rev().take(keep).cloned().collect();
let mut merged = active;
merged.extend(completed.into_iter().rev());
goal_conv.pinned_goals = merged;
} else {
goal_conv.pinned_goals.truncate(MAX_GOALS);
}
}
}
// ── Plan-driven Phase 1:Plan→事件载荷 + plan_id 生成辅助 ──
//
// 对齐 docs/02-架构设计/专项设计/aichat-plan-driven-设计-2026-08-01.md Phase 1:
// LLM 出 Plan 后 emit AiPlanCreated(步骤 + 状态 pending)。Plan 是 DAG(kahn to_layers
// 分层),事件载荷按 layer 分组(同层并行,层间串行),前端据此画 DAG 卡片(Phase 3)。
/// 生成 plan_id(时间戳 + 计数器,无需 ulid 依赖)。
///
/// 形如 `plan-1757000000000-42`,单调递增 + 进程内唯一。与 df-ai-core 的 new_message_id
/// 同思路(AtomicU64 计数),但本处 plan_id 仅供事件载荷标识,无落库无溯源需求。
fn ulid_like_id() -> String {
use std::sync::atomic::{AtomicU64, Ordering as O};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let ts = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis())
.unwrap_or(0);
let n = COUNTER.fetch_add(1, O::SeqCst);
format!("{}-{}", ts, n)
}
/// 把 Plan(DAG)转为 AiPlanCreated 事件载荷(按 layer 分组)。
///
/// - 调 `plan.to_layers()`(kahn 拓扑分层)得层级顺序;
/// - 每层任务映射为 SubTaskInfo(id / persona_id / intent / status="pending");
/// - persona_id 由 `registry.recommend_for_intent` 推荐(对齐 dispatch 路径);
/// - to_layers 失败(环,正常不会发生——decompose_with_llm 已 validate 兜底)→
/// fallback 单层平铺(不阻断事件 emit)。
fn build_plan_layers_payload(
plan: &df_ai::planner::Plan,
coord: &Coordinator,
) -> Vec<super::PlanLayerInfo> {
let layers = plan.to_layers().unwrap_or_else(|_| {
// 环(unreachable:decompose_with_llm 已 validate 兜底)→ fallback 单层平铺
vec![plan.tasks.clone()]
});
layers
.into_iter()
.map(|layer| super::PlanLayerInfo {
items: layer
.into_iter()
.map(|t| {
let persona_id = coord.recommend_persona_id(&t.intent);
super::SubTaskInfo {
id: t.id,
persona_id,
intent: t.intent,
status: "pending".to_string(),
}
})
.collect(),
})
.collect()
}