优化: 子代理截断信号与缓存失效

- max_tokens 截断标记穿透到完成卡
- 缓存写后失效补测试锁定
- 派发参数卡头信息补全
This commit is contained in:
lxy
2026-10-04 16:27:32 +08:00
parent d9e0ce73d1
commit 08e9d0292d
6 changed files with 165 additions and 17 deletions
@@ -69,6 +69,9 @@ enum StreamOutcome {
resolved_model: String,
/// DeepSeek thinking 模式推理内容(需回传到下一轮请求)
reasoning_content: Option<String>,
/// 流因 provider max_tokens 上限截断(finish_reason=length / stop_reason=max_tokens),
/// 本轮文本/tool_calls 可能不完整,调用方据此标记该轮产出截断。
truncated: bool,
},
/// error 字段供外层全 candidate 耗尽时 emit 最终 AiError(单气泡聚合)。
InitFailedExhausted { error: String },
@@ -175,10 +178,11 @@ async fn stream_with_retry(
).await;
match result {
StreamResult::Complete { text, tool_calls, usage, reasoning_content } => {
StreamResult::Complete { text, tool_calls, usage, reasoning_content, truncated_by_max_tokens } => {
return StreamOutcome::Success {
text, tool_calls, usage, incomplete: false,
resolved_model: resolved_model.to_string(), reasoning_content,
truncated: truncated_by_max_tokens,
};
}
StreamResult::Partial { text, tool_calls, usage, reasoning_content } => {
@@ -192,6 +196,8 @@ async fn stream_with_retry(
return StreamOutcome::Success {
text, tool_calls, usage, incomplete: true,
resolved_model: resolved_model.to_string(), reasoning_content,
// MidStream 已按不完整处理(incomplete=true),截断标志不再叠加
truncated: false,
};
}
StreamResult::InitFailed { retryable, error } => {
@@ -257,6 +263,9 @@ pub(crate) enum RoundOutcome {
/// 本轮 prompt 是否估算兜底(= 进 stream_round 时的 round_usage.prompt_tokens == 0),
/// 供 loop 主体 push 消息级 is_estimated,与 round_usage 同点派生防漂移。
round_is_estimated: bool,
/// 本轮流因 provider max_tokens 上限截断(P2-1),文本/tool_calls 可能不完整。
/// 子循环据此把收敛报告标 Truncated,不冒充 Completed;主循环据此提示用户。
truncated_by_max_tokens: bool,
},
/// 流返回后 stale(epoch 已变,被新 loop 接管):guard 已 disarm,
/// loop 应直接 return(不再 push/save/执行工具)。
@@ -265,8 +274,9 @@ pub(crate) enum RoundOutcome {
/// teardown_round save/reset/emit)→ loop 直接 return。
MidStreamSaved,
/// Fatal(候选 Fatal 或全候选耗尽):guard.reset + epoch 检查 + emit AiError +
/// save 已完成 → loop 直接 return。
Fatal,
/// save 已完成 → loop 直接 return。`error` 为最后一条诊断文本(P2-4:子循环据此
/// 在出口报告写明失败原因,父会话可针对性补救,不再拿到无因失败)。
Fatal { error: String },
}
/// 单轮流式调用:抽自 `run_agentic_loop` for 循环体的「单轮流式」自包含块。
@@ -365,9 +375,9 @@ pub(crate) async fn stream_round(
);
}
let (full_text, tool_calls_acc, round_usage, incomplete, round_reasoning_content) = {
let (full_text, tool_calls_acc, round_usage, incomplete, round_reasoning_content, round_truncated) = {
// outcome 累积成功结果(Complete/Partial),InitFailed 不写入。
let mut outcome: Option<(String, std::collections::HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage, bool, Option<String>)> = None;
let mut outcome: Option<(String, std::collections::HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage, bool, Option<String>, bool)> = None;
// 追踪最后一个 Exhausted candidate 的诊断文本,供「全 candidate 耗尽」时 emit 最终 AiError。
// Fatal 分支即时 emit(终态不切候选),Exhausted 分支仅记录 error 不 emit(可能切下一 candidate 成功,
// emit 会留残留气泡)。
@@ -418,14 +428,14 @@ pub(crate) async fn stream_round(
).await;
match stream_outcome {
StreamOutcome::Success { text, tool_calls, usage, incomplete, resolved_model: m, reasoning_content: rc } => {
StreamOutcome::Success { text, tool_calls, usage, incomplete, resolved_model: m, reasoning_content: rc, truncated } => {
// 成功:更新迭代级 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.clone();
outcome = Some((text, tool_calls, usage, incomplete, last_reasoning_content.clone()));
outcome = Some((text, tool_calls, usage, incomplete, last_reasoning_content.clone(), truncated));
break 'candidate;
}
StreamOutcome::InitFailedExhausted { error } => {
@@ -449,14 +459,14 @@ pub(crate) async fn stream_round(
guard.reset().await;
// F1:旧 loop(被新 loop 接管)不 emit 错误气泡(owner loop 负责呈现)。
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return RoundOutcome::Fatal;
return RoundOutcome::Fatal { error };
}
emit_fatal_error(app_handle, conv_id, &error).await;
// G1.1(2026-08-05):Fatal 退出也落库 user 消息(镜像下方 Exhausted 分支),
// 否则重启/切会话从 DB 恢复时末条 user 丢失(与 Exhausted 同根因不对称)。
// epoch 校验通过后、return 前补 save(owner loop 才写,防 owner 已转移仍写)。
save_conversation(session_arc, db, conv_id, None, Some(&*resolved_model), true).await;
return RoundOutcome::Fatal;
return RoundOutcome::Fatal { error };
}
}
}
@@ -471,16 +481,16 @@ pub(crate) async fn stream_round(
"[ai] 全 provider 候选流前失败重试耗尽,放弃本轮",
);
guard.reset().await;
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return RoundOutcome::Fatal;
}
let err_msg = last_exhausted_error
.unwrap_or_else(|| "AI 调用失败:所有候选 provider 重试耗尽".to_string());
if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch {
return RoundOutcome::Fatal { error: err_msg };
}
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 RoundOutcome::Fatal;
return RoundOutcome::Fatal { error: err_msg };
}
}
};
@@ -584,6 +594,7 @@ pub(crate) async fn stream_round(
tool_calls: tool_calls_acc,
round_usage: round_usage_full,
round_is_estimated,
truncated_by_max_tokens: round_truncated,
}
}
+80 -2
View File
@@ -257,6 +257,11 @@ pub(crate) fn readonly_cache_args_key(tool_name: &str, args: &serde_json::Value)
///
/// 与 find_cached_high_risk_result 区别:后者治"超时→重试→重新审批"循环( High risk),
/// 本函数治"LLM 失忆→同参重调"死循环( Low risk 只读),问题形态相似但触发原因不同。
///
/// **写后失效**(治「只读结果缓存无写失效」缺陷):会话内 write_file/run_command 等
/// 非只读工具成功执行后,历史只读结果已可能过期(读的是改前世界)。本函数扫描时按
/// per_conv.readonly_cache_floor 过滤——floor 之后的索引才算可信,之前的旧调用全部
/// 跳过(过界即脏),写后 LLM 重读同参工具必然 miss 缓存走真执行拿到新内容。
pub(crate) async fn find_cached_readonly_result(
session_arc: &Arc<Mutex<AiSession>>,
conv_id: &str,
@@ -281,7 +286,11 @@ pub(crate) async fn find_cached_readonly_result(
let cached: Option<(String, String, u32)> = (async {
let session = session_arc.lock().await;
let Some(conv) = session.conv_read(conv_id) else { return None };
let msgs: Vec<&ChatMessage> = conv.messages.iter().collect();
// 写后失效:floor 之前的只读结果一律不可信(写过世界状态),只扫 [floor, len)。
// floor > len 防御(pop_last_assistant_round/clear 缩短历史后索引漂移),钳到 len
// 等于扫描区间空 → 无缓存可命中,走真执行(保守方向安全)。
let floor = conv.readonly_cache_floor.min(conv.messages.len());
let msgs: Vec<&ChatMessage> = conv.messages.iter().skip(floor).collect();
// 1) 反向扫描 assistant tool_calls,找最近一条同名同参调用 → 拿到旧 tool_call_id,
// 同时累计同名同参调用总数(hit_count 供「第 N 次命中缓存」威慑)。
@@ -342,6 +351,29 @@ pub(crate) async fn find_cached_readonly_result(
/// 只读缓存白名单已收敛为注册元数据(readonly 维度,DRY 债 #5):
/// find_cached_readonly_result 改查 df_ai::ai_tools::is_readonly,本文件不再持名单。
/// 写后失效:非只读工具成功回填后推进会话级缓存失效下界(readonly_cache_floor)。
///
/// 调用点 = 全部写工具真执行成功的回填处(Low 风险并行回填 / trust 放行回填 /
/// 边界放行回填 / ai_approve 审批后执行回填)。非只读工具(写文件/跑命令/改 DB 等)
/// 成功执行即视作世界状态变更,历史里的只读结果从此不可信——floor 推进到写结果
/// push 之后,find_cached_readonly_result 扫描时跳过 floor 之前的全部旧调用(过界即脏)。
///
/// 单调不减(max):多写工具同轮/跨轮先后执行,floor 取最大(最后写),前段只读
/// 结果全失效——保守方向安全(失效仅多一次真执行,误命中才喂脏数据)。
///
/// 须在持 session 锁的短临界区内调用(与写结果 push 同锁,防 floor 与 messages 漂移)。
/// is_readonly 未注册的工具按非只读处理(保守推进 floor,防未知新写工具漏失效)。
pub(crate) fn mark_readonly_cache_stale(conv: &mut super::super::PerConvState, tool_name: &str) {
crate::commands::ai::tool_registry::ensure_tool_meta_seeded();
if df_ai::ai_tools::is_readonly(tool_name) {
return; // 只读工具不改世界状态,缓存照常有效
}
let new_floor = conv.messages.len();
if conv.readonly_cache_floor < new_floor {
conv.readonly_cache_floor = new_floor;
}
}
/// 递归对 JSON 对象的键做字典序排序(就地),数组成员也递归排序。
fn sort_object_keys(v: &mut serde_json::Value) {
match v {
@@ -545,7 +577,53 @@ mod tests {
assert_ne!(a, b, "read_file 不同 offset 应不同 key(分段读正常)");
}
// ── AC-2 ③ 列举命令检测测试 ──
// ── 写后失效测试 ──
/// mark_readonly_cache_stale:非只读工具推进 floor,只读工具不动。
#[test]
fn mark_stale_advances_floor_for_write_tools_only() {
crate::commands::ai::tool_registry::ensure_tool_meta_seeded();
let mut conv = super::super::super::PerConvState::new();
// 三条历史消息(user + assistant + tool)
conv.messages.push(df_ai::provider::ChatMessage::user("hi"));
conv.messages.push(df_ai::provider::ChatMessage::assistant("ack"));
conv.messages.push(df_ai::provider::ChatMessage::tool_result("t1", "r"));
assert_eq!(conv.readonly_cache_floor, 0, "初值 0");
// read_file 只读:不推进
mark_readonly_cache_stale(&mut conv, "read_file");
assert_eq!(conv.readonly_cache_floor, 0, "只读工具不推进 floor");
// write_file 写:推进到当前 len(3)
mark_readonly_cache_stale(&mut conv, "write_file");
assert_eq!(conv.readonly_cache_floor, 3, "写工具推进 floor 到 len");
// 再只读:不动(单调不减由 max 保证)
mark_readonly_cache_stale(&mut conv, "grep");
assert_eq!(conv.readonly_cache_floor, 3);
// run_command(写语义):取 max 不回退
conv.messages.push(df_ai::provider::ChatMessage::user("again"));
mark_readonly_cache_stale(&mut conv, "run_command");
assert_eq!(conv.readonly_cache_floor, 4, "第二个写工具推进到新 len");
}
/// floor 过滤逻辑纯函数级锁定:floor=min 钳制语义(消息缩短后 floor 不越界)。
#[test]
fn readonly_cache_floor_clamped_by_len() {
let mut conv = super::super::super::PerConvState::new();
conv.messages.push(df_ai::provider::ChatMessage::user("a"));
conv.messages.push(df_ai::provider::ChatMessage::user("b"));
mark_readonly_cache_stale(&mut conv, "write_file");
assert_eq!(conv.readonly_cache_floor, 2);
// 模拟 pop_last_assistant_round/clear 缩短历史后:len=0,floor 语义钳到 0
let floor = conv.readonly_cache_floor.min(conv.messages.len());
conv.messages.clear();
let floor2 = conv.readonly_cache_floor.min(conv.messages.len());
assert_eq!(floor, 2, "len 未缩短时 floor 原样");
assert_eq!(floor2, 0, "len 缩短后 floor 钳到 0(空扫描区间,安全)");
}
#[test]
fn listing_command_detects_powershell_and_unix() {
@@ -7,7 +7,7 @@
//! `resolve_all` 按 [`MentionRef::kind_str`] 分发到对应 Resolver,单条 resolve 失败
//! 收集进 errors 列表不阻断整批注入(返回 `(Vec<Augmentation>, Vec<ResolveError>)`)。
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use df_types::augmentation::{Augmentation, MentionRef, ResolveError};
@@ -45,6 +45,10 @@ impl ResolverRegistry {
/// - 失败(Resolver 返 Err / kind 无对应 Resolver)→ push 进 `errors`,**不中断**
/// 后续 mention(整批注入最大化保留可用上下文)。
///
/// 同一引用去重(先到先得):`/技能` 前导与 @mention span 可指向同一技能,按
/// `kind + ref_id`(skill 的 ref_id 即 name)跳过重复,避免同一正文重复 resolve
/// 并在 prompt 注入两份。首现位置决定保留哪条,顺序不变。
///
/// 返回 `(成功列表, 失败列表)`,调用方据 errors 数决定是否提示用户某 mention 跳过。
pub async fn resolve_all(
&self,
@@ -53,7 +57,12 @@ impl ResolverRegistry {
) -> (Vec<Augmentation>, Vec<ResolveError>) {
let mut augs = Vec::with_capacity(refs.len());
let mut errors = Vec::new();
let mut seen: HashSet<String> = HashSet::with_capacity(refs.len());
for rf in refs {
// 去重键 = kind + ref_id(id 或 skill 的 name):同键 mention 只 resolve 首条
if !seen.insert(format!("{}:{}", rf.kind_str(), rf.ref_id())) {
continue;
}
let kind = rf.kind_str();
match self.resolvers.get(kind) {
Some(resolver) => match resolver.resolve(rf, loc).await {
@@ -139,4 +148,48 @@ mod tests {
assert!(augs.is_empty(), "两条 mention kind 均≠test 应全 KindMismatch");
assert_eq!(errors.len(), 2, "两条失败都应收集");
}
/// 测试用 Resolver:kind 固定 "skill",resolve 固定返 Skill(模拟成功注入)。
struct OkSkillResolver;
#[async_trait]
impl MentionResolver for OkSkillResolver {
fn kind(&self) -> &'static str {
"skill"
}
async fn resolve(
&self,
rf: &MentionRef,
_loc: ProviderLocality,
) -> Result<Augmentation, ResolveError> {
let name = match rf {
MentionRef::Skill { name } => name.clone(),
_ => String::new(),
};
Ok(Augmentation::Skill {
name,
source: "test".into(),
body: "BODY".into(),
})
}
}
#[tokio::test]
async fn resolve_all_dedups_same_ref_first_wins() {
// 去重(先到先得):/技能 前导 + @mention span 指向同一技能时只 resolve/注入一份;
// 不同技能不受影响;首现位置决定保留顺序。
let mut reg = ResolverRegistry::new();
reg.register(Arc::new(OkSkillResolver));
let refs = vec![
MentionRef::Skill { name: "review".into() },
MentionRef::Skill { name: "review".into() },
MentionRef::Skill { name: "idea".into() },
];
let (augs, errors) = reg.resolve_all(&refs, ProviderLocality::Local).await;
assert!(errors.is_empty());
assert_eq!(augs.len(), 2, "同名技能两条应去重为一份");
match &augs[0] {
Augmentation::Skill { name, .. } => assert_eq!(name, "review", "先到先得保留首条"),
other => panic!("expected Skill, got {:?}", other),
}
}
}
+3 -1
View File
@@ -90,7 +90,9 @@ async function doSend(text: string, skill?: string, force = false, parts?: Conte
state.messages.push({
id: userMsgId as MessageId,
role: 'user',
content: text.trim(),
// 技能调用留痕(问题3,与后端 chat.rs user_content 同规则):带技能时本地气泡即刻补
// /技能名 前缀,与落库文本一致——否则发送瞬间无痕、切回会话才见前缀,前后显示不一。
content: skill ? `/${skill} ${text.trim()}` : text.trim(),
// 仅在非空时挂 parts(纯文本消息保持 undefined,渲染走原 content 路径零回归)
parts: parts && parts.length > 0 ? parts : undefined,
// Input Augmentation: 仅在非空时挂 mentionSpans(供消息列表 chip 渲染;空则 undefined 零回归)
+2
View File
@@ -117,6 +117,8 @@ export default {
// fetch_url header display name (useToolCardHeader): host-shortened prefix + fallback when no url
fetchUrlPrefix: 'Fetch',
fetchUrlFallback: 'Fetch URL',
// spawn_agent header display name (P3-3: task truncated to 25 chars, consumes backend "dispatch sub-agent" goal metadata)
spawnAgentPrefix: 'Spawn agent: {task}',
installObscura: 'Install obscura via AI',
obscuraInstallPrompt: 'Please install obscura (a Rust headless browser, used by fetch_url render=true for JS rendering). Try `cargo install obscura` first; if cargo is unavailable, download the prebuilt binary for this platform from the GitHub releases (github.com/h4ckf0r0day/obscura) and place it on PATH. After installing, verify with `obscura --version` and tell me I can retry the web fetch.',
// grep cross-file content search results per output_mode phrasing
+2
View File
@@ -118,6 +118,8 @@ export default {
// fetch_url 头部显示名(useToolCardHeader):host 精简前缀 + 无 url 时兜底
fetchUrlPrefix: '获取',
fetchUrlFallback: '获取网页',
// spawn_agent 头部显示名(P3-3:任务书截 25 字,消费后端「派发子代理」目标元数据语义)
spawnAgentPrefix: '派发子代理: {task}',
installObscura: 'AI 帮装 obscura',
obscuraInstallPrompt: '请帮我安装 obscura(Rust 无头浏览器,用于 fetch_url 的 render=true JS 渲染)。优先尝试 `cargo install obscura`,若 cargo 不可用则从 GitHub releases(github.com/h4ckf0r0day/obscura)下载对应平台预编译二进制并放入 PATH。安装完成后用 `obscura --version` 验证,然后告诉我可以重试网页抓取了。',
// grep 跨文件内容搜索结果按 output_mode 分文案