641 lines
31 KiB
Rust
641 lines
31 KiB
Rust
//! 上下文管理器 — 管理对话上下文和 token 预算
|
||
//!
|
||
//! 职责：
|
||
//! - 维护消息历史及其 token 计数缓存
|
||
//! - 提供预算感知的消息裁剪（保护工具调用三元组）
|
||
//! - 为 run_agentic_loop 提供受控的消息视图
|
||
//!
|
||
//! 裁剪策略与模型选择是正交维度：本模块只管「窗口多大、怎么裁」，
|
||
//! 用哪个 model / 是否启用 reasoning 由调用方在 CompletionRequest 层决定。
|
||
//!
|
||
//! 纯函数 / 数据类型 / 常量（TokenEstimator / ContextConfig / MessageGroup /
|
||
//! TrackedMessage / EvictionUnit / classify_group / PROTECT_COUNT /
|
||
//! TOOL_MISSING_PREFIX）已抽至 [`crate::context_helpers`]，本模块 `use` 复用，
|
||
//! 并 `pub use` 重导出以保持 `df_ai::context::*` 历史路径对外可见（零调用方变更）。
|
||
//!
|
||
//! # 子模块
|
||
//! - [`sanitize`]：畸形配对自愈（`sanitize_messages` / `drop_reverse_orphans` /
|
||
//! `assert_placeholder_pairing` / `ensure_sequence_legal`）及其单测。
|
||
//! 实现为模块级 `pub fn`，本模块通过 `ContextManager` 上的关联函数薄包装转发，
|
||
//! 保持 `ContextManager::sanitize_messages(...)` 调用路径不变（零调用方变更）。
|
||
|
||
mod sanitize;
|
||
|
||
use crate::context_helpers::{
|
||
classify_group, PLACEHOLDER_INTEGRITY_ENABLED, PROTECT_COUNT,
|
||
};
|
||
// 重导出：保持 `df_ai::context::TokenEstimator` / `df_ai::context::ContextConfig` 等
|
||
// 历史路径对外可见（agentic.rs / commands/ai/mod.rs 等调用方零变更）。
|
||
// `pub use` 同时把类型带入本模块命名空间，供 ContextManager 结构体字段与 impl 直接引用。
|
||
pub use crate::context_helpers::{
|
||
EvictionUnit, ContextConfig, MessageGroup, TokenEstimator, TrackedMessage,
|
||
};
|
||
|
||
use crate::provider::{ChatMessage, MessageRole, MessageStatus};
|
||
|
||
// ============================================================
|
||
// 上下文管理器
|
||
// ============================================================
|
||
|
||
/// 上下文管理器
|
||
///
|
||
/// 唯一的消息真相来源（替代原来的 `Vec<ChatMessage>`）。
|
||
/// 裁剪仅影响发送视图（`build_for_request`），不影响持久化（`all_messages_clone`）。
|
||
pub struct ContextManager {
|
||
messages: Vec<TrackedMessage>,
|
||
/// 当前历史总 token 数（不含 system prompt）
|
||
history_tokens: u32,
|
||
config: ContextConfig,
|
||
estimator: TokenEstimator,
|
||
/// 压缩重入标志（F-15 §4.3）：true 表示一次 LLM 压缩正在进行中。
|
||
/// agentic loop 顶部检测，防同一轮内多次触发压缩互相覆盖。纯内存态，不落库。
|
||
is_compressing: bool,
|
||
/// [P2 改进5] 主题切换检测标记。push user 消息时若发现末两条 user 消息的 topic
|
||
/// 都非 None 且不同(双高置信),置位本字段,格式 "old|new"。agentic loop 顶部读并
|
||
/// 消费(insert 系统标记后清空)。纯内存态,不落库。保守:任一 topic 为 None 不置位(宁可漏报)。
|
||
pending_topic_marker: Option<String>,
|
||
/// msg-split-phase1:已落库(ai_messages 表)的消息条数。
|
||
///
|
||
/// 增量写路径用:save_conversation 据此判定 `[persisted_msg_count..len)` 是"自上次 save
|
||
/// 后新增的消息"(纯 append),走 `insert_batch`(INSERT OR IGNORE)只插新行,跳过昂贵的
|
||
/// `replace_conversation`(DELETE 整对话 + 全量重插)。
|
||
///
|
||
/// 仅 append-only 路径才有效——一旦 [`needs_full_rewrite`] 被置位(DB 与内存可能脱钩:
|
||
/// compress 改 status / replace_tool_result_content 改 content / pop_last_assistant_round
|
||
/// 缩短 / insert_at 中段插入改 seq / clear 清空),save 必须走全量重写收敛,本字段随之重置。
|
||
///
|
||
/// restore_from_messages(DB 加载)后设为 len(DB 已是该真相源,append 基线对齐到当前长度)。
|
||
/// 纯内存态,不落库(下次进程重启从 DB reload 时由 restore_from_messages 重置)。
|
||
persisted_msg_count: usize,
|
||
/// msg-split-phase1:全量重写需求标志。
|
||
///
|
||
/// `true` = 自上次 save 后发生过修改既有消息 / 缩短 / 中段插入 / 清空,内存与 DB 可能脱钩,
|
||
/// 下次 save 必须走 `replace_conversation` 全量重写收敛(DELETE + INSERT 全部行)。
|
||
/// save 全量重写成功后清零;append-only save(insert_batch)成功后保持 false。
|
||
///
|
||
/// 任何修改既有消息(非纯 push append)的入口置位:clear / pop_last_assistant_round /
|
||
/// truncate_after_user_message / replace_tool_result_content / replace_last_active_user_content /
|
||
/// insert_at / compress_old_messages / messages_mut(可变借用兜底,调用方可能改 status/content)。
|
||
/// 误置位零代价(只是下次 save 多走一次全量重写,正确性不变);漏置位才会丢更新——故保守置位。
|
||
needs_full_rewrite: bool,
|
||
}
|
||
|
||
impl ContextManager {
|
||
pub fn new(config: ContextConfig) -> Self {
|
||
Self {
|
||
messages: Vec::new(),
|
||
history_tokens: 0,
|
||
config,
|
||
estimator: TokenEstimator::default(),
|
||
is_compressing: false,
|
||
pending_topic_marker: None,
|
||
persisted_msg_count: 0,
|
||
needs_full_rewrite: false,
|
||
}
|
||
}
|
||
|
||
/// 追加消息（自动计算 token 并更新缓存）
|
||
///
|
||
/// 不在此处淘汰——push 可能发生在 agentic loop 中间（追加 tool_result），
|
||
/// 此时不应裁剪正在使用的活跃消息。裁剪在 `build_for_request` 时统一处理。
|
||
pub fn push(&mut self, message: ChatMessage) {
|
||
let tokens = self.estimator.estimate_message(&message);
|
||
let group = classify_group(&message);
|
||
// 仅 active 消息计入 token 预算（F-15 §3.3）：truncated / archived_segment /
|
||
// compressed 不进 LLM 上下文，token 虚高会致 build_for_request 误判超预算
|
||
// 触发不必要裁剪。!active 消息仍 push 到 self.messages 全量保留（持久化不受影响），
|
||
// sanitize_messages step0(is_active 过滤)在发送视图统一剔除。
|
||
if message.is_active() {
|
||
self.history_tokens += tokens;
|
||
}
|
||
// [P2 改进5] 主题推断(仅 user 消息):IntentRecognizer 识别意图,置信 >= 0.7 且
|
||
// 非 Unknown 则把 Intent 标签存入 TrackedMessage.topic 供主题切换检测。
|
||
// 保守:低置信(None)不标,避免误报。topic 不参与裁剪/压缩(只供 marker 检测)。
|
||
let topic: Option<String> = if matches!(message.role, MessageRole::User) {
|
||
let (intent, conf) = crate::intent::IntentRecognizer::recognize(&message.content);
|
||
if conf >= 0.7 && !matches!(intent, crate::intent::Intent::Unknown) {
|
||
Some(intent.as_str().to_string())
|
||
} else {
|
||
None
|
||
}
|
||
} else {
|
||
None
|
||
};
|
||
// [P2 改进5] 主题切换检测:push 前(本消息即将成末条 user),若已存在一条更早 user
|
||
// 且本消息 topic 与之都非 None 且不同 → 置位 pending_topic_marker("old|new")。
|
||
// 双高置信(两条 topic 都非 None)才标,任一 None 不标(宁可漏报不误报)。
|
||
if matches!(message.role, MessageRole::User) {
|
||
if let Some(prev_topic) = self.last_user_topic() {
|
||
if let Some(this_topic) = &topic {
|
||
if prev_topic != *this_topic {
|
||
self.pending_topic_marker =
|
||
Some(format!("{}|{}", prev_topic, this_topic));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
self.messages.push(TrackedMessage {
|
||
message,
|
||
token_count: tokens,
|
||
group,
|
||
topic,
|
||
});
|
||
}
|
||
|
||
/// 清空所有消息
|
||
pub fn clear(&mut self) {
|
||
self.messages.clear();
|
||
self.history_tokens = 0;
|
||
self.is_compressing = false;
|
||
self.pending_topic_marker = None;
|
||
// msg-split-phase1:全清后 DB 必须也清(调用方 chat.rs:1149 走 delete_range),
|
||
// 全量重写基线重置到 0。needs_full_rewrite=true 保险(若 save 先于 delete_range 触发,
|
||
// 全量重写空列表也会清 DB)。
|
||
self.persisted_msg_count = 0;
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
|
||
/// 消息数量
|
||
pub fn len(&self) -> usize {
|
||
self.messages.len()
|
||
}
|
||
|
||
pub fn is_empty(&self) -> bool {
|
||
self.messages.is_empty()
|
||
}
|
||
|
||
/// 当前历史占用的 token 数（不含 system prompt）
|
||
pub fn history_tokens(&self) -> u32 {
|
||
self.history_tokens
|
||
}
|
||
|
||
/// 预算上限
|
||
pub fn budget_limit(&self) -> u32 {
|
||
self.config.budget_limit()
|
||
}
|
||
|
||
// ── 核心：构建请求消息（受控裁剪版本）──
|
||
|
||
/// 构建发送给 LLM 的消息列表
|
||
///
|
||
/// `sys_tokens` 为调用方已估算好的 system prompt token 数。
|
||
/// 超预算时自动裁剪旧消息（保护工具调用三元组 + 最近 PROTECT_COUNT 条）。
|
||
/// 返回 (消息列表, 是否发生了裁剪)。
|
||
pub fn build_for_request(&self, sys_tokens: u32) -> (Vec<ChatMessage>, bool) {
|
||
let budget = self.budget_limit();
|
||
let available = budget.saturating_sub(sys_tokens);
|
||
|
||
// system prompt 自身超预算：裁剪无法缓解(仍返回保护区兜底),warn 便于诊断
|
||
if sys_tokens > budget {
|
||
tracing::warn!(
|
||
"system prompt (~{} tokens) 超过上下文预算 ({}),裁剪无法缓解",
|
||
sys_tokens, budget
|
||
);
|
||
}
|
||
|
||
// 未超预算 → 直接返回全量（仍做畸形配对自愈，防历史中毒触发 provider 500 死循环）
|
||
if self.history_tokens <= available {
|
||
let sanitized = Self::sanitize_messages(self.all_messages_clone());
|
||
// 出口断言:占位配对完整性,失败降级 TOOL_MISSING_PREFIX 自愈(防 400 orphan)
|
||
return (
|
||
Self::assert_placeholder_pairing(sanitized, PLACEHOLDER_INTEGRITY_ENABLED),
|
||
false,
|
||
);
|
||
}
|
||
|
||
// 超预算 → 视图裁剪（不修改 self.messages，保证 all_messages_clone 仍返回全量）
|
||
let protect_start = self.messages.len().saturating_sub(PROTECT_COUNT);
|
||
let units = self.build_eviction_units(protect_start);
|
||
|
||
let mut removed: u64 = 0;
|
||
let mut trim_end = 0;
|
||
for unit in &units {
|
||
if self.history_tokens.saturating_sub(removed as u32) <= available {
|
||
break;
|
||
}
|
||
removed += unit.token_sum as u64;
|
||
trim_end = unit.end;
|
||
}
|
||
|
||
if trim_end == 0 {
|
||
tracing::warn!(
|
||
"history (~{} tokens) 超预算 ({}) 但无可淘汰单元(全在保护区 {} 条),发送兜底可能触发 provider 超限",
|
||
self.history_tokens, available, PROTECT_COUNT
|
||
);
|
||
// 兜底全量也过 sanitize(对齐分支 1/3),防绕过序列修复直送 provider
|
||
// 触发"首条 assistant 非法"/orphan/连续 role。原裸返 all_messages_clone 不过滤
|
||
// truncated/中毒三元组/首条非法——是主 loop 唯一的 sanitize 漏洞(大体量 tool_result
|
||
// 致超预算且保护区满时命中)。异常会话(开头连续 assistant/tool 无 user)经
|
||
// ensure_sequence_legal 清空后,由协议层 ensure_leading_user 补 user 占位降级,不阻塞。
|
||
// view-only:不改 self.messages 持久化(与分支 1/3 一致)。
|
||
let sanitized = Self::sanitize_messages(self.all_messages_clone());
|
||
return (
|
||
Self::assert_placeholder_pairing(sanitized, PLACEHOLDER_INTEGRITY_ENABLED),
|
||
false,
|
||
);
|
||
}
|
||
|
||
let msgs: Vec<ChatMessage> = self.messages[trim_end..]
|
||
.iter()
|
||
.map(|t| t.message.clone())
|
||
.collect();
|
||
|
||
tracing::info!(
|
||
"context_trimmed: skip {} messages, ~{} tokens (view-only, full history retained)",
|
||
trim_end, removed
|
||
);
|
||
let sanitized = Self::sanitize_messages(msgs);
|
||
// 出口断言:占位配对完整性,失败降级 TOOL_MISSING_PREFIX 自愈(防 400 orphan)
|
||
(
|
||
Self::assert_placeholder_pairing(sanitized, PLACEHOLDER_INTEGRITY_ENABLED),
|
||
true,
|
||
)
|
||
}
|
||
|
||
// ── 畸形配对自愈（转发至 [`sanitize`] 子模块，保持 ContextManager::xxx 调用路径）──
|
||
|
||
/// 畸形配对自愈 — 转发到 [`sanitize::sanitize_messages`]。
|
||
///
|
||
/// 保留为 `ContextManager` 关联函数以兼容历史调用路径（`Self::sanitize_messages` /
|
||
/// `ContextManager::sanitize_messages`），实现见子模块文档。
|
||
pub fn sanitize_messages(messages: Vec<ChatMessage>) -> Vec<ChatMessage> {
|
||
sanitize::sanitize_messages(messages)
|
||
}
|
||
|
||
/// 发送视图出口断言 — 转发到 [`sanitize::assert_placeholder_pairing`]。
|
||
pub fn assert_placeholder_pairing(
|
||
messages: Vec<ChatMessage>,
|
||
enabled: bool,
|
||
) -> Vec<ChatMessage> {
|
||
sanitize::assert_placeholder_pairing(messages, enabled)
|
||
}
|
||
|
||
/// 全量克隆（持久化 save_conversation / build_for_request 未裁剪分支，不受裁剪影响）
|
||
pub fn all_messages_clone(&self) -> Vec<ChatMessage> {
|
||
self.messages.iter().map(|t| t.message.clone()).collect()
|
||
}
|
||
|
||
/// 只读最近 N 条消息(尾部切片 clone)。BE2(AC-EFF-R2-1):断路器/探索熔断检查等只关心
|
||
/// 最近消息的读路径,用本方法避免 all_messages_clone 每轮全量 clone(长对话每轮 O(n) 深克隆浪费)。
|
||
/// 尾部顺序保持时间正序(与 all_messages_clone 一致,调用方从尾部反向扫即可)。
|
||
pub fn recent_messages(&self, n: usize) -> Vec<ChatMessage> {
|
||
let skip = self.messages.len().saturating_sub(n);
|
||
self.messages[skip..]
|
||
.iter()
|
||
.map(|t| t.message.clone())
|
||
.collect()
|
||
}
|
||
|
||
/// 从 Vec 恢复（兼容从 DB 加载）
|
||
pub fn restore_from_messages(&mut self, messages: Vec<ChatMessage>) {
|
||
// clear() 会置 needs_full_rewrite=true + persisted_msg_count=0;但本入口是"DB 刚加载",
|
||
// 加载后的 messages 与 DB 完全一致(它们就是从 DB 来的),append 基线应对齐到当前 len,
|
||
// 且不需要全量重写(DB 已是真相源)。故 clear 后立即覆盖这两个字段。
|
||
let len = messages.len();
|
||
self.clear();
|
||
for msg in messages {
|
||
self.push(msg);
|
||
}
|
||
self.persisted_msg_count = len;
|
||
self.needs_full_rewrite = false;
|
||
}
|
||
|
||
/// 就地替换某条 tool_result 的内容（兼容审批 replace_tool_result）
|
||
/// 返回 true 如果找到并替换了
|
||
///
|
||
/// 反向遍历：tool_result 由 append 进入历史，被替换的通常是最近的审批占位，
|
||
/// 从尾部查找命中即停，避免对长历史做正向 O(n) 累积扫描。
|
||
pub fn replace_tool_result_content(&mut self, tool_call_id: &str, new_content: &str) -> bool {
|
||
let pos = self.messages.iter().rposition(|t| {
|
||
matches!(t.message.role, MessageRole::Tool)
|
||
&& t.message.tool_call_id.as_deref() == Some(tool_call_id)
|
||
});
|
||
|
||
let Some(i) = pos else { return false };
|
||
|
||
// 先更新 content，再重估 token 并校正总量
|
||
let old_tokens = self.messages[i].token_count;
|
||
self.messages[i].message.content = new_content.to_string();
|
||
let new_tokens = self.estimator.estimate_message(&self.messages[i].message);
|
||
self.messages[i].token_count = new_tokens;
|
||
self.history_tokens = self.history_tokens.saturating_sub(old_tokens).saturating_add(new_tokens);
|
||
// msg-split-phase1:改既有消息 content,DB 需全量重写收敛(append-only 路径不覆盖旧行)。
|
||
self.needs_full_rewrite = true;
|
||
true
|
||
}
|
||
|
||
/// 弹出末尾连续的 assistant 消息（含其 tool_calls 三元组尾随 tool_result）
|
||
///
|
||
/// 用于「重新生成」(UX-02):删掉最后一条 AI 回复(可能跨多轮 tool_calls + tool_results
|
||
/// 紧随其后),保留触发它的 user 消息,以便重跑 agentic loop 再生成。
|
||
///
|
||
/// 语义:从末尾向前弹出,直到弹出至少一条 assistant 消息;若弹出 assistant 后紧邻的更早
|
||
/// 消息仍是 assistant/tool(同一轮多块),继续一并弹出,确保不留半截三元组污染下轮。
|
||
/// user 消息作为停止边界(不弹出 user),保证重生成时历史末尾是 user 消息。
|
||
pub fn pop_last_assistant_round(&mut self) -> bool {
|
||
if self.messages.is_empty() {
|
||
return false;
|
||
}
|
||
let mut popped_any = false;
|
||
// 从尾向前:先弹掉末尾非 user 的消息(assistant / tool),直到遇到 user 或空
|
||
while let Some(last) = self.messages.last() {
|
||
if matches!(last.message.role, MessageRole::User) {
|
||
break;
|
||
}
|
||
let removed = self.messages.pop().expect("just checked non-empty");
|
||
self.history_tokens = self.history_tokens.saturating_sub(removed.token_count);
|
||
if matches!(removed.message.role, MessageRole::Assistant) {
|
||
popped_any = true;
|
||
}
|
||
}
|
||
// msg-split-phase1:从末尾弹出 → len 缩短,append-only 基线(persisted_msg_count)
|
||
// 会大于新 len。下次 save 必须全量重写(删 DB 中已不存在的尾行)。
|
||
if popped_any {
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
popped_any
|
||
}
|
||
|
||
/// 编辑某条 user 消息后，将其后所有消息标记为 truncated（UX-09 编辑重生成）。
|
||
///
|
||
/// 软删语义：保留在内存真相源 + DB（可追溯），但 sanitize_messages 过滤后不进 LLM 上下文，
|
||
/// 前端按 is_active 过滤从视图移除。返回被标 truncated 的条数（0 表示该 user 消息已是末尾，无后续）。
|
||
///
|
||
/// `target_content` 为该 user 消息的预期内容（用于反向唯一定位：末条 user 消息可能内容相同，
|
||
/// 故从尾部向前找第一条 role=User 且 content 匹配且仍 active 的消息）。
|
||
/// 找不到返回 Err(())，调用方据此报错。
|
||
pub fn truncate_after_user_message(&mut self, target_content: &str) -> Result<usize, ()> {
|
||
// 反向找末条 active user 消息且 content 匹配
|
||
let pos = self.messages.iter().rposition(|t| {
|
||
matches!(t.message.role, MessageRole::User)
|
||
&& t.message.content == target_content
|
||
&& t.message.is_active()
|
||
});
|
||
let Some(i) = pos else { return Err(()) };
|
||
// i 之后的全部标 truncated(已 truncated 的跳过,只统计本次新标的)
|
||
let mut count = 0usize;
|
||
for t in self.messages[i + 1..].iter_mut() {
|
||
if t.message.is_active() {
|
||
t.message.status = Some(MessageStatus::Truncated);
|
||
count += 1;
|
||
}
|
||
}
|
||
// msg-split-phase1:改既有消息 status(truncated),DB 需全量重写收敛。
|
||
if count > 0 {
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
Ok(count)
|
||
}
|
||
|
||
/// 替换末条 active user 消息的 content（UX-09 编辑重生成）。
|
||
///
|
||
/// 编辑语义：只能编辑最后一条 user 消息（中间编辑语义复杂，拒绝）。返回 Err(()) 表示无 active user 消息。
|
||
/// 成功后调用方应紧接着 truncate_after_user_message(new_content) 软删其后续消息。
|
||
pub fn replace_last_active_user_content(&mut self, new_content: &str) -> Result<(), ()> {
|
||
let pos = self.messages.iter().rposition(|t| {
|
||
matches!(t.message.role, MessageRole::User) && t.message.is_active()
|
||
});
|
||
let Some(i) = pos else { return Err(()) };
|
||
let old_tokens = self.messages[i].token_count;
|
||
self.messages[i].message.content = new_content.to_string();
|
||
let new_tokens = self.estimator.estimate_message(&self.messages[i].message);
|
||
self.messages[i].token_count = new_tokens;
|
||
self.history_tokens = self.history_tokens.saturating_sub(old_tokens).saturating_add(new_tokens);
|
||
// msg-split-phase1:改既有 user 消息 content,DB 需全量重写收敛。
|
||
self.needs_full_rewrite = true;
|
||
Ok(())
|
||
}
|
||
|
||
/// 只读迭代（兼容 ensure_conversation_title 的 .iter().filter() 等）
|
||
pub fn iter(&self) -> impl Iterator<Item = &ChatMessage> {
|
||
self.messages.iter().map(|t| &t.message)
|
||
}
|
||
|
||
/// 消息级溯源:取末条指定 role 消息的 id(ULID)。
|
||
///
|
||
/// 从尾部反向扫描(末条消息命中即停,避免全量 O(n) 正扫累积),返回最近一条
|
||
/// `role` 匹配且 `id` 非空消息的 id。无匹配或老消息无 id → None(向前兼容:
|
||
/// 老反序列化消息 id=None,溯源写入降级为 None,展示侧兼容 `conv:` 旧格式)。
|
||
///
|
||
/// 用途:
|
||
/// - `MessageRole::Assistant`:audit/知识提炼写入时取当前 assistant 消息 id
|
||
/// (LLM 返回带 tool_calls 的 assistant 已 push,process_tool_calls 入口取)
|
||
/// - `MessageRole::User`:知识注入 referenced 事件溯源取触发检索的 user 消息 id
|
||
fn last_message_id_by_role(&self, role: MessageRole) -> Option<String> {
|
||
self.messages
|
||
.iter()
|
||
.rev()
|
||
.find(|t| {
|
||
std::mem::discriminant(&t.message.role) == std::mem::discriminant(&role)
|
||
})
|
||
.and_then(|t| t.message.id.clone())
|
||
}
|
||
|
||
/// 末条 assistant 消息的 id(消息级溯源用)。
|
||
pub fn last_assistant_message_id(&self) -> Option<String> {
|
||
self.last_message_id_by_role(MessageRole::Assistant)
|
||
}
|
||
|
||
/// 末条 user 消息的 id(消息级溯源用)。
|
||
pub fn last_user_message_id(&self) -> Option<String> {
|
||
self.last_message_id_by_role(MessageRole::User)
|
||
}
|
||
|
||
// ── [P2 改进5] 主题切换检测(保守,双高置信才标) ──
|
||
|
||
/// 取末条 user 消息的 topic 标签(供 push 时主题切换检测)。
|
||
///
|
||
/// 从尾部反向扫描 user 消息,取最近一条 role=User 的 TrackedMessage.topic。
|
||
/// 老消息(未接改进5 推断)topic=None → 返 None(向前兼容)。无 user 消息 → None。
|
||
fn last_user_topic(&self) -> Option<String> {
|
||
self.messages
|
||
.iter()
|
||
.rev()
|
||
.find(|t| matches!(t.message.role, MessageRole::User))
|
||
.and_then(|t| t.topic.clone())
|
||
}
|
||
|
||
/// 取并消费 pending_topic_marker(供 agentic loop 顶部读 → insert 系统标记 → 清空)。
|
||
///
|
||
/// 返回 "old|new" 格式字符串(push 时末两条 user topic 都非 None 且不同置位)。
|
||
/// 取出即清空(一次性消费,防同 marker 重复 insert)。无 marker → None。
|
||
pub fn take_topic_marker(&mut self) -> Option<String> {
|
||
self.pending_topic_marker.take()
|
||
}
|
||
|
||
// ── F-15 上下文管理增强辅助方法（压缩链路核心 API）──
|
||
//
|
||
// 已全量接入压缩链路:
|
||
// - agentic/mod.rs 自动压缩(trigger 判定 + LLM/关键词兜底 + 重入保护)
|
||
// set_compressing/is_compressing(重入标志) / has_compressible_messages(触发判定)
|
||
// messages_mut(取 active 喂 LLM) / compress_old_messages(标 compressed 扣 token)
|
||
// insert_at(摘要/续接锚点 system 消息插入首位)
|
||
// - commands/ai/compress.rs 与 commands/ai/commands/chat.rs 走 IPC 压缩入口
|
||
// - build_eviction_units(下方)供会话分段与压缩定位共用同一分组逻辑。
|
||
|
||
/// 配置（只读视图，供 agentic.rs 计算压缩触发阈值 `config().budget_limit()`）
|
||
pub fn config(&self) -> &ContextConfig {
|
||
&self.config
|
||
}
|
||
|
||
/// 可变消息切片（供标记 status="compressed"/"archived_segment" + 调整 token）
|
||
///
|
||
/// 调用方约定:仅改 `message.status` / `message.content`,不增删条目(增删走
|
||
/// [`push`] / [`insert_at`]),否则 `history_tokens` 会与实际脱钩。
|
||
///
|
||
/// msg-split-phase1:**若调用方借此句柄改了既有消息(status/content),必须紧接着调
|
||
/// [`mark_needs_full_rewrite`]**,否则下次 save 走 append-only 增量路径会漏更新旧行,
|
||
/// DB 与内存脱钩。仅读(过滤后 clone)不需调。本方法无法自行置位(借用冲突:返回 &mut
|
||
/// 切片时不能再持有 &mut self 标志)。
|
||
pub fn messages_mut(&mut self) -> &mut [TrackedMessage] {
|
||
&mut self.messages
|
||
}
|
||
|
||
/// msg-split-phase1:显式标记"下次 save 须全量重写"。
|
||
///
|
||
/// 供 [`messages_mut`] 的调用方在改完既有消息后调用(本 struct 无法在返回 &mut 切片时
|
||
/// 自行置位)。也可供任何绕过本 impl 直接改 messages 的路径兜底。幂等(重复置 true 无害)。
|
||
pub fn mark_needs_full_rewrite(&mut self) {
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
|
||
/// msg-split-phase1:读取已落库消息条数(append-only 增量写路径的基线)。供 save_conversation
|
||
/// 判定 `[persisted_msg_count..len)` 是否为"自上次 save 后新增"。
|
||
pub fn persisted_msg_count(&self) -> usize {
|
||
self.persisted_msg_count
|
||
}
|
||
|
||
/// msg-split-phase1:推进已落库消息条数(insert_batch/replace_conversation 成功后调用)。
|
||
/// 同时清 needs_full_rewrite(DB 已与内存一致)。供 save_conversation 在写库成功后调用。
|
||
pub fn advance_persisted_count(&mut self, new_count: usize) {
|
||
// 防回退:若调用方误传更小值(理论上不应发生),取 max 保不丢基线(下次 save 仍能收敛)。
|
||
self.persisted_msg_count = self.persisted_msg_count.max(new_count);
|
||
self.needs_full_rewrite = false;
|
||
}
|
||
|
||
/// msg-split-phase1:是否需要全量重写(改过既有消息 / 缩短 / 中段插入 / 清空)。
|
||
pub fn needs_full_rewrite(&self) -> bool {
|
||
self.needs_full_rewrite
|
||
}
|
||
|
||
/// 在给定位置插入一条消息(其余向后移),并把它计入 token 预算(active 才计)。
|
||
///
|
||
/// 供压缩点插入摘要 system 消息。`index` 越界则 panic(对齐 Vec::insert 语义,
|
||
/// 调用方负责算合法 index,如 `compress_end` 已由 `compress_old_messages` 校验)。
|
||
pub fn insert_at(&mut self, index: usize, message: ChatMessage) {
|
||
let tokens = self.estimator.estimate_message(&message);
|
||
let group = classify_group(&message);
|
||
if message.is_active() {
|
||
self.history_tokens += tokens;
|
||
}
|
||
self.messages.insert(index, TrackedMessage {
|
||
message,
|
||
token_count: tokens,
|
||
group,
|
||
topic: None,
|
||
});
|
||
// msg-split-phase1:中段插入会改变 index 及之后所有消息的 seq(append-only 基线
|
||
// persisted_msg_count 按"末尾追加"语义计算 seq,中段插入后 seq 错位),DB 必须全量重写。
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
|
||
/// 按淘汰单元分组消息范围(三元组原子性),供压缩定位/分段标记复用同一分组逻辑。
|
||
///
|
||
/// 返回每个单元的右开区间 end + token 总和,保证:
|
||
/// - 工具调用三元组(Head + Tail* + 紧随的 Standalone Assistant)在同一单元
|
||
/// - 保护区 `[protect_start, len)` 内的消息不纳入任何单元
|
||
///
|
||
/// 公开会话分段(`archived_segment` 按组原子标记)与压缩定位共用。
|
||
pub fn build_eviction_units(&self, protect_start: usize) -> Vec<EvictionUnit> {
|
||
let mut units = Vec::new();
|
||
let mut i = 0usize;
|
||
|
||
while i < protect_start {
|
||
let mut token_sum = 0u32;
|
||
|
||
if self.messages[i].group == MessageGroup::ToolCallHead {
|
||
token_sum += self.messages[i].token_count;
|
||
i += 1;
|
||
while i < protect_start && self.messages[i].group == MessageGroup::ToolResultTail {
|
||
token_sum += self.messages[i].token_count;
|
||
i += 1;
|
||
}
|
||
if i < protect_start
|
||
&& self.messages[i].group == MessageGroup::Standalone
|
||
&& matches!(self.messages[i].message.role, MessageRole::Assistant)
|
||
{
|
||
token_sum += self.messages[i].token_count;
|
||
i += 1;
|
||
}
|
||
} else {
|
||
token_sum += self.messages[i].token_count;
|
||
i += 1;
|
||
}
|
||
|
||
units.push(EvictionUnit { end: i, token_sum });
|
||
}
|
||
|
||
units
|
||
}
|
||
|
||
/// 保护区外是否存在可压缩消息(供 agentic loop 顶部触发判断)。
|
||
///
|
||
/// "可压缩"= status 为 None/active 的消息(已 compressed/archived_segment/truncated
|
||
/// 不参与二次压缩,幂等)。`protect_start` 为保护区起点(如 `len - PROTECT_COUNT`)。
|
||
pub fn has_compressible_messages(&self, protect_start: usize) -> bool {
|
||
let end = protect_start.min(self.messages.len());
|
||
// 排除 system 角色(压缩摘要 / 话题切换锚点)。这些是上下文锚点非压缩目标——
|
||
// 若计入,压缩摘要 insert_at(0) 落在可压缩区 [0..protect_start) 且 is_active(status=None),
|
||
// 致每轮 has_compressible 恒 true → 无限循环压缩(用户报"压缩后每轮提示已压缩并停止")。
|
||
// compress_old_messages 不改:被调用时仍标旧 system 摘要 compressed(被新摘要替代,防堆积)。
|
||
self.messages[..end]
|
||
.iter()
|
||
.any(|t| t.message.is_active() && !matches!(t.message.role, MessageRole::System))
|
||
}
|
||
|
||
/// 把保护区 `[0, compress_end)` 范围内的 active 消息标记为 `status="compressed"`,
|
||
/// 同步从 `history_tokens` 扣除其 token,返回被压缩消息的克隆(供喂 LLM 摘要)。
|
||
///
|
||
/// **幂等**:已 compressed(或任何 !active)的消息跳过,不会被二次压缩;`history_tokens`
|
||
/// 也只扣首次标记的 token。返回的 Vec 仅含**本次新标记**的消息(已 compressed 的不返)。
|
||
///
|
||
/// **单向不可逆**:压缩后 DB 原始消息保留,但 LLM 上下文里被 is_active 白名单隔离
|
||
/// (sanitize step0 过滤)。`compress_end` 越界自动 clamp 到 `messages.len()`。
|
||
///
|
||
/// 返回空 Vec 表示本批次无可压缩消息(全部已 compressed 或范围空),调用方据此跳过 LLM 调用。
|
||
pub fn compress_old_messages(&mut self, compress_end: usize) -> Vec<ChatMessage> {
|
||
let end = compress_end.min(self.messages.len());
|
||
let mut newly_compressed = Vec::new();
|
||
for t in self.messages[..end].iter_mut() {
|
||
if t.message.is_active() {
|
||
newly_compressed.push(t.message.clone());
|
||
t.message.status = Some(MessageStatus::Compressed);
|
||
self.history_tokens = self.history_tokens.saturating_sub(t.token_count);
|
||
}
|
||
}
|
||
// msg-split-phase1:改了既有消息 status(active → compressed),DB 需全量重写收敛
|
||
// (append-only 增量路径不覆盖旧行)。仅在确实标了新 compressed 时置位(无可压缩时
|
||
// newly_compressed 为空,状态未变,无需全量重写)。
|
||
if !newly_compressed.is_empty() {
|
||
self.needs_full_rewrite = true;
|
||
}
|
||
newly_compressed
|
||
}
|
||
|
||
/// 压缩重入标志(读)。true 表示一次 LLM 压缩正在进行中,触发方应跳过本轮压缩。
|
||
pub fn is_compressing(&self) -> bool {
|
||
self.is_compressing
|
||
}
|
||
|
||
/// 压缩重入标志(写)。`true`=开始压缩(进入 agentic loop 顶部前置置位),
|
||
/// `false`=压缩结束(无论成功或降级)。调用方必须成对调用,防止永久卡死。
|
||
pub fn set_compressing(&mut self, v: bool) {
|
||
self.is_compressing = v;
|
||
}
|
||
|
||
// build_eviction_units / classify_group 等已在上方公开或文件级定义。
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod manager_tests;
|