Files
DevFlow/src-tauri/src/commands/ai/agentic/context_lifecycle.rs
T
lxy cb23061088 新增: 跨会话召回四层架构与脉络卡
- 会话终态提炼: 删除前+停靠双挂点,快照与清水位同临界区
- 目标快照: 恢复引导按最终有效意图生成,旧目标不复活
- 知识检索项目维度软排序,WorkingContext 持久化迁移
- 脉络卡: published 知识按项目聚合纯读侧,带入上下文经输入框显式注入
- 压缩与缓存能力在主循环/子代理/提示词三面接线
- 两份专项设计文档与看板销账
2026-10-11 22:27:20 +08:00

619 lines
34 KiB
Rust

//! Agentic 循环上下文生命周期:F-15 自动压缩(智能裁剪)。
//!
//! 从 `mod.rs::run_agentic_loop` 机械抽取,后经 L0-3 后台化(见下)。
//!
//! 触发条件:history_tokens > budget*0.6 且 保护区外有可压缩消息 且 未在压缩中。
//!
//! 流程(对齐 ai_chat_compress_context IPC 的 read-but-don't-mutate 模式):
//! ① 读 active 克隆(不改 status / 不扣 token)→ 后台喂 LLM 出摘要;
//! ② LLM 成功 → compress_old_messages(标 compressed + 扣 token)+ push(摘要 system 尾部追加);
//! ③ LLM 失败 → 消息状态完全不变(未改 status / 未扣 token),降级走原 build_for_request 裁剪。
//!
//! 口径决策(任务让"你判断"):选**延迟 mutate(成功才改)**而非"失败回滚 status"。
//! 理由:ContextManager.history_tokens 字段私有、无 set_history_tokens 公开接口;
//! 若先 compress_old_messages(扣 token)再 LLM,失败回滚需精确恢复 history_tokens,
//! 但 ChatMessage 克隆不含 token_count,无法等量加回——回滚 token 不精确。
//! 延迟 mutate 则失败时零副作用(消息状态/token 完全不变),语义最干净。
//! 注:延迟 mutate 的窗口(active_msgs 读出→LLM 出摘要期间)不持锁;L0-3 后台化后窗口拉长
//! (loop 继续推进,可能 push 新消息),但 loop 只 push 增长、protect_start 基于触发时 len,
//! 压缩区间始终落在当前轮之前,不误伤当前轮(见 maybe_auto_compress L0-3 注释)。
//!
//! 安全(L0-3 后台化后):provider 在后台 task 内经 `secret::build_provider_for` 重建
//! (不再复用 loop 顶部已 build 的 provider——后者是 Box 非 Clone 无法移入 task;多一次
//! keyring resolve 在后台线程不阻塞 loop),api_key 经 df_storage::secret 闭环;
//! summary/error payload/日志均不含 api_key。
//! is_compressing 防重入:set_compressing(true) 在触发时置位,后台 task 各出口
//! (成功/失败/conv 已删/provider 重建失败)均复位,防重入不泄漏。
//! 单轮问答(history_tokens 未超 0.6*budget)不触发,零行为变化。
//! messages 操作改 per_conv。
//!
//! L0-3(2026-08-09):压缩 LLM 调用改后台 fire-and-forget(tokio::spawn),不再同步阻塞本轮
//! loop(非流式 LLM 最长 60s)。当前轮用未压缩上下文正常走(不丢回复),压缩结果下一轮生效。
use std::path::Path;
use std::sync::Arc;
use tauri::{AppHandle, Manager};
use tokio::sync::Mutex;
use df_ai::context::ContextManager;
use df_ai::context_helpers::{extract_keyword_summary, CompressedSummary};
use df_ai::provider::ChatMessage;
use df_storage::models::AiProviderRecord;
use crate::state::{AppState, LlmConcurrency};
// compress_via_llm 位于 commands::ai::compress(mod.rs 的 super::compress)。
use super::super::compress::compress_via_llm;
use super::super::event_bus::emit_ai_chat_event;
use super::{AiChatEvent, AiSession};
/// 压缩保护区条数(对齐 clear/compress IPC 的 PROTECT_COUNT=6)。
/// protect_start = len.saturating_sub(PROTECT_COUNT):保护区内的最近 6 条(含本轮三元组)
/// 不参与压缩,避免压缩正在使用的活跃消息。
const PROTECT_COUNT: usize = 6;
/// 改进3 B 常量开关:压缩失败时是否启用关键词摘要兜底(默认 true)。
///
/// true(默认):LLM 压缩失败 → compress_old_messages 标 compressed(释放 token)
/// + push(system 关键词摘要)尾部追加,保留用户反复提及的主题词作续接锚点。
/// false(回退):回退原裸裁剪行为(不插关键词摘要,仅 build_for_request 裁剪保最近 6 条)。
/// 排障/对比用:置 false 即可观察无兜底时的裁剪效果。
const KEYWORD_FALLBACK_ENABLED: bool = true;
/// 会话消息列表的内容版本哨兵(939⑪):首条消息的全局唯一 id(空列表 = None)。
///
/// 压缩**触发时**快照本值,后台 task **提交前**重新读取比对:相等才允许提交。
/// 首条消息 id 由 `new_message_id()` 全局唯一——loop 只 append(不改首条),而 `clear()`
/// 重建后首条必换新 id,故它是「上下文是否被整体替换」的准确且零成本哨兵。
/// 不用 len:loop 后台压缩期间正常 append 会使 len 增长,以 len 相等为守卫会误判版本
/// 不符致后台压缩永不生效。
fn head_id_of(messages: &mut ContextManager) -> Option<String> {
messages.messages_mut().first().and_then(|t| t.message.id.clone())
}
/// 内容版本是否仍匹配(939⑪):快照与当前首条 id 相等才允许提交。
fn version_matches(snapshot: &Option<String>, current: &Option<String>) -> bool {
snapshot == current
}
/// panic 兜底复位(939⑫):清 is_compressing 防其永真致压缩永久停摆。
///
/// conv 已删(per_conv 无条目)跳过——标志随 conv 销毁无意义(对齐各正常出口语义)。
async fn reset_compressing_after_panic(session_arc: &Arc<Mutex<AiSession>>, conv_id: &str) {
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(conv_id) {
session.conv(conv_id).messages.set_compressing(false);
}
}
/// 自动压缩(智能裁剪)——在 build_for_request 之前预处理。
///
/// **L0-3(2026-08-09)后台化**:压缩 LLM 调用不再同步阻塞本轮 loop(非流式 LLM 最长 60s)。
/// 触发时仅做零成本准备(读触发条件 / 读 active 克隆 / 算关键词兜底)后立即 `tokio::spawn`
/// 独立 task fire-and-forget 执行压缩,本函数返回 `false` 让当前轮**用未压缩上下文正常走**
/// (build_for_request 不受影响,当前轮回复不丢);压缩结果由后台 task 在完成时更新
/// per_conv.messages(标 compressed + 插摘要),**下一轮** build_for_request 生效。
/// `is_compressing` 防重入:触发时置 true,后台 task 各出口(成功/失败/conv 已删)均复位。
///
/// 返回值:
/// - `true`:conv 已删除,调用方应立即 `return`(退出整个 loop)。
/// - `false`:正常结束(无论是否触发压缩、压缩是否成功),调用方继续后续流程。
///
/// 参数沿用原内联代码引用的局部:
/// - `provider_config`:compress 沿用主 candidate(非 fallback 范围)。后台 task 内经
/// `secret::build_provider_for` 重建 provider(复用 loop 顶部同款三步打包,仅多一次
/// keyring resolve,在后台线程不阻塞 loop)。
/// - `iteration`:当前轮次(仅用于日志)。
pub(super) async fn maybe_auto_compress(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
app_handle: &AppHandle,
provider_config: &AiProviderRecord,
llm_concurrency: &LlmConcurrency,
iteration: usize,
sys_tokens: u32,
) -> bool {
let prev_compressing = session_arc.lock().await.conv_read(&conv_id).map(|c| c.messages.is_compressing()).unwrap_or(false);
if !prev_compressing {
// 读触发条件(history_tokens / budget / has_compressible_messages),持锁快照判断。
let (should_compress, protect_start, pre_compress_tokens) = {
let session = session_arc.lock().await;
let mgr = match session.conv_read(&conv_id) {
Some(c) => c,
None => {
tracing::warn!(stale_conv = %conv_id, "[ai] conv 已删除,loop 退出(压缩段入口)");
return true;
}
};
let mgr = &mgr.messages;
let protect_start = mgr.len().saturating_sub(PROTECT_COUNT);
let history_tokens = mgr.history_tokens();
let available = mgr.budget_limit().saturating_sub(sys_tokens);
// 触发阈值 0.6*available(可用预算=budget-sys_tokens):available*6/10 < history_tokens
let should = protect_start > 0
&& (available as u64) * 6 / 10 < history_tokens as u64
&& mgr.has_compressible_messages(protect_start);
(should, protect_start, history_tokens)
};
if should_compress {
// emit 压缩开始(emit+publish 双写经 emit_ai_chat_event)+ 置位防重入 + 读 active 克隆(不改 status)。
emit_ai_chat_event(app_handle, AiChatEvent::AiCompressing {
conversation_id: Some(conv_id.to_string()),
});
let (mut active_msgs, lang, head_id) = {
let mut session = session_arc.lock().await;
let conv = session.conv(&conv_id);
conv.messages.set_compressing(true);
// 读 active 克隆(不改 status):filter is_active,LLM 失败则消息状态完全不变。
// protect_start 是上一段锁内算的快照,两段锁之间消息可能被清空/裁剪变短,
// 切片前按当时实际长度钳制,防 index out of bounds panic
// (后续 compress_old_messages 内部自带 min(len) 钳制,口径一致)。
let msgs = conv.messages.messages_mut();
// 939⑪ 内容版本哨兵:快照触发时首条消息 id,后台 task 提交前比对
// (clear() 重建后闭包内提交会落到新上下文,靠此守卫丢弃)。
let head_id = msgs.first().and_then(|t| t.message.id.clone());
let cut = protect_start.min(msgs.len());
let active_msgs: Vec<ChatMessage> = msgs[..cut]
.iter()
.filter(|t| t.message.is_active())
.map(|t| t.message.clone())
.collect();
let lang = conv.agent_language.clone()
.unwrap_or_else(|| "zh-CN".to_string());
(active_msgs, lang, head_id)
};
// T2-修复(方案A):展开 namespace 引用为真实内容,再喂压缩 LLM/关键词摘要。
// 同 agentic/mod.rs 的 build_for_request 出口展开。未展开则压缩 LLM 只看到
// "namespace://..." 占位符,摘要基于占位符丢主题。read_only None(条目被 LRU 淘汰 /
// 重启后 namespace 清空 / 跨会话残留引用)→ 替换为 EVICTED_PLACEHOLDER 提示文案
// (非保留无意义 namespace:// URI)+ warn 日志。
{
let session = session_arc.lock().await;
for m in &mut active_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 = session
.namespace_store
.eviction_rich_notice(&path)
.unwrap_or_else(|| df_ai::namespace_store::EVICTED_PLACEHOLDER.to_string());
}
}
}
}
// 改进3 B:在 active_msgs move 进压缩前,先算关键词摘要兜底文本。
// LLM 压缩失败时仍想保留用户反复提及的主题词(续接锚点),避免裸裁剪丢主题。
// KEYWORD_FALLBACK_ENABLED=false → 跳过(回退原裸裁剪行为,排障/对比用)。
let keyword_fallback: String = if KEYWORD_FALLBACK_ENABLED {
extract_keyword_summary(&active_msgs)
} else {
String::new()
};
// 保护区外无 active 可压缩(active_msgs 空)→ noop,同步复位 is_compressing
// 后返回(不 spawn 空任务)。对齐原 Ok(None) 分支语义。
if active_msgs.is_empty() {
session_arc.lock().await.conv(&conv_id).messages.set_compressing(false);
return false;
}
// 压缩可召回开关(KV `df-ai-compress-recall`,默认开):开则压缩标记时把原文
// 存档 + 摘要尾部拼召回句柄。spawn 前读值拷贝,后台 task 不持 AppState 引用
//(State 借用非 'static,不可移入 task)。
let (recall_on, data_dir) = {
let st = app_handle.state::<AppState>();
(
super::super::prompt::compress_recall_enabled(&st).await,
st.data_dir.clone(),
)
};
// ── L0-3 后台化:压缩 LLM 调用改 fire-and-forget(tokio::spawn 独立 task) ──
//
// 原实现在此 `compress_via_llm(...).await` 同步阻塞整轮(非流式 LLM 最长 60s),
// 期间 loop 无法推进 build_for_request / stream,当前轮回复整体延迟。
//
// 现改为 spawn 独立 task 后台执行压缩,本函数立即返回 false:
// - 当前轮 build_for_request 用**未压缩上下文**,回复不受压缩阻塞影响(关键:不丢当前轮);
// - 后台 task 完成后更新 per_conv.messages(compress_old_messages + insert 摘要),
// 下一轮 build_for_request 生效;
// - is_compressing 防重入:触发时已置 true(上方),后台 task 各出口(成功/失败/conv 已删)
// 均复位,下一轮 maybe_auto_compress 检测到 true 即跳过,不会并发二次压缩。
//
// 选型理由(后台化 vs 降级独立线程):并发安全成立——所有 per_conv 访问经
// session_arc 锁(数据无竞争),且压缩只读「触发时刻快照的 active_msgs」+ 完成后
// 一次性 mutate,与 loop 的 push 均为「锁内短段」,protect_start 基于触发时 len
// (loop 只 push 增长,压缩区间始终落在当前轮之前),故选 tokio::spawn 后台化,
// 无需降级独立线程。provider 在 task 内重建(secret::build_provider_for),避免
// 与 loop 共享 Box<dyn LlmProvider>(后者非 Clone,无法移入 task)。
//
// 无感知降级:provider 重建失败 / LLM 失败 → 仅复位 is_compressing + 关键词兜底,
// 不发 AiError(前端 AiError 会误判生成结束致"压缩后停止"),loop 继续 stream。
let session_arc = session_arc.clone();
let app_handle = app_handle.clone();
let conv_id = conv_id.to_string();
let provider_config = provider_config.clone();
let llm_concurrency = llm_concurrency.clone();
// 939⑫ panic 兜底句柄:内层 async move 会消费 session_arc/conv_id,panic 分支
// 需另持一份用于复位 is_compressing(防其永真致压缩永久停摆)。
let panic_session = session_arc.clone();
let panic_conv_id = conv_id.clone();
tokio::spawn(async move {
// 939⑫ 整个后台 body 包 catch_unwind(复用 mod.rs::unwind_to_result 薄包装):
// panic 以 Err 返回 → 复位 is_compressing + warn,防压缩永久停摆。依赖
// panic=unwind(tokio 默认);panic=abort 构建下退化为进程终止(尽力兜底)。
let outcome = super::unwind_to_result(async move {
// 后台重建 provider(secret::build_provider_for 三步打包:resolve→ensure→build)。
// 失败 → 同 LLM 失败降级:复位 is_compressing + 关键词摘要兜底(标 compressed 释放
// token + 插关键词),保持「失败→兜底→扣 token」语义——否则 history_tokens 未降,
// 下轮 maybe_auto_compress 会重复触发(每次仅 spawn 后台任务,不阻塞 loop,但浪费)。
let provider = match super::super::secret::build_provider_for(&provider_config) {
Ok(p) => p,
Err(e) => {
tracing::warn!(
conv_id = %conv_id,
error = %e,
keyword_fallback_len = keyword_fallback.len(),
"[ai] 后台压缩 provider 重建失败,降级走关键词摘要兜底(KEYWORD_FALLBACK_ENABLED={})",
KEYWORD_FALLBACK_ENABLED,
);
// 复位 is_compressing 收口进 keyword_fallback_commit 单锁(909①-补:
// 原先此处单独取锁复位,与提交之间留并发二次压缩窗口)。
keyword_fallback_commit(
&session_arc,
&conv_id,
protect_start,
&keyword_fallback,
pre_compress_tokens,
"压缩 provider 重建失败",
&head_id,
recall_on.then_some(data_dir.as_path()),
)
.await;
return;
}
};
// 压缩调用(复用主 candidate 的 provider_config,api_key 经 secret 闭环)。
// 成功 → Some(summary);失败 → Err。
let compress_outcome: Result<Option<CompressedSummary>, String> =
compress_via_llm(
provider.as_ref(),
&provider_config,
active_msgs,
&lang,
&conv_id,
&llm_concurrency,
).await.map(Some);
match compress_outcome {
Ok(Some(summary)) => {
// LLM 成功 → 标 compressed(扣 token)+ 摘要 system 尾部追加 + set_compressing(false)。
// conv 已删(per_conv 无条目)时丢弃压缩结果——惰性重建会污染新 conv,
// is_compressing 标志随 conv 销毁无意义,直接 return。
{
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(&conv_id) {
return;
}
let conv = session.conv(&conv_id);
// 939⑪ 内容版本守卫:提交前比对触发时快照,不符(期间 clear() 重建致
// 上下文整体替换)→ 丢弃压缩结果。不复位 is_compressing(clear 已复位;
// 若新会话已启新压缩,此处复位会误伤新压缩信令)。
if !version_matches(&head_id, &head_id_of(&mut conv.messages)) {
tracing::warn!(
conv_id = %conv_id,
"[ai] 压缩完成但上下文版本已变(疑似 clear 重建),丢弃旧摘要"
);
return;
}
if recall_on {
conv.messages.set_compress_recall(&data_dir, &conv_id);
}
let _compressed = conv.messages.compress_old_messages(protect_start);
// 召回句柄拼在摘要尾部(存档落盘成功才有):LLM 需要原文时 read_file
// 回读;句柄行只进 system 锚点,前端事件仍用原 nl_summary 不外发。
let anchor_text = match conv.messages.take_archive_path() {
Some(p) => df_ai::context::archive::with_recall_line(&summary.nl_summary, &p),
None => summary.nl_summary.clone(),
};
// NL 摘要尾部追加(而非 insert_at(0) 头部插入):头部插入会位移整条
// 前缀字节,使 DeepSeek 等自动前缀缓存的 provider 全部缓存失效
// (命中率掉到个位数)。尾部追加紧邻本轮内容 LLM 仍可见,前缀稳定
// 命中率最大化。JSON 卡片回填 WorkingContext,决策随压缩重锚定。
conv.messages.push_system_anchor(ChatMessage::system(&anchor_text));
conv.messages.set_compressing(false);
conv.working_context.sync_after_compress(&summary.json_card);
}
tracing::info!(
conv_id = %conv_id,
iteration,
pre_tokens = pre_compress_tokens,
"[ai] 自动压缩成功(后台),摘要已尾部追加"
);
// 治 Task#1:loop 自动压缩用 AiAutoCompressed(非手动变体),
// 桌面端静默(仅复位 isCompressing,不弹 toast 不 switchConversation)。
// miniapp 仍插摘要气泡(对端发生压缩告知用户)。
emit_ai_chat_event(&app_handle, AiChatEvent::AiAutoCompressed {
conversation_id: Some(conv_id.clone()),
summary: summary.nl_summary,
});
}
Ok(None) => {
// 保护区外无 active 可压缩(已全 compressed/archived)→ noop,仅复位 is_compressing。
// 单次锁内 check+mutate(防 conv 在两次锁间被删致惰性重建污染新 conv)。
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(&conv_id) {
let conv = session.conv(&conv_id);
// 939⑪ 版本守卫:上下文已被整体替换 → 不复位(否则误伤新会话信令)。
if version_matches(&head_id, &head_id_of(&mut conv.messages)) {
conv.messages.set_compressing(false);
}
}
}
Err(e) => {
// LLM 失败 → 改进3 B:仍标 compressed 释放 token + 关键词摘要尾部追加(非裸裁剪)。
//
// 旧行为:消息状态完全不变,降级走 build_for_request 裁剪(丢主题)。
// 新行为(KEYWORD_FALLBACK_ENABLED=true 默认):
// - compress_old_messages 标 [..protect_start] active 为 compressed(释放 token,
// 与成功路径一致,后续 build_for_request 不再把它们进 LLM 上下文);
// - keyword_fallback 非空 → push(system 关键词摘要)尾部追加作续接锚点;
// - keyword_fallback 空(无 user 消息/无可提取词)→ 不插,等价旧裁剪(保底)。
// 持久化语义不变:compressed 仍软删可追溯(DB 全量保留),与成功路径一致。
// 压缩失败不发 AiError——前端 AiError case 无条件 setStreaming(false)
// 误判生成结束(后端 loop 实际继续),致"压缩后停止"。降级为 warn + 兜底释放
// token + build_for_request 裁剪,loop 继续 stream_llm,用户自然看到后续回复。
// 复位 is_compressing 收口进 keyword_fallback_commit 单锁(909①-补,
// 同 provider 重建失败路径;消锁外复位→再取锁提交的并发窗口)。
tracing::warn!(
conv_id = %conv_id,
error = %e,
keyword_fallback_len = keyword_fallback.len(),
"[ai] 自动压缩失败(后台),降级走关键词摘要兜底(KEYWORD_FALLBACK_ENABLED={})",
KEYWORD_FALLBACK_ENABLED,
);
keyword_fallback_commit(
&session_arc,
&conv_id,
protect_start,
&keyword_fallback,
pre_compress_tokens,
"压缩失败",
&head_id,
recall_on.then_some(data_dir.as_path()),
)
.await;
}
}
}) // 闭 catch_unwind 内 async move
.await;
if let Err(msg) = outcome {
reset_compressing_after_panic(&panic_session, &panic_conv_id).await;
tracing::warn!(
conv_id = %panic_conv_id,
error = %msg,
"[ai] 后台压缩 task panic,已复位 is_compressing(防压缩永久停摆)"
);
}
});
}
}
false
}
/// 关键词兜底提交单点(DRY #8 收敛,2026-10-01)。
///
/// 抽自 [`maybe_auto_compress`] 后台 task 内的逐字双胞胎块:压缩 provider 重建失败路径
/// 与压缩 LLM 失败路径共用「复位 is_compressing → KEYWORD_FALLBACK_ENABLED 门控 → 锁内
/// check+mutate(标 compressed 释放 token + 关键词摘要尾部追加,非空才插)→ 成功插入打 info」。
/// 两条失败路径只差日志正文,经 `log_head` 参数化(拼为
/// "[ai] {log_head}兜底(后台):关键词摘要已尾部追加(compressed 标记已扣 token)")。
///
/// **复位重入标志收口单锁(909①-补,2026-10-10)**:`set_compressing(false)` 与兜底提交在同一
/// 次锁内原子完成——原调用方在锁外单独取锁复位,与这里的提交之间留出窗口(期间的压缩触发读到
/// 的仍是 false,可并发二次压缩,放大楔入概率)。注意复位在 `KEYWORD_FALLBACK_ENABLED` 门控
/// **之前**:开关关闭时同样必须复位,否则 is_compressing 永真 → 压缩永久停摆。
///
/// conv 已删(per_conv 无条目)→ 直接返回(标志随 conv 销毁无需复位;对齐原 return 语义:
/// 防惰性重建污染新 conv)。
///
/// **内容版本守卫(939⑪)**:`head_id` 为触发时快照的首条消息 id;提交前比对不符(期间
/// `clear()` 重建致上下文整体替换)→ 直接返回:既不标 compressed / 不插旧关键词摘要污染
/// 新上下文,也不复位 is_compressing(clear 已复位;复位会误伤新会话可能已启的新压缩信令)。
///
/// `recall`:压缩可召回开关(Some = 开,携带存档 data_dir)。开则压缩标记时原文落盘存档,
/// 关键词摘要尾部拼召回句柄行;None 零行为变更(不落盘、摘要无句柄)。
async fn keyword_fallback_commit(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
protect_start: usize,
keyword_fallback: &str,
pre_tokens: u32,
log_head: &str,
head_id: &Option<String>,
recall: Option<&Path>,
) {
let inserted = {
let mut session = session_arc.lock().await;
if !session.per_conv.contains_key(conv_id) {
return;
}
let conv = session.conv(conv_id);
// 939⑪ 内容版本守卫:上下文已被整体替换 → 丢弃兜底提交(标志亦不复位)。
if !version_matches(head_id, &head_id_of(&mut conv.messages)) {
tracing::warn!(
conv_id = %conv_id,
"[ai] {}兜底提交前上下文版本已变(疑似 clear 重建),丢弃",
log_head
);
return;
}
// 复位与提交同锁原子(909①-补):消「锁外复位→再取锁提交」之间的并发二次压缩窗口。
conv.messages.set_compressing(false);
if !KEYWORD_FALLBACK_ENABLED {
// 开关关闭 → 回退原裸裁剪行为:不压缩不插摘要,仅完成复位。
false
} else {
// 压缩可召回:开(Some)则压缩标记时原文落盘存档,关键词摘要尾部拼召回句柄行。
if let Some(dir) = recall {
conv.messages.set_compress_recall(dir, conv_id);
}
let _compressed = conv.messages.compress_old_messages(protect_start);
if keyword_fallback.is_empty() {
false
} else {
// 召回句柄拼在摘要尾部(存档落盘成功才有),只进 system 锚点。
let anchor_text = match conv.messages.take_archive_path() {
Some(p) => df_ai::context::archive::with_recall_line(keyword_fallback, &p),
None => keyword_fallback.to_string(),
};
// 尾部追加(而非 insert_at(0) 头部插入):头部插入位移整条前缀字节,
// 使 DeepSeek 等自动前缀缓存 provider 全部缓存失效(命中率掉到个位数)。
// 尾部追加紧邻本轮内容 LLM 仍可见,前缀稳定命中率最大化。
conv.messages.push_system_anchor(ChatMessage::system(&anchor_text));
true
}
}
};
if inserted {
tracing::info!(
conv_id = %conv_id,
pre_tokens,
"[ai] {}兜底(后台):关键词摘要已尾部追加(compressed 标记已扣 token)",
log_head
);
}
}
#[cfg(test)]
mod tests {
use super::*;
/// 建 conv 并灌 n 条 active 消息,置 is_compressing(true)(模拟触发时已置位)。
/// 返回触发时的内容版本哨兵(首条消息 id),供 keyword_fallback_commit 的版本守卫比对。
async fn seed_compressing(arc: &Arc<Mutex<AiSession>>, conv_id: &str, n: usize) -> Option<String> {
let mut session = arc.lock().await;
let conv = session.conv(conv_id);
for i in 0..n {
conv.messages.push(if i % 2 == 0 {
ChatMessage::user(format!("用户消息{i}"))
} else {
ChatMessage::assistant(format!("助手回复{i}"))
});
}
conv.messages.set_compressing(true);
head_id_of(&mut conv.messages)
}
/// 非空关键词:复位 is_compressing 与兜底提交(标 compressed + 尾部插锚点)一并生效。
#[tokio::test]
async fn keyword_fallback_commit_resets_and_inserts() {
let arc = Arc::new(Mutex::new(AiSession::new()));
let head = seed_compressing(&arc, "c1", 4).await;
keyword_fallback_commit(&arc, "c1", 4, "关键词锚点", 0, "压缩失败", &head, None).await;
let session = arc.lock().await;
let conv = session.conv_read("c1").expect("conv 应存在");
assert!(!conv.messages.is_compressing(), "提交后 is_compressing 应复位 false");
let msgs = conv.messages.all_messages_clone();
assert_eq!(msgs.len(), 5, "压缩后应尾部追加 1 条关键词 system 锚点");
for (i, m) in msgs[..4].iter().enumerate() {
assert!(!m.is_active(), "第 {i} 条原消息应被标 compressed(释放 token)");
}
assert_eq!(msgs[4].content, "关键词锚点", "末条应为关键词锚点正文");
assert!(
matches!(msgs[4].role, df_ai::provider::MessageRole::System),
"锚点应为 system 角色"
);
}
/// 空关键词:复位不能丢(否则 is_compressing 永真 → 压缩永久停摆),但不插锚点。
#[tokio::test]
async fn keyword_fallback_commit_empty_keeps_reset() {
let arc = Arc::new(Mutex::new(AiSession::new()));
let head = seed_compressing(&arc, "c2", 4).await;
keyword_fallback_commit(&arc, "c2", 4, "", 0, "压缩 provider 重建失败", &head, None).await;
let session = arc.lock().await;
let conv = session.conv_read("c2").expect("conv 应存在");
assert!(!conv.messages.is_compressing(), "关键词为空仍应复位 is_compressing");
assert_eq!(conv.messages.len(), 4, "关键词为空不应插锚点,len 不变");
for m in conv.messages.all_messages_clone() {
assert!(!m.is_active(), "空关键词路径仍应标 compressed 扣 token");
}
assert_eq!(conv.messages.history_tokens(), 0, "无锚点插入,token 应清零");
}
/// conv 已删(per_conv 无条目):早返不惰性重建(防污染新 conv)。
#[tokio::test]
async fn keyword_fallback_commit_missing_conv_noop() {
let arc = Arc::new(Mutex::new(AiSession::new()));
keyword_fallback_commit(&arc, "gone", 4, "锚点", 0, "压缩失败", &None, None).await;
let session = arc.lock().await;
assert!(
session.per_conv.is_empty(),
"conv 已删不应惰性重建 per_conv 条目"
);
}
}
// 939⑪⑫ 后台压缩守卫契约:纯函数级断言(版本不符→丢弃 / panic→标志复位)。
#[cfg(test)]
mod guard_tests {
use super::{head_id_of, reset_compressing_after_panic, version_matches, AiSession};
use df_ai::context::{ContextConfig, ContextManager};
use std::sync::Arc;
use tokio::sync::Mutex;
// 939⑪:内容版本不符(首条 id 变化 / 列表被清空 / 空→非空)即判定丢弃。
#[test]
fn version_guard_discards_on_head_change() {
let a = Some("msg_a".to_string());
let b = Some("msg_b".to_string());
assert!(version_matches(&a, &a), "同版本必须放行");
assert!(!version_matches(&a, &b), "首条 id 变(clear 重建)必须丢弃");
assert!(!version_matches(&a, &None), "列表被清空必须丢弃");
assert!(!version_matches(&None, &a), "空→非空(新会话)必须丢弃");
}
// 939⑪:空列表的内容版本哨兵为 None。
#[test]
fn head_id_of_empty_list_is_none() {
let mut mgr = ContextManager::new(ContextConfig::default());
assert_eq!(head_id_of(&mut mgr), None);
}
// 939⑫:后台 body panic → is_compressing 被复位,压缩不停摆。
#[tokio::test]
async fn panic_resets_compressing_flag() {
let session = Arc::new(Mutex::new(AiSession::new()));
let conv_id = "conv-panic";
session.lock().await.conv(conv_id).messages.set_compressing(true);
// 模拟后台 body panic,经 catch_unwind 薄包装转受控 Err
let r = super::super::unwind_to_result(async { panic!("boom in compress task") }).await;
assert!(r.is_err(), "panic 应被捕获为 Err");
// panic 分支:复位 is_compressing
reset_compressing_after_panic(&session, conv_id).await;
assert!(
!session.lock().await.conv_read(conv_id).unwrap().messages.is_compressing(),
"panic 后 is_compressing 必须复位,否则压缩永久停摆"
);
}
}