修复: 压缩边界对齐与断路器学习语义

- 压缩边界按对话单元回退,不再劈开工具对
- 停止与接管退出补跳过结果,畸形不再新增
- 失败指纹纳入参数,改参重试不再累计熔断
- 成功断连失败段,红绿循环不再误熔断
- 脚本非零退出豁免,同命令死磕由签名熔断兜底
- 熔断前加自纠警示轮,临时路径写删豁免
This commit is contained in:
lxy
2026-10-01 19:35:22 +08:00
parent b8416dea63
commit 4417b3ddd3
4 changed files with 416 additions and 47 deletions
+84 -2
View File
@@ -618,11 +618,24 @@ impl ContextManager {
/// 也只扣首次标记的 token。返回的 Vec 仅含**本次新标记**的消息(已 compressed 的不返)。
///
/// **单向不可逆**:压缩后 DB 原始消息保留,但 LLM 上下文里被 is_active 白名单隔离
/// (sanitize step0 过滤)。`compress_end` 越界自动 clamp 到 `messages.len()`。
/// (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());
// 压缩边界按对话单元(三元组)对齐:调用方传入的 end(如 len - PROTECT_COUNT)周期性
// 落在单元中间(assistant 头与其 tool 结果之间),只压半个三元组会让保护区以孤立的
// 头或尾开头,成为 sanitize 每请求告警的确定性供给。按全长分组取「完整单元边界中
// 不超过 raw_end 的最大者」回退(只少压不多压,不侵占保护区);三元组要么整段压缩
// 要么整段保留。空范围/空历史无单元,取 0。
let raw_end = compress_end.min(self.messages.len());
let end = self
.build_eviction_units(self.messages.len())
.iter()
.map(|u| u.end)
.take_while(|&e| e <= raw_end)
.last()
.unwrap_or(0);
let mut newly_compressed = Vec::new();
for t in self.messages[..end].iter_mut() {
if t.message.is_active() {
@@ -656,3 +669,72 @@ impl ContextManager {
#[cfg(test)]
mod manager_tests;
#[cfg(test)]
mod compress_boundary_tests {
use super::*;
use crate::provider::ToolCall;
fn mgr() -> ContextManager {
ContextManager::new(ContextConfig {
max_tokens: 100_000,
..Default::default()
})
}
/// 构造 4 个周期轮次,每轮 [user, assistant(tool_calls), tool_result, assistant]:
/// user 独占一个单元,后三条(头+尾+紧随独立 assistant)合并为一个单元,
/// 单元边界落在 4/8/12/16。
fn periodic_rounds() -> ContextManager {
let mut m = mgr();
for r in 0..4 {
m.push(ChatMessage::user(&format!("问题 {}", r)));
m.push(ChatMessage::assistant_with_tools(
"",
vec![ToolCall::new(format!("c{}", r), "run_command", "{}")],
));
m.push(ChatMessage::tool_result(format!("c{}", r), "ok"));
m.push(ChatMessage::assistant("小结"));
}
m
}
#[test]
fn compress_end_mid_unit_retreats_to_unit_boundary() {
let mut m = periodic_rounds();
// len=16,调用方口径 len - PROTECT_COUNT(6) = 10,落在第 3 轮三元组中间
// (9 头 / 10 尾 / 11 独立)。旧行为会把头压掉却留下孤立的 tool 结果。
// 回退到完整单元边界 9:只多压前面的 user,不劈开三元组(只少压不多压)。
let compressed = m.compress_old_messages(10);
assert_eq!(compressed.len(), 9, "边界回退到完整单元边界 9");
let msgs = m.messages_mut();
for i in 9..16 {
assert!(msgs[i].message.is_active(), "第 {} 条应保持 active", i);
}
// 关键回归点:第 3 轮三元组(9 头 / 10 尾 / 11 独立)整体保持 active,头未被劈开
assert!(msgs[9].message.is_active(), "三元组头不应被单独压缩");
assert!(msgs[10].message.is_active(), "三元组尾不应残留 active 而头被压");
assert!(msgs[11].message.is_active());
assert_eq!(msgs[9].group, MessageGroup::ToolCallHead);
}
#[test]
fn compress_end_on_boundary_unchanged() {
let mut m = periodic_rounds();
// end=8 恰为单元边界,行为与对齐前一致(整 8 条压缩)。
let compressed = m.compress_old_messages(8);
assert_eq!(compressed.len(), 8);
let msgs = m.messages_mut();
assert!(msgs[..8].iter().all(|t| !t.message.is_active()));
assert!(msgs[8..].iter().all(|t| t.message.is_active()));
}
#[test]
fn compress_end_oversized_still_compresses_all() {
let mut m = periodic_rounds();
// 越界 clamp 到 len:全量单元分组后末单元 end=len,语义与改动前一致。
let compressed = m.compress_old_messages(999);
assert_eq!(compressed.len(), 16);
assert_eq!(m.history_tokens(), 0);
}
}
+75 -9
View File
@@ -323,26 +323,47 @@ pub(crate) fn normalize_loop_path(p: &str) -> String {
p.replace('\\', "/").to_lowercase()
}
/// 从工具调用提取回环样本:write_file/delete_file 取 path 参数规范化,其他工具 None 占位。
/// path 缺失/非字符串 → None(无路径无从判同路径,退化为占位)。
/// 从工具调用提取回环样本:write_file/delete_file 记规范化 path;read_file 记规范化 path
/// (该路径的「使用」证据);run_command 记整条命令(判定时按「命令含路径」匹配使用);
/// 其他工具 None 占位。path/command 缺失/非字符串 → None(无从判,退化为占位)。
pub(crate) fn destructive_loop_sample(name: &str, args: &serde_json::Value) -> DestructiveLoopSample {
match name {
"write_file" | "delete_file" => {
let p = arg_str(args, "path");
if p.is_empty() { None } else { Some((name.to_string(), normalize_loop_path(&p))) }
}
_ => None,
}
let target = match name {
"write_file" | "delete_file" | "read_file" => arg_str(args, "path"),
"run_command" => arg_str(args, "command"),
_ => return None,
};
if target.is_empty() { None } else { Some((name.to_string(), normalize_loop_path(&target))) }
}
/// 破坏性回环检测:窗口内「write_file(P) 后 GAP 步内 delete_file(P)」或反向 → 命中路径。
///
/// GAP=6(全工具调用序位):写删交替紧凑才判回环,间隔远的写与删是正常迭代修复不算。
/// 两类豁免(降误伤,治 .trash 临时脚本卫生流程被误熔断):
/// - 路径命中临时/回收模式(.trash、/temp/、/tmp/、/target/、.cache 等):写完即删是
/// 产物卫生,整对豁免;
/// - 写删对之间有对该路径的使用(read 精确命中 / run 命令含路径):用后清理,该对豁免。
/// 真「写完即删用户文件」(非临时路径且中间无使用)仍拦。
/// 返回首个命中对的规范化路径(警示/熔断文本可引用),无命中 → None。
pub(crate) fn detect_destructive_loop(samples: &[DestructiveLoopSample]) -> Option<String> {
/// 回环对最大间隔步数(按全部工具调用序位)。
const DESTRUCTIVE_LOOP_GAP: usize = 6;
// 临时/回收产物路径(入参须为 normalize_loop_path 后的形式:分隔符统一 /、小写)。
let is_temp_recycle = |p: &str| -> bool {
const MARKERS: &[&str] = &[".trash", "/temp/", "/tmp/", "/target/", ".cache"];
MARKERS.iter().any(|m| p.contains(m))
|| p.starts_with("temp/")
|| p.starts_with("tmp/")
};
// 写删对之间对该路径的使用:read 精确命中;run 按命令文本含路径匹配。
let is_use_between = |s: &DestructiveLoopSample, p: &str| -> bool {
match s.as_ref() {
Some((n, v)) if n == "read_file" => v == p,
Some((n, v)) if n == "run_command" => !p.is_empty() && v.contains(p),
_ => false,
}
};
for i in 0..samples.len() {
let (na, pa) = match samples[i].as_ref() {
Some(s) => s,
@@ -353,8 +374,14 @@ pub(crate) fn detect_destructive_loop(samples: &[DestructiveLoopSample]) -> Opti
"delete_file" => "write_file",
_ => continue,
};
if is_temp_recycle(pa) {
continue;
}
// j 限 (i, i+GAP]:紧凑间隔内找反向同路径操作。
for j in (i + 1)..samples.len().min(i + 1 + DESTRUCTIVE_LOOP_GAP) {
if is_use_between(&samples[j], pa) {
break; // 对内有使用=用后清理,该写删对豁免
}
if let Some((nb, pb)) = samples[j].as_ref() {
if pb == pa && nb == opposite {
return Some(pa.clone());
@@ -614,6 +641,45 @@ mod tests {
assert_eq!(detect_destructive_loop(&samples), Some("p.txt".to_string()), "间隔 6 步 = GAP 应命中");
}
/// 豁免一:临时/回收路径(.trash、/tmp/ 等)写删是产物卫生流程,非自毁,不判回环。
#[test]
fn loop_temp_recycle_path_exempted() {
for p in [".trash/analyze.py", "C:/data/tmp/analyze.py", ".cache/scratch.txt"] {
let samples = loop_samples_of(&[
("write_file", serde_json::json!({"path": p, "content": "x"})),
("delete_file", serde_json::json!({"path": p})),
]);
assert_eq!(detect_destructive_loop(&samples), None, "临时路径 {} 写删应豁免", p);
}
}
/// 豁免二:写删对之间有对该路径的成功使用(run/read)是用后清理,非自毁。
#[test]
fn loop_use_between_pair_not_detected() {
let run_between = loop_samples_of(&[
("write_file", serde_json::json!({"path": "scripts/a.py"})),
("run_command", serde_json::json!({"command": "python scripts/a.py"})),
("delete_file", serde_json::json!({"path": "scripts/a.py"})),
]);
assert_eq!(detect_destructive_loop(&run_between), None, "写后运行过再删是用后清理");
let read_between = loop_samples_of(&[
("write_file", serde_json::json!({"path": "scripts/a.py"})),
("read_file", serde_json::json!({"path": "scripts/a.py"})),
("delete_file", serde_json::json!({"path": "scripts/a.py"})),
]);
assert_eq!(detect_destructive_loop(&read_between), None, "写后读过再删是确认后清理");
}
/// 真自毁仍拦:非临时路径写完即删,对内无任何使用。
#[test]
fn loop_bare_write_delete_still_detected() {
let samples = loop_samples_of(&[
("write_file", serde_json::json!({"path": "out.txt", "content": "x"})),
("delete_file", serde_json::json!({"path": "out.txt"})),
]);
assert_eq!(detect_destructive_loop(&samples), Some("out.txt".to_string()), "无使用的写完即删仍判自毁");
}
/// 场景样本不足:仅 3 个签名(即使全同)。判 false(不判)。
#[test]
fn scenario_insufficient_sample_not_repetitive() {
+92 -1
View File
@@ -771,6 +771,10 @@ pub(crate) async fn run_agentic_loop(
// 治"全 loop 累加不衰减":早期偶发失败不会与后期叠加误熔断。
let mut fail_window: std::collections::VecDeque<String> = std::collections::VecDeque::new();
// L1 断路器两段式警示位(对齐 G2 WARN_FIRST):同 key 失败达 THRESHOLD-1 时注入一次
// 系统警示给 LLM 自纠机会,达 THRESHOLD 才熔断。置位后本 loop 不重复警示。
let mut breaker_warned: bool = false;
// G2 探索熔断:重复探索计数器(loop 生命周期累计,与 fail_window 同生命周期)。
// 每轮 process_tool_calls 后 check_stall_breaker 取最近 N 个 assistant tool_calls 签名,
// 喂 is_repetitive_exploration 判重复,重复 stall_count+=1,非重复重置 0。
@@ -1086,7 +1090,7 @@ pub(crate) async fn run_agentic_loop(
&mut guard, &mut tokens, &mut saved_token_snapshot,
&resolved_model,
tool_calls_acc,
&mut fail_window,
&mut fail_window, &mut breaker_warned,
&mut stall_count, &mut stall_warned, &mut loop_count, &mut loop_warned,
).await {
return;
@@ -1612,6 +1616,24 @@ async fn finalize_round_output(
// 停止信号 或 已被新 loop 接管(stale):已生成文本入库后退出,不再执行后续工具调用。
// F1:stale 时旧 loop 在此退出,不执行工具(防重复工具执行);teardown_round 内部跳过 emit。
if stop_flag.load(Ordering::SeqCst) || loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
// 本轮 tool_calls 尚未执行,而 assistant 头(含 tool_calls)已 push 且 P0-2 已落库,
// 直接退出会给历史留下无 result 的悬空 tool_calls(sanitize 每请求告警的确定性供给)。
// 退出前逐个补跳过型 tool_result 闭环三元组,随后走既有 teardown save 落库。
// 短锁段纯写,锁边界与 epoch 语义不变:stale 时 teardown 因 epoch 不匹配跳过 save,
// 补的 result 留在内存,由 owner loop 的下一次 save 收敛落库。
if !tool_calls_acc.is_empty() {
let mut drafts: Vec<(u32, &super::ToolCallDraft)> =
tool_calls_acc.iter().map(|(k, d)| (*k, d)).collect();
drafts.sort_by_key(|(k, _)| *k);
let mut session = session_arc.lock().await;
if session.per_conv.contains_key(conv_id) {
let conv = session.conv(conv_id);
for (_, d) in drafts {
conv.messages.push(ChatMessage::tool_result(
&d.id, "用户停止或被接管,工具未执行"));
}
}
}
// save_usage 增量(自上次 save 后新增);emit_usage 累计快照(前端展示本 loop 总量)
let save_usage = usage_delta_since(tokens, saved_token_snapshot);
let emit_usage = tokens_snapshot(tokens);
@@ -1697,6 +1719,7 @@ async fn process_round_tools(
resolved_model: &str,
tool_calls_acc: std::collections::HashMap<u32, super::ToolCallDraft>,
fail_window: &mut std::collections::VecDeque<String>,
breaker_warned: &mut bool,
stall_count: &mut u32,
stall_warned: &mut bool,
loop_count: &mut u32,
@@ -1732,6 +1755,13 @@ async fn process_round_tools(
if CIRCUIT_BREAKER_ENABLED {
let (max_count, sample_key) = count_recent_failures(session_arc, conv_id, fail_window).await;
// 锁已随作用域 drop,可安全 await/emit(避免持锁 await 死锁)。
// 两段式(对齐 G2 WARN_FIRST 骨架):达 THRESHOLD-1 先经 insert_system_notice 注入
// 警示(尾部追加不破前缀),带失败样本 key 与建议动作,给 LLM 查环境/换工具/调参数的
// 自纠机会;达 THRESHOLD 才熔断。
if max_count == CIRCUIT_BREAKER_THRESHOLD.saturating_sub(1) && !*breaker_warned {
*breaker_warned = true;
insert_circuit_breaker_warning(session_arc, conv_id, iteration, &sample_key).await;
}
if max_count >= CIRCUIT_BREAKER_THRESHOLD {
tracing::warn!(
conv_id = %conv_id,
@@ -2114,6 +2144,30 @@ async fn insert_stall_warning(
.await;
}
// L1 断路器两段式警示:同 key 失败达 THRESHOLD-1 时注入系统软提示(尾部追加,不破前缀)。
// 文本带失败样本 key(工具与参数指纹、错误指纹)与建议动作,给 LLM 自纠机会,熔断前先警示。
async fn insert_circuit_breaker_warning(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
iteration: usize,
sample_key: &str,
) {
let warn_text = format!(
"⚠ 工具连续失败(样本: {})。同一做法再试大概率仍失败,建议先查环境(路径/权限/依赖是否就绪),再换用其他工具或调整参数与方案;若确认无法继续,直接向用户说明阻塞点。",
sample_key
);
helpers::insert_system_notice(
session_arc,
conv_id,
&warn_text,
&format!(
"L1 断路器:同类失败警示已 insert(软提示,给 LLM 自纠机会) iteration={} sample_key={}",
iteration, sample_key
),
)
.await;
}
// ── 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
@@ -2250,6 +2304,43 @@ async fn emit_ai_completed_once(
"AI 回复已就绪,点击托盘图标查看",
);
}
// 编排会话完成通知桥:工作流 AiAgentNode 挂起的等待方在此投递最终报告。
// 本函数是全部正常出口(5 类 teardown_round 出口 + 正常完成直调)的统一单点,
// 到达者必为 owner loop(epoch 守卫在各调用方:teardown_round 入口 / 正常完成双检),
// force_send 接管后的旧 loop 不触发。无等待方(普通用户会话)has 判空短路,零会话锁开销。
{
let app_state = app_handle.state::<AppState>();
if app_state.has_agent_completion_waiter(conv_id) {
let report = last_active_assistant_text(&app_state.ai_session, conv_id).await;
let delivered = app_state.notify_agent_completion(conv_id, report);
tracing::info!(
conv_id = %conv_id,
delivered,
"[ai] 编排会话收敛,完成通知桥投递最终报告"
);
}
}
}
/// 取会话末条 active assistant 消息文本(编排会话最终报告投递用)。
///
/// conv 不存在 / 无 assistant 消息返空串,原样投递,空文本由节点侧判「无输出」失败
/// (用户 stop 在任何输出前的场景)。读内存 per_conv 非常规出口:正常完成路径 save 落库
/// 在后台 spawn 里,时序上可能晚于本函数,内存消息是唯一已确定就绪的真相源。
async fn last_active_assistant_text(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
) -> String {
let session = session_arc.lock().await;
session
.conv_read(conv_id)
.map(|c| c.messages.all_messages_clone())
.unwrap_or_default()
.into_iter()
.rev()
.find(|m| matches!(m.role, MessageRole::Assistant) && m.is_active())
.map(|m| m.content)
.unwrap_or_default()
}
/// 构造 AiCompleted 事件(emit_ai_completed_once 内部去重用,字段口径见其文档)。
@@ -46,8 +46,9 @@ pub(crate) fn check_retry_exhausted(
}
// ── count_recent_failures: L1 断路器失败计数(扁平抽自原嵌套 5 层块) ──
// 读末尾连续 Tool 消息,失败包按结构化信号(tool_name + 错误类别)归一 key,
// 读末尾连续 Tool 消息,失败包按结构化信号(tool_name + 参数指纹 + 错误类别)归一 key,
// 追加进滚动窗口 fail_window(超窗 CIRCUIT_BREAKER_WINDOW 淘汰最旧)。
// 成功断连:只统计最近一段连续失败段,段中遇过成功先清空窗口(自适应达成即清零)。
// 返回 (max_count, sample_key):窗口内同 key 最高频次 + 该 key;max_count=0 表示无失败。
//
// 两处相对原实现的关键修正(原 key=内容前 40 字符 + 全 loop 累加不衰减):
@@ -72,7 +73,7 @@ pub(crate) async fn count_recent_failures(
None => Vec::new(),
}
};
// 末尾连续 Tool 消息(本轮工具回填结果)。
// 末尾连续 Tool 消息(本轮工具回填结果),最新在前。
let recent_tool_results: Vec<&ChatMessage> = messages
.iter().rev()
.take_while(|m| matches!(m.role, MessageRole::Tool))
@@ -81,21 +82,20 @@ pub(crate) async fn count_recent_failures(
return (0u32, String::new());
}
let id_to_name = build_id_to_name(&messages);
// 本轮失败的归一 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).map(|s| s.as_str()))
.unwrap_or("unknown_tool");
let key = failure_key(tool_name, &m.content);
// 成功断连:尾部回扫遇成功 tool_result 即停,只统计「最近一段连续失败段」内的失败;
// 段中遇过成功即自适应达成,清空窗口重计。红绿修复循环(失败失败成功失败)的成功
// 会切断与历史失败的累计,不再误熔断。
let (segment_keys, saw_success) = latest_failure_segment(&recent_tool_results, &id_to_name);
if saw_success {
fail_window.clear();
}
// 本段失败的归一 key 追加进窗口(顺序:消息正序,即工具执行顺序)。
for key in segment_keys {
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() {
@@ -107,20 +107,59 @@ pub(crate) async fn count_recent_failures(
.unwrap_or((0u32, String::new()))
}
/// 归一失败 key:tool_name + 错误类别。保证同一工具的同类错误稳定映射到同一 key。
/// 提取「最近一段连续失败段」的归一 key(最新在前扫描,遇首个非失败结果即停)。
///
/// - run_command 等带 exit_code → `tool_name::exit=N`(非零退出码即失败类别)。
/// - status=failed/error envelope → `tool_name::error:<错误文本稳定哈希>`:
/// 返回 (段内失败 key 按执行顺序, 回扫是否遇过成功)。成功=模型自适应达成,
/// 调用方据第二个返回值清空累计窗口,使红绿修复循环不再与历史失败叠加。
fn latest_failure_segment(
results_newest_first: &[&ChatMessage],
id_map: &std::collections::HashMap<String, (String, String)>,
) -> (Vec<String>, bool) {
let mut keys = Vec::new();
let mut saw_success = false;
for m in results_newest_first {
if !is_failure_content(&m.content) {
saw_success = true;
break;
}
let (tool_name, args_fp) = m
.tool_call_id
.as_deref()
.and_then(|id| id_map.get(id))
.map(|(n, fp)| (n.as_str(), fp.as_str()))
.unwrap_or(("unknown_tool", ""));
keys.push(failure_key(tool_name, args_fp, &m.content));
}
keys.reverse();
(keys, saw_success)
}
/// 参数指纹:args 稳定哈希(8 位十六进制)。解析成功按 serde_json 规范化序列化后哈希
/// (键序稳定),解析失败按原文哈希兜底。参数变=指纹变=失败 key 变:模型改参重试是
/// 学习行为,不应与旧参数的失败累计同 key。指纹非敏感信息,仅断路器分组用。
fn args_fingerprint(args: &str) -> String {
let canonical = serde_json::from_str::<serde_json::Value>(args)
.map(|v| v.to_string())
.unwrap_or_else(|_| args.to_string());
format!("{:08x}", fnv1a_32(canonical.as_bytes()))
}
/// 归一失败 key:tool_name + 参数指纹 + 错误类别。同工具同参数的同类错误稳定映射到
/// 同一 key;参数变=新 key=不累计(改参重试是学习行为)。
///
/// - run_command 等带 exit_code → `tool_name@指纹::exit=N`(非零退出码即失败类别)。
/// - status=failed/error envelope → `tool_name@指纹::failed:<错误文本稳定哈希>`:
/// 用 fxhash 风格简单 a 哈希避免裸截内容前缀(前 40 字符易把同类失败拆多 key
/// 或异类误聚)。哈希值非敏感信息(仅断路器分组用),无碰撞顾虑(N=20 窗口内几无碰撞)。
/// - 内容解析失败(非 JSON)→ is_failure_content 已返回 false 不会进本函数,
/// 此处兜底返回 `tool_name::unknown` 防御。
fn failure_key(tool_name: &str, content: &str) -> String {
/// 此处兜底返回 `tool_name@指纹::unknown` 防御。
fn failure_key(tool_name: &str, args_fp: &str, content: &str) -> String {
let base = format!("{}@{}", tool_name, args_fp);
let Ok(v) = serde_json::from_str::<serde_json::Value>(content) else {
return format!("{}::unknown", tool_name);
return format!("{}::unknown", base);
};
if let Some(exit) = v.get("exit_code").and_then(|x| x.as_i64()) {
return format!("{}::exit={}", tool_name, exit);
return format!("{}::exit={}", base, exit);
}
// 取 error 字段文本做稳定哈希;无 error 字段则按 status 兜底。
let bucket = match v.get("status").and_then(|s| s.as_str()) {
@@ -134,23 +173,26 @@ fn failure_key(tool_name: &str, content: &str) -> String {
.unwrap_or("")
.trim();
if err_text.is_empty() {
return format!("{}::{}", tool_name, bucket);
return format!("{}::{}", base, bucket);
}
let h = fnv1a_32(err_text.as_bytes());
format!("{}::{}:{:08x}", tool_name, bucket, h)
format!("{}::{}:{:08x}", base, bucket, h)
}
/// 构建 tool_call_id → tool_name 反查表(断路器失败分组与 tool_result 摘要压缩共用)。
/// 构建 tool_call_id → (tool_name, 参数指纹) 反查表(断路器失败分组与摘要压缩共用)。
///
/// Tool 结果消息只带 tool_call_id,真实工具名在前驱 Assistant 消息的
/// tool_calls[].function.name 里;查不到的 id 由调用方按各自语义兜底。
fn build_id_to_name(messages: &[ChatMessage]) -> std::collections::HashMap<String, String> {
/// Tool 结果消息只带 tool_call_id,真实工具名与参数在前驱 Assistant 消息的
/// tool_calls[].function 里;查不到的 id 由调用方按各自语义兜底。
fn build_id_to_name(
messages: &[ChatMessage],
) -> std::collections::HashMap<String, (String, String)> {
let mut id_to_name = 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.clone(), c.function.name.clone());
let fp = args_fingerprint(&c.function.arguments);
id_to_name.insert(c.id.clone(), (c.function.name.clone(), fp));
}
}
}
@@ -209,15 +251,21 @@ pub(crate) async fn last_round_has_match_evidence(
// 失败 content 由 audit/mod.rs 包成 JSON envelope {"status":"failed","error":...};
// 成功 content 是 result JSON(run_command 含 exit_code,其他工具无 status 字段即成功)或 namespace:// 占位符。
// 非 JSON(namespace 占位符 / 重试 guard skip 纯文本)→ 无结构化失败信号 → 不判(false)。
// run_command 的 exit_code!=0 是脚本语义(测试红绿/grep 无匹配是有效结果,模型改参重试即
// 自适应),不计失败;仅 envelope status=failed(超时/启动失败/拒绝执行)计失败。同命令
// 反复死磕由 G2 签名熔断兜底(run_command 签名=整条命令)。status 先判,防失败 envelope
// 兼带 exit_code 时被脚本语义豁免。
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 等:非零退出 = 真失败
if matches!(v.get("status").and_then(|s| s.as_str()), Some("failed") | Some("error")) {
return true;
}
matches!(v.get("status").and_then(|s| s.as_str()), Some("failed") | Some("error"))
|| v.get("error").is_some()
if v.get("exit_code").is_some() {
return false; // 脚本语义:结果 JSON 带 exit_code 即已正常执行,非零退出不算失败
}
v.get("error").is_some()
}
// ── summarize_tool_results: tool_result view-only 摘要压缩(扁平抽自原嵌套 map 块) ──
@@ -251,12 +299,12 @@ pub(crate) fn summarize_tool_results(
return m;
}
let content_len = m.content.len();
// 解析真实工具名:tool_call_id 查表;查不到(异常/老数据)退化为 "tool"。
// 解析真实工具名: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())
.map(|(n, _)| n.as_str())
.unwrap_or("tool");
// read_file 豁免压缩:AI 定向读代码,压缩 content 阉割代码分析能力。
// read_file handler 自带 limit 硬上限 2000 行,无爆 prompt 风险。
@@ -309,3 +357,85 @@ fn tool_view_decision_key(content: &str) -> u64 {
content.hash(&mut h);
h.finish()
}
#[cfg(test)]
mod tests {
use super::*;
use df_ai::provider::ToolCall;
/// 构造 assistant(tool_calls) 一条,返回消息与 id 反查表。
fn map_of(calls: Vec<ToolCall>) -> std::collections::HashMap<String, (String, String)> {
let msgs = vec![ChatMessage::assistant_with_tools("", calls)];
build_id_to_name(&msgs)
}
// B1:参数变计新 key,同参数稳定同 key。
#[test]
fn failure_key_changes_when_args_change() {
let map = map_of(vec![
ToolCall::new("c1", "run_command", r#"{"command":"cargo test foo"}"#),
ToolCall::new("c2", "run_command", r#"{"command":"cargo test bar"}"#),
ToolCall::new("c3", "run_command", r#"{"command":"cargo test foo"}"#),
]);
let (n1, f1) = map.get("c1").unwrap().clone();
let (n2, f2) = map.get("c2").unwrap().clone();
let (n3, f3) = map.get("c3").unwrap().clone();
assert_eq!(n1, "run_command");
assert_ne!(f1, f2, "参数变=指纹变");
assert_eq!(f1, f3, "同参数指纹稳定");
let body = r#"{"status":"failed","error":"boom"}"#;
assert_ne!(
failure_key(&n1, &f1, body),
failure_key(&n2, &f2, body),
"同工具不同参数的失败应计新 key(改参重试是学习行为)"
);
assert_eq!(
failure_key(&n1, &f1, body),
failure_key(&n3, &f3, body),
"同工具同参数同错误应稳定同 key"
);
}
// B2:尾部回扫遇成功即停,只统计最近一段连续失败段。
#[test]
fn latest_failure_segment_stops_at_success() {
let map = map_of(vec![ToolCall::new("c1", "run_command", r#"{"command":"x"}"#)]);
let fail = r#"{"status":"failed","error":"boom"}"#;
let ok = r#"{"exit_code":0,"stdout":"done"}"#;
let msgs = vec![
ChatMessage::tool_result("c1", ok), // 最旧:成功
ChatMessage::tool_result("c1", fail),
ChatMessage::tool_result("c1", fail), // 最新
];
let newest_first: Vec<&ChatMessage> = msgs.iter().rev().collect();
let (keys, saw_success) = latest_failure_segment(&newest_first, &map);
assert_eq!(keys.len(), 2, "只统计最近连续失败段(2 条)");
assert!(saw_success, "回扫遇过成功=自适应达成");
// 最新一条即成功:段空,窗口应清零。
let msgs2 = vec![ChatMessage::tool_result("c1", fail), ChatMessage::tool_result("c1", ok)];
let newest_first2: Vec<&ChatMessage> = msgs2.iter().rev().collect();
let (keys2, saw2) = latest_failure_segment(&newest_first2, &map);
assert!(keys2.is_empty());
assert!(saw2);
// 窗口内无成功:段完整保留,saw_success=false(跨轮累计语义不变)。
let msgs3 = vec![ChatMessage::tool_result("c1", fail), ChatMessage::tool_result("c1", fail)];
let newest_first3: Vec<&ChatMessage> = msgs3.iter().rev().collect();
let (keys3, saw3) = latest_failure_segment(&newest_first3, &map);
assert_eq!(keys3.len(), 2);
assert!(!saw3);
}
// B3:run_command 非零退出是脚本语义不计失败;envelope status=failed 仍计失败。
#[test]
fn run_command_exit_exempt_but_envelope_counts() {
assert!(!is_failure_content(r#"{"command":"cargo test","exit_code":1,"stdout":"FAILED"}"#));
assert!(!is_failure_content(r#"{"command":"grep x .","exit_code":0}"#));
assert!(is_failure_content(r#"{"status":"failed","error":"command timed out after 60s"}"#));
// status 先判:失败 envelope 兼带 exit_code 不被脚本语义豁免。
assert!(is_failure_content(r#"{"status":"failed","exit_code":-1,"error":"spawn failed"}"#));
// 含 error 字样的成功结果(读代码/搜失败字样)仍不判失败。
assert!(!is_failure_content(r#"{"exit_code":0,"stdout":"// error handling"}"#));
}
}