From d406a3172238423eafac06a28305a9844158be39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sat, 3 Oct 2026 15:35:08 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96:=20agentic=E9=98=B2=E6=8A=A4?= =?UTF-8?q?=E4=B8=8E=E5=88=A0=E9=99=A4=E5=AE=A1=E6=89=B9=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=9F=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 循环主体 catch_unwind 防护,panic 走正典终态不再卡死 - 会话删除停 in-flight 循环+删除侧单点补发终态 - 审批超时到点主动收敛,执行等待循环加删除判死 - 子代理出口整轨迹落库,拒绝工具结构化+报告协议 - 压缩会话 id 修正,begin_turn 喂用户原文 - 远程桥会话列表按 origin 过滤子会话,新增中断续跑命令 --- .../commands/ai/agentic/context_lifecycle.rs | 8 +- src-tauri/src/commands/ai/agentic/helpers.rs | 213 ++++++-- src-tauri/src/commands/ai/agentic/mod.rs | 504 ++++++++++++------ .../src/commands/ai/agentic/stream_relay.rs | 2 +- .../src/commands/ai/agentic/sub_agent.rs | 192 ++++++- src-tauri/src/commands/ai/audit/mod.rs | 266 ++++++++- src-tauri/src/commands/ai/commands/chat.rs | 169 ++++-- .../src/commands/ai/commands/conversation.rs | 44 +- src-tauri/src/commands/ai/conversation.rs | 6 +- src-tauri/src/commands/ai/mod.rs | 13 + src-tauri/src/commands/ai/remote_bridge.rs | 26 +- src-tauri/src/lib.rs | 2 + 12 files changed, 1139 insertions(+), 306 deletions(-) diff --git a/src-tauri/src/commands/ai/agentic/context_lifecycle.rs b/src-tauri/src/commands/ai/agentic/context_lifecycle.rs index 49e8fbb..9389232 100644 --- a/src-tauri/src/commands/ai/agentic/context_lifecycle.rs +++ b/src-tauri/src/commands/ai/agentic/context_lifecycle.rs @@ -119,8 +119,12 @@ pub(super) async fn maybe_auto_compress( let conv = session.conv(&conv_id); conv.messages.set_compressing(true); // 读 active 克隆(不改 status):filter is_active,LLM 失败则消息状态完全不变。 - let active_msgs: Vec = conv.messages.messages_mut() - [..protect_start] + // protect_start 是上一段锁内算的快照,两段锁之间消息可能被清空/裁剪变短, + // 切片前按当时实际长度钳制,防 index out of bounds panic + // (后续 compress_old_messages 内部自带 min(len) 钳制,口径一致)。 + let msgs = conv.messages.messages_mut(); + let cut = protect_start.min(msgs.len()); + let active_msgs: Vec = msgs[..cut] .iter() .filter(|t| t.message.is_active()) .map(|t| t.message.clone()) diff --git a/src-tauri/src/commands/ai/agentic/helpers.rs b/src-tauri/src/commands/ai/agentic/helpers.rs index 52233da..b3df3ea 100644 --- a/src-tauri/src/commands/ai/agentic/helpers.rs +++ b/src-tauri/src/commands/ai/agentic/helpers.rs @@ -96,15 +96,17 @@ pub async fn try_continue_agent_loop( Some(cid) => cid, None => conv_id.to_string(), }; - // 零 token 收敛信号,非估算(无真实 LLM 调用)。 - emit_ai_chat_event(app, AiChatEvent::AiCompleted { - total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, - prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 0, reasoning_tokens: 0, - is_estimated: false, - incomplete: None, - conversation_id: Some(emit_conv_id), - pinned_goals: snap.pinned_goals_snapshot.clone(), - }); + // 走正典 emit_ai_completed_once(零 token 收敛信号,非估算,事件字段与裸 emit 等价): + // 随正典出口一并终态化占位 tool_result + 清该 conv 的 pending_approvals, + // 防挂起审批残留的 __PENDING__: 占位下次发送喂给 LLM。 + // notify=false:本站点是 generating 已复位的收尾信号,非真实完成,不发系统通知 + // (对齐原裸 emit 无通知行为)。 + let zero_usage = df_ai::provider::TokenUsage::default(); + super::emit_ai_completed_once( + app, &emit_conv_id, &zero_usage, false, + None, None, None, true, + &snap.pinned_goals_snapshot, false, + ).await; } return; } @@ -167,15 +169,17 @@ pub async fn try_continue_agent_loop( // B-Phase2:ConvState 读侧切无锁 conv_states(零锁竞争,无需 session lock)。 let still_generating = state.conv_states.is_active(conv_id); if !still_generating { - // 零 token 收敛信号,非估算(无真实 LLM 调用)。 - emit_ai_chat_event(app, AiChatEvent::AiCompleted { - total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, - prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 0, reasoning_tokens: 0, - is_estimated: false, - incomplete: None, - conversation_id: Some(conv_id_owned.clone()), - pinned_goals: snap.pinned_goals_snapshot.clone(), - }); + // 走正典 emit_ai_completed_once(零 token 收敛信号,非估算,事件字段与裸 emit 等价): + // 随正典出口一并终态化占位 tool_result + 清该 conv 的 pending_approvals, + // 防挂起审批残留的 __PENDING__: 占位下次发送喂给 LLM。 + // notify=false:本站点是 generating 已复位的收尾信号,非真实完成,不发系统通知 + // (对齐原裸 emit 无通知行为)。 + let zero_usage = df_ai::provider::TokenUsage::default(); + super::emit_ai_completed_once( + app, &conv_id_owned, &zero_usage, false, + None, None, None, true, + &snap.pinned_goals_snapshot, false, + ).await; return; } @@ -266,34 +270,49 @@ pub(crate) fn tool_call_signature(name: &str, args: &serde_json::Value) -> Strin /// /// 样本 = (签名, 是否写类调用)。策略(后两条任一命中即 true): /// - 样本不足(len < `REPETITION_MIN_SAMPLE`=6)→ false(不判,小样本误杀风险高)。 -/// - **唯一签名数 / 总数 < 0.4**(超 60% 重复)→ true。 -/// 覆盖「多个签名轮换但整体高度重复」(如 a/b/c/a/b/c/d/a/b),唯一率低 = 没有新探索方向。 +/// - **有效单元数 / 总样本数 < 0.4**(折损超 60%)→ true。 +/// 覆盖「多个签名轮换但整体高度重复」(如 a/b/c/a/b/c/d/a/b),有效单元占比低 = 没有新探索方向。 /// - **或:某签名「无写间隔连续段」>= 3** → true。覆盖「单点反复」。 /// -/// 写间隔豁免(脚本失败致执行中断的根治):验证命令(测试/构建/lint)反复跑但两次之间 +/// 有效单元数(唯一率分子):每签名首现 +1;再现时满足任一豁免也 +1—— +/// 「与上次出现间夹有写类样本(区间不含自身:世界已变,重查是新鲜验证活动)」或 +/// 「自身是写类」(写类对称豁免,对齐单点分支「写不作死磕计数对象」语义); +/// 未豁免的再现 +0(无写证据的重复折损)。豁免的再现计为新单元而非去重删除: +/// 写后重验是世界变化后的新活动,不与历史失败累计,分母保持全样本数不变。 +/// +/// 写间隔豁免(两分支对称):验证命令(测试/构建/lint/grep 验证残留)反复跑但两次之间 /// 夹有写类调用(patch/write 改了世界)是红绿迭代——重跑是验证行为,不与历史失败累计; -/// 中间只有读类调用不算证据(世界未变,重跑仍是无进展重复)。原「总出现次数 >= 3」口径 -/// 会把红绿迭代误判漂移(12 样本窗口内同一测试命令必然出现 >= 3 次),阈值不变、纯死磕 -/// 语义不变,只把"死磕"修准为"中间无写证据的重复"。 +/// 中间只有读类调用不算证据(世界未变,重跑仍是无进展重复)。单点分支按「无写间隔连续段」 +/// 计数;唯一率分支按上述有效单元数口径——否则清理任务「write→grep 验证→write→grep 验证」 +/// 窗口内同 grep 签名重复超 60% 会误判游荡。 +/// 阈值不变、纯死磕语义不变,只把"重复折损"修准为"无写证据的重复"。 pub(crate) fn is_repetitive_exploration(samples: &[(String, bool)]) -> bool { /// 最小样本量:不足此数不判定(避免早期误杀,如刚启动 2-3 个 grep 全不同不应熔断)。 const REPETITION_MIN_SAMPLE: usize = 6; /// 单签名无写间隔连续段阈值:达此即判单点死磕漂移。 const REPETITION_SINGLE_MAX: usize = 3; - /// 唯一签名占比阈值:低于此(重复超 60%)判整体游荡不前进。 + /// 有效单元占比阈值:低于此(折损超 60%)判整体游荡不前进。 const REPETITION_UNIQUE_RATIO: f64 = 0.4; if samples.len() < REPETITION_MIN_SAMPLE { return false; } let total = samples.len(); - let unique = { - let mut s: Vec<&String> = samples.iter().map(|(sig, _)| sig).collect(); - s.sort(); - s.dedup(); - s.len() - }; - let unique_ratio = unique as f64 / total as f64; + // 有效单元数:首现 +1;再现时「与上次出现间夹写(samples[j+1..i],不含自身)」或 + // 「自身是写」则豁免 +1,否则 +0 折损。HashMap 按签名记上次下标,一趟扫 O(n)。 + let mut last_idx: std::collections::HashMap<&str, usize> = std::collections::HashMap::new(); + let mut units = 0usize; + for (i, (sig, is_write)) in samples.iter().enumerate() { + let is_new_unit = match last_idx.get(sig.as_str()) { + None => true, // 首现 + Some(&j) => *is_write || samples[j + 1..i].iter().any(|(_, w)| *w), + }; + if is_new_unit { + units += 1; + } + last_idx.insert(sig.as_str(), i); + } + let unique_ratio = units as f64 / total as f64; if unique_ratio < REPETITION_UNIQUE_RATIO { return true; } @@ -778,6 +797,136 @@ mod tests { assert!(!is_repetitive_exploration(&reps(sigs))); } + // ── 唯一率分支写间隔豁免测试 ── + + /// 场景清理任务红绿验证:grep 同签名反复验证残留,但每次再现前都夹有 write(世界已变, + /// 重查是验证行为)——唯一率分支豁免后不应判游荡。旧口径 12 样本窗口内同签名重复 6 次 + /// 必误熔断。 + #[test] + fn scenario_cleanup_verify_with_writes_not_repetitive() { + let g = tool_call_signature("grep", &serde_json::json!({"pattern": "TODO", "path": "src"})); + let run = tool_call_signature("run_command", &serde_json::json!({"command": "cargo check"})); + let read = tool_call_signature("read_file", &serde_json::json!({"path": "a.rs", "offset": 0, "limit": 50})); + let w = |n: &str| { + tool_call_signature("patch_file", &serde_json::json!({"path": format!("{}.rs", n), "old": "x", "new": "y"})) + }; + // grep 位于 1/3/5/8/10,每次再现与上次之间都夹 patch_file 写。 + let samples: Vec<(String, bool)> = vec![ + (run.clone(), false), + (g.clone(), false), + (w("a"), true), + (g.clone(), false), + (w("b"), true), + (g.clone(), false), + (read.clone(), false), + (w("c"), true), + (g.clone(), false), + (w("d"), true), + (g.clone(), false), + (w("e"), true), + ]; + assert_eq!(samples.len(), 12); + assert!(!is_repetitive_exploration(&samples), "清理任务 grep 验证+每次夹写不应判游荡"); + } + + /// 场景纯游荡仍熔断:全读无写,豁免无写证据可用,唯一率 3/9≈0.33 < 0.4 → true(语义不变)。 + #[test] + fn scenario_pure_wandering_still_repetitive() { + let seq = ["a", "b", "a", "c", "a", "b", "a", "c", "a"]; + let samples: Vec<(String, bool)> = seq.iter() + .map(|p| (tool_call_signature("grep", &serde_json::json!({"pattern": p, "path": "x"})), false)) + .collect(); + assert!(is_repetitive_exploration(&samples), "无写间隔的轮换游荡应判重复"); + } + + /// 场景紧邻同签名死磕:紧邻重复(j+1==i 空区间)不豁免,计入唯一率折损,且单点 + /// 无写间隔 streak 达 3,双分支兜底 → true。 + #[test] + fn scenario_adjacent_grinding_still_repetitive() { + let seq = ["a", "a", "b", "a", "a", "w", "a", "a", "b"]; + let samples: Vec<(String, bool)> = seq.iter() + .enumerate() + .map(|(i, s)| { + let sig = tool_call_signature("read_file", + &serde_json::json!({"path": format!("{}.txt", s), "offset": 0, "limit": 50})); + (sig, i == 5) // 仅 "w" 位是写类调用 + }) + .collect(); + assert!(is_repetitive_exploration(&samples), "紧邻重复(无写间隔)死磕应仍判重复"); + } + + /// 场景写间隔豁免翻转判定:同签名序列 [g,w1,g,w2,g,w3],无写版 g 三连无写间隔 + /// streak 达 3 → true;有写版每次再现夹写全部豁免 → false。 + #[test] + fn scenario_write_interval_exemption_flips_verdict() { + let g = tool_call_signature("grep", &serde_json::json!({"pattern": "v", "path": "x"})); + let w = |p: &str| { + tool_call_signature("patch_file", &serde_json::json!({"path": p, "old": "a", "new": "b"})) + }; + let no_write: Vec<(String, bool)> = vec![ + (g.clone(), false), (w("p1"), false), (g.clone(), false), + (w("p2"), false), (g.clone(), false), (w("p3"), false), + ]; + let with_write: Vec<(String, bool)> = vec![ + (g.clone(), false), (w("p1"), true), (g.clone(), false), + (w("p2"), true), (g.clone(), false), (w("p3"), true), + ]; + assert!(is_repetitive_exploration(&no_write), "同序列无写版应判死磕"); + assert!(!is_repetitive_exploration(&with_write), "有写版每次再现夹写应豁免不判"); + } + + /// 读/写同签名交替 [g,w]×6(g 同签名读,w 同签名写)。推演:g 首现 +1,再现均 + /// 夹 w 豁免 +1(5 次);w 首现 +1,再现自身是写豁免 +1(5 次)→ units=12,ratio=12/12=1.0; + /// 单点分支 g 段间恒夹写 max_streak=1,w 不计数 → false。 + /// 旧算法豁免再现不入集合(首现已入,dedup 零影响)unique=2,ratio=2/12 误熔断 true(红绿回归锚点)。 + #[test] + fn scenario_read_write_alternation_not_repetitive() { + let g = tool_call_signature("grep", &serde_json::json!({"pattern": "v", "path": "x"})); + let w = tool_call_signature("patch_file", &serde_json::json!({"path": "p.rs", "old": "a", "new": "b"})); + let samples: Vec<(String, bool)> = (0..6) + .flat_map(|_| [(g.clone(), false), (w.clone(), true)]) + .collect(); + assert_eq!(samples.len(), 12); + assert!(!is_repetitive_exploration(&samples), "读/写同签名交替(红绿迭代)不应判漂移"); + } + + /// 纯同签名读死磕 [g]×6。推演:首现 1 单元,再现区间空且非写全折损 → + /// units=1,ratio=1/6≈0.17 < 0.4 → true。 + #[test] + fn scenario_pure_read_grinding_repetitive() { + let g = tool_call_signature("grep", &serde_json::json!({"pattern": "v", "path": "x"})); + let samples: Vec<(String, bool)> = vec![(g, false); 6]; + assert!(is_repetitive_exploration(&samples), "纯同签名读死磕应判漂移"); + } + + /// 全读轮换游荡 [a,b,c,a,b,c,d,a,b]。推演:再现间无写全折损,units=4 + /// (a/b/c/d 首现),ratio=4/9≈0.44 未低于 0.4(唯一率分支不触发);a 无写间隔 + /// streak=3(位 0/3/7)达单点阈值 → true(经单点分支兜底命中)。 + #[test] + fn scenario_all_read_wandering_repetitive() { + let sig = |p: &str| tool_call_signature("grep", &serde_json::json!({"pattern": p, "path": "x"})); + let seq = ["a", "b", "c", "a", "b", "c", "d", "a", "b"]; + let samples: Vec<(String, bool)> = seq.iter().map(|p| (sig(p), false)).collect(); + assert!(is_repetitive_exploration(&samples), "全读轮换游荡应判漂移"); + } + + /// write→grep 交替但 write 各异签名(清理任务红绿迭代)。推演:6 个 write + /// 签名各异全首现 +6,g 首现 +1、再现前均夹各异 write 豁免 +5 → units=12,ratio=1.0; + /// g 段间恒夹写 max_streak=1 → false。 + #[test] + fn scenario_write_grep_alternate_distinct_writes_not_repetitive() { + let g = tool_call_signature("grep", &serde_json::json!({"pattern": "TODO", "path": "src"})); + let samples: Vec<(String, bool)> = (0..6) + .flat_map(|i| { + let w = tool_call_signature("patch_file", + &serde_json::json!({"path": format!("f{}.rs", i), "old": "x", "new": "y"})); + [(w, true), (g.clone(), false)] + }) + .collect(); + assert_eq!(samples.len(), 12); + assert!(!is_repetitive_exploration(&samples), "write 各异+同 grep 验证交替不应判游荡"); + } + // ── is_pure_greeting 纯问候判定测试 ── #[test] diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index 14814c5..efd9656 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -984,7 +984,7 @@ pub(crate) async fn run_agentic_loop( &coord_usage, // Coordinator 路径无 round(未 stream),无估算语义 → false。 false, - None, None, true, + None, None, None, true, &pinned_goals_snapshot, &app_handle, &loop_epoch_arc, my_epoch, @@ -1004,7 +1004,34 @@ pub(crate) async fn run_agentic_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; + // + // acquire 包 30s 超时:阻塞式 acquire 在同 conv 配额被占满时永久 parked(如竞态双 spawn 的 + // 第三个 loop),而 heartbeat 任务已在此前启动持续喂前端看门狗 → UI 永远转圈无终止事件。 + // 超时走 AiError 收敛,范式对齐上方 provider 解析 30s 超时:guard.reset → epoch 检查 → + // 落库 user 消息 → emit → return。 + let _conv_per_conv_permit = match tokio::time::timeout( + std::time::Duration::from_secs(30), + llm_concurrency.acquire_per_conv(&conv_id), + ).await { + Ok(permit) => permit, + Err(_elapsed) => { + guard.reset().await; + // 旧 loop 不 emit(被新 loop 接管时,错误由 owner loop 负责)。 + if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { + return; + } + // 入口超时早退也落库 user 消息(调用方已 push 内存,防重启/切会话丢末条 user)。 + save_conversation(&session_arc, &db, &conv_id, None, None, true).await; + tracing::error!(conv_id = %conv_id, "[ai] 等待 per_conv 并发配额超时(30s),同对话 loop 疑似积压"); + let err_msg = "等待对话并发配额超时(30s),请稍后重试".to_string(); + emit_ai_chat_event(&app_handle, AiChatEvent::AiError { + error: err_msg, + error_type: Some(ErrorType::Unknown), + conversation_id: Some(conv_id.clone()), + }); + return; + } + }; // 防瞬时抖动(G4.1):per-provider 连续饱和计数(跨迭代累计)。 // 候选循环内 Exhausted +1 / Acquired 清零;链首(主 provider)连续 @@ -1014,121 +1041,212 @@ pub(crate) async fn run_agentic_loop( // 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 { - // ── 轮次准备段:入口 stop/stale 收尾闸门 → 主题标记 → conv 一致性(iteration_used/epoch/心跳) - // → 轮通知 → 自动压缩 → 请求消息构建(build_for_request + namespace 展开 + 易变尾段 - // + tool_result 摘要 + 占位断言)→ estimated_prompt + agentic_req 重建 ── - // (实现见下方 prepare_round_inputs,Exit → 本轮终止 return,与原 early-return 等价) - let (messages, estimated_prompt, agentic_req) = match prepare_round_inputs( - &session_arc, &db, &app_handle, &conv_id, iteration, - &stop_flag, &loop_epoch_arc, my_epoch, - &mut guard, &mut tokens, &mut saved_token_snapshot, - last_round_estimated, &provider_config, &llm_concurrency, - &pinned_goals_snapshot, sys_tokens, &system_prompt, &volatile_tail, &intent, - ).await { - PreparedRound::Ready { messages, estimated_prompt, agentic_req } => { - (messages, estimated_prompt, agentic_req) - } - PreparedRound::Exit => return, - }; - // ── 流式消费段:单轮流式调用(candidate 链 + MidStream 保文)── - // 实现见 stream_relay::stream_round。RoundOutcome 四态:Continue 携输出继续; - // StaleAfterStream(流后 epoch 变,guard 已 disarm)/ MidStreamSaved(保文已完整收尾)/ - // Fatal(候选 Fatal 或全耗尽,guard.reset + emit + save 已完成)→ 本轮终止 return。 - let (full_text, tool_calls_acc, round_usage, round_is_estimated) = match stream_round( - &session_arc, &db, &conv_id, - &messages, &tool_defs, &app_handle, - &stop_flag, ¬ify, - max_retries, &model_override, &agentic_req, estimated_prompt, - &mut last_round_estimated, &mut last_reasoning_content, - &mut resolved_model, &mut provider_config, - &mut provider_saturation, - &mut tokens, &mut saved_token_snapshot, &mut guard, - &candidates, &llm_concurrency, - &pinned_goals_snapshot, - &loop_epoch_arc, my_epoch, - ).await { - RoundOutcome::Continue { full_text, tool_calls, round_usage, round_is_estimated } => { - (full_text, tool_calls, round_usage, round_is_estimated) - } - RoundOutcome::StaleAfterStream => return, - RoundOutcome::MidStreamSaved => return, - RoundOutcome::Fatal => return, - }; - // global/per_conv permit 已上移 loop 入口整 loop 持有, - // stream 后不再立即释放(会话级并发语义:工具执行期间也占槽)。per-provider permit - // (_provider_permit)仍在 candidate 循环内随作用域 Drop 自动释放。 + // panic 防护:loop 主体(for + 达上限收尾 + 正常完成收尾)整体包 catch_unwind。 + // 背景:本函数由 IPC 层 tauri::async_runtime::spawn 托管,主体 panic 时任务静默死亡, + // conv 卡 Generating(无终态事件,前端无法恢复)。捕获后经 unwind_to_result 把 panic 转 + // 受控 Err,走既有 teardown_round 正典出口补发终态;ConvState 复位由 guard 的 Drop 兜底 + // 在 unwind 时完成,此处补齐 save 落库与 AiCompleted emit。 + // 借用结构:async 块非 move,guard/tokens/converged 等仍由外层函数持有,unwind 销毁的是 + // 块帧,外层变量完好,panic 分支才能用同一份 guard/tokens 走 teardown_round。 + // 范围:仅 loop 主体;provider 解析/配额获取等已有独立错误分支的段落不包,避免语义重叠。 + // 依赖 panic=unwind(tokio 默认),panic=abort 构建下 catch_unwind 不生效(尽力兜底)。 + let outcome = unwind_to_result(async { + for iteration in start_iteration..effective_max { + // ── 轮次准备段:入口 stop/stale 收尾闸门 → 主题标记 → conv 一致性(iteration_used/epoch/心跳) + // → 轮通知 → 自动压缩 → 请求消息构建(build_for_request + namespace 展开 + 易变尾段 + // + tool_result 摘要 + 占位断言)→ estimated_prompt + agentic_req 重建 ── + // (实现见下方 prepare_round_inputs,Exit → 本轮终止 return,与原 early-return 等价) + let (messages, estimated_prompt, agentic_req) = match prepare_round_inputs( + &session_arc, &db, &app_handle, &conv_id, iteration, + &stop_flag, &loop_epoch_arc, my_epoch, + &mut guard, &mut tokens, &mut saved_token_snapshot, + last_round_estimated, &provider_config, &llm_concurrency, + &pinned_goals_snapshot, sys_tokens, &system_prompt, &volatile_tail, &intent, + ).await { + PreparedRound::Ready { messages, estimated_prompt, agentic_req } => { + (messages, estimated_prompt, agentic_req) + } + PreparedRound::Exit => return, + }; - // ── 轮收尾段:token 累加 + assistant push(+ G1 目标提取)+ 落库 + stop 收尾退出 - // + 收敛/目标验证续轮判定 ── - // (实现见下方 finalize_round_output;Converged → 正常收敛,置 converged 后 break) - match finalize_round_output( - &session_arc, &db, &app_handle, &conv_id, iteration, - &stop_flag, &loop_epoch_arc, my_epoch, - &mut guard, &mut tokens, &mut saved_token_snapshot, - last_round_estimated, &resolved_model, &provider_config, &llm_concurrency, - &pinned_goals_snapshot, - &full_text, &tool_calls_acc, &round_usage, round_is_estimated, - &last_reasoning_content, - &mut goal_verify_rounds, - ).await { - // RunTools → 继续工具调用段(工具调用数由该段内部取,与原时序一致) - RoundEpilogue::RunTools => {} - RoundEpilogue::Converged => { - // 无工具调用且无续轮需求 → 正常收敛退出(原 converged = true; break,语义不变) - converged = true; - break; - } - RoundEpilogue::Exit => return, - }; + // ── 流式消费段:单轮流式调用(candidate 链 + MidStream 保文)── + // 实现见 stream_relay::stream_round。RoundOutcome 四态:Continue 携输出继续; + // StaleAfterStream(流后 epoch 变,guard 已 disarm)/ MidStreamSaved(保文已完整收尾)/ + // Fatal(候选 Fatal 或全耗尽,guard.reset + emit + save 已完成)→ 本轮终止 return。 + let (full_text, tool_calls_acc, round_usage, round_is_estimated) = match stream_round( + &session_arc, &db, &conv_id, + &messages, &tool_defs, &app_handle, + &stop_flag, ¬ify, + max_retries, &model_override, &agentic_req, estimated_prompt, + &mut last_round_estimated, &mut last_reasoning_content, + &mut resolved_model, &mut provider_config, + &mut provider_saturation, + &mut tokens, &mut saved_token_snapshot, &mut guard, + &candidates, &llm_concurrency, + &pinned_goals_snapshot, + &loop_epoch_arc, my_epoch, + ).await { + RoundOutcome::Continue { full_text, tool_calls, round_usage, round_is_estimated } => { + (full_text, tool_calls, round_usage, round_is_estimated) + } + RoundOutcome::StaleAfterStream => return, + RoundOutcome::MidStreamSaved => return, + RoundOutcome::Fatal => return, + }; + // global/per_conv permit 已上移 loop 入口整 loop 持有, + // stream 后不再立即释放(会话级并发语义:工具执行期间也占槽)。per-provider permit + // (_provider_permit)仍在 candidate 循环内随作用域 Drop 自动释放。 - // ── 工具调用段:工具执行(Low 自动 / Medium+High 待审批)+ 三熔断(L1 失败/G2 探索/ - // 写删回环)+ 审批暂停 ── - // (实现见下方 process_round_tools,Exit → 本轮终止 return) - if let ToolsOutcome::Exit = process_round_tools( - &session_arc, &tools_arc, &db, &app_handle, &conv_id, iteration, - &loop_epoch_arc, my_epoch, - &mut guard, &mut tokens, &mut saved_token_snapshot, - &resolved_model, - tool_calls_acc, - &mut fail_window, &mut breaker_warned, - &mut stall_count, &mut stall_warned, &mut loop_count, &mut loop_warned, - ).await { + // ── 轮收尾段:token 累加 + assistant push(+ G1 目标提取)+ 落库 + stop 收尾退出 + // + 收敛/目标验证续轮判定 ── + // (实现见下方 finalize_round_output;Converged → 正常收敛,置 converged 后 break) + match finalize_round_output( + &session_arc, &db, &app_handle, &conv_id, iteration, + &stop_flag, &loop_epoch_arc, my_epoch, + &mut guard, &mut tokens, &mut saved_token_snapshot, + last_round_estimated, &resolved_model, &provider_config, &llm_concurrency, + &pinned_goals_snapshot, + &full_text, &tool_calls_acc, &round_usage, round_is_estimated, + &last_reasoning_content, + &mut goal_verify_rounds, + ).await { + // RunTools → 继续工具调用段(工具调用数由该段内部取,与原时序一致) + RoundEpilogue::RunTools => {} + RoundEpilogue::Converged => { + // 无工具调用且无续轮需求 → 正常收敛退出(原 converged = true; break,语义不变) + converged = true; + break; + } + RoundEpilogue::Exit => return, + }; + + // ── 工具调用段:工具执行(Low 自动 / Medium+High 待审批)+ 三熔断(L1 失败/G2 探索/ + // 写删回环)+ 审批暂停 ── + // (实现见下方 process_round_tools,Exit → 本轮终止 return) + if let ToolsOutcome::Exit = process_round_tools( + &session_arc, &tools_arc, &db, &app_handle, &conv_id, iteration, + &loop_epoch_arc, my_epoch, + &mut guard, &mut tokens, &mut saved_token_snapshot, + &resolved_model, + tool_calls_acc, + &mut fail_window, &mut breaker_warned, + &mut stall_count, &mut stall_warned, &mut loop_count, &mut loop_warned, + ).await { + return; + } + + // 全部自动执行完成 → 继续下一轮 + } + + // 达 MAX 未收敛(LLM 末轮仍想调工具被截断,末轮 tool_result 不再回传 LLM):自动完成收尾。 + // 历史为暂停态(emit AiMaxRoundsReached + 保持 generating=true 仿审批等待,等用户点继续/ + // 停止),因操作卡不可达时 generating 永真无声卡死,已改为 guard 复位 + teardown_round 发 + // AiCompleted(incomplete=true) 自动终止;前端续/停操作卡(pendingMaxRounds)随之成死链。 + // ai_continue_loop/ai_stop_loop IPC 仍保留,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)", + ); + // save_usage 增量;emit_usage 累计快照(前端展示本 loop 总量) + let save_usage = usage_delta_since(&tokens, &mut saved_token_snapshot); + let emit_usage = tokens_snapshot(&tokens); + // P1 bg 化:达轮次上限属需介入(等用户点继续或停止),系统通知补位(条件发送)。 + // 本出口 teardown 通知关(false)防同一事件双弹。 + crate::commands::ai::notify::notify_session_event( + &app_handle, + crate::commands::ai::notify::NotifyKind::NeedAction, + &conv_id, + "已达最大轮次", + "AI 循环未收敛,已达最大轮次,已自动收尾", + ); + // 统一走 teardown_round:save(Some usage, Some model) + 不 spawn_title(对齐原无 title) + + // emit(Some(true), publish=false)。**do_publish=false 保留原 max_iterations 不 publish 行为** + // (与其他 5 路径不一致是历史现状,本次仅收敛重复代码不改 publish 策略,语义零变更)。 + teardown_round( + &session_arc, &db, &conv_id, + Some(&save_usage), Some(&resolved_model), + false, + &provider_config, &llm_concurrency, + &mut guard, + &emit_usage, + last_round_estimated, + Some(true), None, Some("max_rounds"), false, + &pinned_goals_snapshot, + &app_handle, + &loop_epoch_arc, my_epoch, + false, + ).await; return; } - // 全部自动执行完成 → 继续下一轮 - } + // 正常完成 + // F1 并发 epoch:正常完成前若已被新 loop 接管,不 spawn 后台 save/标题/提炼、不 reset/emit + // (owner loop 负责收尾),直接退出。 + if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { + guard.disarm(); + return; + } + // save_usage 增量(自上次 save 后新增,正常收敛时上轮 save 已落,此处通常为 0); + // emit_usage 用 tokens 累计快照。normal_usage.total 直接取 tokens.total() 保持三字段一致。 + let save_usage = usage_delta_since(&tokens, &mut saved_token_snapshot); + { + 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(&save_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; + }); + } - // 达 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)", - ); - // save_usage 增量;emit_usage 累计快照(前端展示本 loop 总量) + // F1 并发 epoch:emit 前 final stale 检查(防上方后台 spawn 块期间被 force_send 接管后仍 emit + // 重复 AiCompleted)。guard.reset 本身 epoch-aware(stale 跳过迁移),此处主要防 emit。 + if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { + guard.disarm(); + return; + } + 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 = tokens_snapshot(&tokens); + // F1 并发 epoch:emit 前最后一次 stale 检查(guard.reset 已幂等,此处防 emit 重复 AiCompleted)。 + if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { + return; + } + emit_ai_completed_once( + &app_handle, &conv_id, &normal_usage, + last_round_estimated, + None, None, None, true, + &pinned_goals_snapshot, + true, + ).await; + }).await; + + // panic 分支:记 error(conv_id + panic 消息,不取全 backtrace),再走与既有异常分支 + // (达 MAX 自动收尾)同构的 teardown_round。emit 字段对齐 try_continue 正典:AiCompleted + // 收敛信号(incomplete 三元 None + do_publish=true + notify=false,非裸 emit),职责是 + // 解卡前端并清占位 tool_result 与审批残留,不代表内容真实完成。旧 loop 被 force_send + // 接管的场景由 teardown_round 内部 epoch 守卫兜住(只 disarm 不 save/emit)。 + if let Err(panic_msg) = outcome { + tracing::error!(conv_id = %conv_id, "[ai] agentic loop panic,走受控收尾: {}", panic_msg); let save_usage = usage_delta_since(&tokens, &mut saved_token_snapshot); let emit_usage = tokens_snapshot(&tokens); - // P1 bg 化:达轮次上限属需介入(等用户点继续或停止),系统通知补位(条件发送)。 - // 本出口 teardown 通知关(false)防同一事件双弹。 - crate::commands::ai::notify::notify_session_event( - &app_handle, - crate::commands::ai::notify::NotifyKind::NeedAction, - &conv_id, - "已达最大轮次", - "AI 循环未收敛已暂停,等待继续或停止指令", - ); - // 统一走 teardown_round:save(Some usage, Some model) + 不 spawn_title(对齐原无 title) + - // emit(Some(true), publish=false)。**do_publish=false 保留原 max_iterations 不 publish 行为** - // (与其他 5 路径不一致是历史现状,本次仅收敛重复代码不改 publish 策略,语义零变更)。 teardown_round( &session_arc, &db, &conv_id, Some(&save_usage), Some(&resolved_model), @@ -1137,66 +1255,35 @@ pub(crate) async fn run_agentic_loop( &mut guard, &emit_usage, last_round_estimated, - Some(true), None, false, + None, None, None, true, &pinned_goals_snapshot, &app_handle, &loop_epoch_arc, my_epoch, false, ).await; - return; } +} - // 正常完成 - // F1 并发 epoch:正常完成前若已被新 loop 接管,不 spawn 后台 save/标题/提炼、不 reset/emit - // (owner loop 负责收尾),直接退出。 - if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { - guard.disarm(); - return; - } - // save_usage 增量(自上次 save 后新增,正常收敛时上轮 save 已落,此处通常为 0); - // emit_usage 用 tokens 累计快照。normal_usage.total 直接取 tokens.total() 保持三字段一致。 - let save_usage = usage_delta_since(&tokens, &mut saved_token_snapshot); - { - 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(&save_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; - }); - } +// panic 载荷转受控错误消息:只认 String 与 &'static str 两种常见载荷,其余类型走兜底文案 +// (不放全 backtrace,定位靠 conv_id + 消息;载荷不可字符串化时保语义不保细节)。 +fn panic_message(payload: &(dyn std::any::Any + Send)) -> String { + payload + .downcast_ref::() + .cloned() + .or_else(|| payload.downcast_ref::<&'static str>().map(|s| s.to_string())) + .unwrap_or_else(|| "agentic loop panic(非字符串载荷)".to_string()) +} - // F1 并发 epoch:emit 前 final stale 检查(防上方后台 spawn 块期间被 force_send 接管后仍 emit - // 重复 AiCompleted)。guard.reset 本身 epoch-aware(stale 跳过迁移),此处主要防 emit。 - if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { - guard.disarm(); - return; +// loop 主体 panic 防护薄包装:Future 正常完成走 Ok,panic 转受控 Err(消息)。 +// 纯函数层无 Tauri/session 依赖,场景回放见 panic_guard_tests;生产侧 run_agentic_loop 把 +// loop 主体 async 块传入,Err 分支走 teardown_round 正典出口。未经过 tokio spawn,不触碰 +// JoinError 语义,外层 spawn 边界行为不变。 +async fn unwind_to_result(fut: F) -> Result { + use std::panic::AssertUnwindSafe; + match futures::FutureExt::catch_unwind(AssertUnwindSafe(fut)).await { + Ok(v) => Ok(v), + Err(p) => Err(panic_message(&*p)), } - 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 = tokens_snapshot(&tokens); - // F1 并发 epoch:emit 前最后一次 stale 检查(guard.reset 已幂等,此处防 emit 重复 AiCompleted)。 - if loop_epoch_arc.load(Ordering::SeqCst) != my_epoch { - return; - } - emit_ai_completed_once( - &app_handle, &conv_id, &normal_usage, - last_round_estimated, - None, None, true, - &pinned_goals_snapshot, - true, - ).await; } @@ -1285,7 +1372,7 @@ async fn prepare_round_inputs( &emit_usage, // 入口 stop:本轮尚未 stream,last_round_estimated 仍为初始 false(无估算)。 last_round_estimated, - None, None, true, + None, None, None, true, pinned_goals_snapshot, app_handle, loop_epoch_arc, my_epoch, @@ -1649,7 +1736,7 @@ async fn finalize_round_output( guard, &emit_usage, last_round_estimated, - None, None, true, + None, None, None, true, pinned_goals_snapshot, app_handle, loop_epoch_arc, my_epoch, @@ -2183,12 +2270,18 @@ async fn fetch_goal_summary(session_arc: &Arc>, conv_id: &str) if goals.is_empty() { return String::new(); } - let goal_summary = goals.iter() - .map(|g| g.text.trim()) + let mut items: Vec = goals.iter() + .map(|g| g.text.trim().to_string()) .filter(|g| !g.is_empty()) - .collect::>() - .join("; "); - format!("(当前目标: {})", goal_summary) + .collect(); + // 超过 3 条截断:全量目标已在 system_prompt「## 当前目标」段锚定,警示里只做提醒, + // 全列会让文案膨胀(实测 5 条目标拼接占满整块警示气泡)。 + if items.len() > 3 { + let total = items.len(); + items.truncate(3); + items.push(format!("…等共 {} 条", total)); + } + format!("(当前目标: {})", items.join("; ")) } // ── insert_tool_call_over_limit_warning: AC-2 ① 单轮工具调用数超限软提示 ── @@ -2263,13 +2356,16 @@ pub(crate) async fn emit_fatal_error(app_handle: &AppHandle, conv_id: &str, erro // emit 单点(teardown_round 各出口 + 正常完成直调),通知收敛此处避免双写; // 用户主动 stop 出口(teardown_round 转发)传 false 关闭。通知内部条件发送 // (notify.rs:主窗可见聚焦不发 + 同类 5 秒节流 + 失败静默)。 -async fn emit_ai_completed_once( +// pub(crate):conv 删除(ai_conversation_delete)在 epoch 递增后单点补发删除终态, +// loop 侧各退出路径因非 owner 静默,删除侧是唯一发出者(调用方无 epoch 守卫,见其注释)。 +pub(crate) async fn emit_ai_completed_once( app_handle: &AppHandle, conv_id: &str, usage: &df_ai::provider::TokenUsage, is_estimated: bool, emit_incomplete: Option, publish_incomplete: Option, + incomplete_reason: Option<&str>, do_publish: bool, pinned_goals: &[super::GoalEntry], notify: bool, @@ -2288,7 +2384,7 @@ async fn emit_ai_completed_once( // publish 端:do_publish=false(max_iterations)跳过 publish(保留原无 publish 行为); // MidStream incomplete 分叉(emit=Some(true)/publish=None)事件值不同,不能并成单次 // emit_ai_chat_event 调用,分路构造(其余 5 路径两端同值,走单次双写)。 - let ev = completed_event(usage, is_estimated, emit_incomplete, conv_id, pinned_goals); + let ev = completed_event(usage, is_estimated, emit_incomplete, incomplete_reason, conv_id, pinned_goals); if do_publish && emit_incomplete == publish_incomplete { emit_ai_chat_event(app_handle, ev); return; @@ -2296,7 +2392,7 @@ async fn emit_ai_completed_once( let _ = app_handle.emit("ai-chat-event", ev); if do_publish { let _ = app_handle.state::().ai_event_bus.publish_event( - completed_event(usage, is_estimated, publish_incomplete, conv_id, pinned_goals), + completed_event(usage, is_estimated, publish_incomplete, None, conv_id, pinned_goals), ); } // P1 bg 化:完成出口系统通知(条件发送,失败静默;stop 出口经调用方传 false 关闭) @@ -2349,10 +2445,13 @@ async fn last_active_assistant_text( } /// 构造 AiCompleted 事件(emit_ai_completed_once 内部去重用,字段口径见其文档)。 +/// incomplete_reason 与 incomplete 配对:emit 侧事件随 emit_incomplete 取值,publish 侧 +/// 分叉事件(与 emit 不一致时)incomplete 为 None,reason 同步 None。 fn completed_event( usage: &df_ai::provider::TokenUsage, is_estimated: bool, incomplete: Option, + incomplete_reason: Option<&str>, conv_id: &str, pinned_goals: &[super::GoalEntry], ) -> AiChatEvent { @@ -2365,6 +2464,7 @@ fn completed_event( reasoning_tokens: usage.reasoning_tokens, is_estimated, incomplete, + incomplete_reason: incomplete_reason.map(str::to_string), conversation_id: Some(conv_id.to_string()), pinned_goals: pinned_goals.to_vec(), } @@ -2599,6 +2699,9 @@ pub(crate) async fn teardown_round( emit_estimated: bool, emit_incomplete: Option, publish_incomplete: Option, + // 不完整原因(emit 侧,与 emit_incomplete 配对):Some("max_rounds")=达最大轮次自动收尾, + // Some("network")=流中途网络中断保文,None=完整/未分类。口径见 AiCompleted 字段文档。 + incomplete_reason: Option<&str>, do_publish: bool, pinned_goals: &[super::GoalEntry], app_handle: &AppHandle, @@ -2629,7 +2732,7 @@ pub(crate) async fn teardown_round( // generating 复位后再 emit Completed(对齐原各路径顺序) emit_ai_completed_once( app_handle, conv_id, emit_usage, emit_estimated, - emit_incomplete, publish_incomplete, do_publish, + emit_incomplete, publish_incomplete, incomplete_reason, do_publish, pinned_goals, notify, ).await; @@ -2810,3 +2913,64 @@ fn build_plan_layers_payload( }) .collect() } + +// panic 防护场景回放:构造 loop 体 panic 的最小等价场景,验证 unwind_to_result 把 panic +// 转为受控 Err、外层状态不随 unwind 丢失、panic 前构造的 RAII 在 unwind 时执行 Drop +// (对应生产侧 GeneratingGuard Drop 复位 ConvState 的兜底)。先于实现落盘,红转绿。 +#[cfg(test)] +mod panic_guard_tests { + use super::*; + + // 正常完成:Ok 直通,包装层不得吞输出或改变返回值 + #[tokio::test] + async fn catch_unwind_ok_path_unchanged() { + let r = unwind_to_result(async { 41 + 1 }).await; + assert_eq!(r, Ok(42)); + } + + // String 载荷 panic 转受控 Err,消息原样保留(生产侧凭此记 tracing::error) + #[tokio::test] + async fn catch_unwind_string_panic_to_controlled_err() { + let r = unwind_to_result(async { panic!("boom {}", 7); }).await; + assert_eq!(r, Err("boom 7".to_string())); + } + + // &'static str 载荷同样收敛 + #[tokio::test] + async fn catch_unwind_str_panic_to_controlled_err() { + let r = unwind_to_result(async { panic!("static boom"); }).await; + assert_eq!(r, Err("static boom".to_string())); + } + + // 非字符串载荷(panic_any)收敛为兜底文案,终态不因载荷类型丢失 + #[tokio::test] + async fn catch_unwind_nonstring_panic_to_fallback_err() { + let r = unwind_to_result(async { std::panic::panic_any(42u32); }).await; + assert!(matches!(r, Err(ref m) if m.contains("panic")), "应走兜底文案: {:?}", r); + } + + // 无泄漏断言:panic 前 mutate 的共享状态对包装层外可见(生产侧对应 panic 分支仍能用 + // 外层 guard/tokens 走 teardown);panic 前构造的 RAII 在 unwind 时执行 Drop + // (对应 GeneratingGuard 在 unwind 路径复位 ConvState 的既有兜底) + #[tokio::test] + async fn catch_unwind_outer_state_survives_and_drop_runs_on_panic() { + struct DropFlag(std::sync::Arc); + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + let outer = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let dropped = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let outer_in_fut = outer.clone(); + let dropped_in_fut = dropped.clone(); + let r = unwind_to_result(async move { + let _raii = DropFlag(dropped_in_fut); + outer_in_fut.store(true, std::sync::atomic::Ordering::SeqCst); + panic!("mid teardown"); + }).await; + assert!(r.is_err()); + assert!(outer.load(std::sync::atomic::Ordering::SeqCst), "panic 前的外层 mutate 应保留"); + assert!(dropped.load(std::sync::atomic::Ordering::SeqCst), "unwind 时 RAII Drop 应执行"); + } +} diff --git a/src-tauri/src/commands/ai/agentic/stream_relay.rs b/src-tauri/src/commands/ai/agentic/stream_relay.rs index dbe6ba4..b021a21 100644 --- a/src-tauri/src/commands/ai/agentic/stream_relay.rs +++ b/src-tauri/src/commands/ai/agentic/stream_relay.rs @@ -565,7 +565,7 @@ pub(crate) async fn stream_round( guard, &emit_usage, *last_round_estimated, - Some(true), None, true, + Some(true), None, Some("network"), true, pinned_goals_snapshot, app_handle, loop_epoch_arc, my_epoch, diff --git a/src-tauri/src/commands/ai/agentic/sub_agent.rs b/src-tauri/src/commands/ai/agentic/sub_agent.rs index 7922756..612c199 100644 --- a/src-tauri/src/commands/ai/agentic/sub_agent.rs +++ b/src-tauri/src/commands/ai/agentic/sub_agent.rs @@ -1,12 +1,23 @@ -//! 子代理循环(spawn_agent v1) — 纯内存子会话 + 拼装既有积木的独立 agentic 循环。 +//! 子代理循环(spawn_agent v1) — 隔离子会话 + 拼装既有积木的独立 agentic 循环。 //! -//! 架构定稿:子代理 = 纯内存子会话,不落库、不进侧栏列表、无标题/知识提炼、无终态事件, -//! 最终轮文本即报告经返回值交付(Claude Code Task 语义同构:隔离上下文、一轮回传报告)。 +//! 架构定稿:子代理 = 隔离子会话,运行期零落库、不进侧栏列表、无标题/知识提炼;统一出口 +//! 整轨迹落库(origin='sub_agent' 标记,列表侧过滤),最终轮文本即报告经返回值交付 +//! (Claude Code Task 语义同构:隔离上下文、一轮回传报告)。 //! -//! ## 零落库机制(单点复用) -//! 子会话 conv_id 启动即写入 `AiSession.deleted_convs`,`save_conversation_inner` 入口检测 -//! 到该集合命中即跳过全部 DB 写入。既有路径(含 stream_round 异常分支内的兜底 save)对该 -//! conv 一律不产生落库,无需给共享链路加子代理分支。 +//! ## 报告协议(S2):拒绝可见性 +//! 工具拒绝维持 fail-closed 就地拒绝(audit/mod.rs deny 单点),但 envelope 携带 +//! tool_name/risk_level/denial_layer 结构化字段;子循环按轮收集被拒清单(解析 envelope), +//! 出口追加「被拒工具」段落到报告尾部 + AiAgentFinished.rejected_tools 透出—— +//! 父会话与用户能看见子代理被拒了什么、哪些子任务因此未完成,而非拿到一份哑报告。 +//! +//! ## 运行期零落库 + 出口整轨迹落库(二期池①,单点复用) +//! 子会话 conv_id 启动即写入 `AiSession.deleted_convs` + `AiSession.conv_origins`, +//! `save_conversation_inner` 入口检测到前者的命中即跳过全部 DB 写入——运行期既有路径 +//! (含 stream_round 异常分支兜底 save)一律不落库,无需给共享链路加子代理分支。 +//! 统一出口(`run_sub_agent_loop`)摘除拦截后调一次 `save_conversation`,整条子轨迹经 +//! 既有 save 链路持久化,origin='sub_agent' 随首插落 `ai_conversations.origin`(V44), +//! 桌面侧栏/miniapp 列表按此过滤。Truncated/Failed 轨迹同样落库(调试与轨迹树数据源); +//! 父删除级联命中(deleted_convs 被级联重打/per_conv 已清)时放弃落库,防空壳复活。 //! //! ## 循环拼装(全部复用既有积木,不另起炉灶) //! - provider 解析:AiProviderRepo.list_all + ProviderPool::select(override 亲和)+ 候选链; @@ -26,7 +37,7 @@ //! 会向子 conv emit AiError、MidStream 分支会触发一次子 conv 的后台标题任务(写库被上者拦, //! 仅一次无效 LLM 调用),均为网络异常罕见路径,接受并记录于此。 -use std::collections::{HashMap, VecDeque}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -47,7 +58,7 @@ use crate::commands::ai::agentic::{ STALL_BREAKER_ENABLED, }; use crate::commands::ai::audit::{process_tool_calls, ToolExecPolicy}; -use crate::commands::ai::conversation::{truncate_for_persist, TokenAccumulator}; +use crate::commands::ai::conversation::{save_conversation, truncate_for_persist, TokenAccumulator}; use crate::commands::ai::event_bus::emit_ai_chat_event; use crate::commands::ai::AiChatEvent; use crate::state::LlmConcurrency; @@ -60,6 +71,10 @@ use super::{ summarize_tool_results, tokens_snapshot, RoundOutcome, }; +/// 子代理会话的 origin 来源标记(落 ai_conversations.origin,V44)。 +/// 桌面侧栏/miniapp 列表按此值过滤子会话;主会话 origin 恒为 NULL 无写入点。 +const SUB_AGENT_ORIGIN: &str = "sub_agent"; + // ============================================================ // 预算常量(三重熔断) // ============================================================ @@ -111,16 +126,20 @@ pub(crate) struct SubAgentOutcome { pub(crate) summary: String, pub(crate) rounds: u32, pub(crate) tokens: u64, + /// 运行期被拒工具汇总(工具名→次数,BTreeMap 序确定性)。S2 报告协议:父会话须能 + /// 看见子代理被拒了什么,而非只拿到一份不知缺口的报告;空 Vec=本次零拒绝。 + pub(crate) rejected_tools: Vec<(String, u32)>, } impl SubAgentOutcome { - /// 启动期失败快捷构造(rounds/tokens 为 0)。 + /// 启动期失败快捷构造(rounds/tokens 为 0,未进循环故无被拒记录)。 pub(crate) fn failed(msg: impl Into) -> Self { Self { status: SubAgentStatus::Failed, summary: msg.into(), rounds: 0, tokens: 0, + rejected_tools: Vec::new(), } } } @@ -262,6 +281,38 @@ pub(crate) fn round_budget_stop( None } +/// 判定 tool_result 内容是否为子代理拒绝 envelope(S2 报告协议解析核心,纯函数可单测)。 +/// 仅识别结构化 JSON 的 status=rejected + reason=sub_agent_denied;非 JSON/正常结果/ +/// 其他拒绝(如主会话 path_blacklist)一律不算,防被拒清单误计。 +pub(crate) fn is_sub_agent_deny_envelope(content: &str) -> bool { + match serde_json::from_str::(content) { + Ok(v) => { + v.get("status").and_then(|s| s.as_str()) == Some("rejected") + && v.get("reason").and_then(|r| r.as_str()) == Some("sub_agent_denied") + } + Err(_) => false, + } +} + +/// 被拒工具清单格式化("工具名×次数")。outcome/事件/报告段三处共用,单一出口。 +pub(crate) fn format_rejected_tools(rejected: &[(String, u32)]) -> Vec { + rejected.iter().map(|(n, c)| format!("{}×{}", n, c)).collect() +} + +/// 报告尾部追加「被拒工具」段落。机制侧兜底:不依赖子 LLM 报告自觉申报 +/// (对齐「机制优先 prompt 说教」),无拒绝原文返回零打扰。纯函数可单测。 +pub(crate) fn append_rejected_section(summary: String, rejected: &[(String, u32)]) -> String { + if rejected.is_empty() { + return summary; + } + let total: u32 = rejected.iter().map(|(_, c)| c).sum(); + let list = format_rejected_tools(rejected).join(", "); + format!( + "{}\n\n---\n被拒工具(共 {} 次): {}\n以上工具在子代理中无执行权限,相关子任务可能未完成,请父会话评估是否接手。", + summary, total, list + ) +} + /// 子代理专用 system prompt(要点:面向父任务执行、最终回复即结论报告、无审批通道)。 pub(crate) fn sub_agent_system_prompt() -> String { "你是 DevFlow 主会话派生的子代理,在隔离上下文中独立执行父任务交办的任务书。\n\ @@ -273,7 +324,8 @@ pub(crate) fn sub_agent_system_prompt() -> String { - 控制轮次:默认最多 6 轮,收敛即停,不做任务书之外的无关探索\n\ ## 输出要求\n\ - 最终回复就是交付给父任务的结论报告,不要寒暄、不要向用户提问\n\ - - 报告结构:结论先行,然后列关键发现/证据/数据,最后列未尽事项与建议" + - 报告结构:结论先行,然后列关键发现/证据/数据,最后列未尽事项与建议\n\ + - 若有工具调用被拒绝:在未尽事项中说明哪些子任务因工具权限受限未完成,父会话据此决定接手方式" .to_string() } @@ -301,7 +353,7 @@ pub(crate) async fn run_sub_agent_loop( llm_concurrency: LlmConcurrency, ) -> SubAgentOutcome { let outcome = run_sub_agent_loop_inner( - session_arc.clone(), tools_arc, db, app_handle.clone(), + session_arc.clone(), tools_arc, db.clone(), app_handle.clone(), parent_conv_id.clone(), child_conv_id.clone(), system_prompt, task_prompt, tool_defs, max_rounds, model_override, llm_concurrency.clone(), @@ -315,13 +367,36 @@ pub(crate) async fn run_sub_agent_loop( status: outcome.status.as_str().to_string(), rounds: outcome.rounds, tokens: outcome.tokens, + // S2 报告协议:被拒清单透出(空则 None,事件面最小噪音);前端不消费也不破坏 + rejected_tools: if outcome.rejected_tools.is_empty() { + None + } else { + Some(format_rejected_tools(&outcome.rejected_tools)) + }, }); - // 清理子会话内存条目(报告已提取,纯内存子会话不留残迹;审计表记录保留可追溯) + // 子轨迹落库(二期池①):摘除零落库拦截,整条轨迹经既有 save 链路持久化。 + // 级联取消/父删除已打 deleted_convs 标记或清 per_conv 时放弃落库(save 的惰性 + // 重建会 INSERT 空壳复活);存活则 origin='sub_agent' 随首插落库,列表侧按此过滤。 + let child_persistable = { + let mut session = session_arc.lock().await; + let alive = !session.deleted_convs.contains(&child_conv_id) + && session.per_conv.contains_key(&child_conv_id); + if alive { + session.deleted_convs.remove(&child_conv_id); + } + alive + }; + if child_persistable { + save_conversation(&session_arc, &db, &child_conv_id, None, None, false).await; + } + + // 清理子会话内存条目(轨迹已持久化或已随级联丢弃;origin 标记随 per_conv 一并回收) { let mut session = session_arc.lock().await; session.per_conv.remove(&child_conv_id); session.save_locks.remove(&child_conv_id); + session.conv_origins.remove(&child_conv_id); } llm_concurrency.release_conv(&child_conv_id).await; @@ -369,10 +444,12 @@ async fn run_sub_agent_loop_inner( if !session.per_conv.contains_key(&parent_conv_id) { return SubAgentOutcome::failed("子代理启动失败:父会话已不存在(可能已被删除)"); } - // 零落库(关键):deleted_convs 使 save_conversation 入口对该 conv 一律跳过, - // 覆盖 stream_round 异常分支兜底 save 等所有既有写库路径(见模块头文档)。 + // 运行期零落库(关键):deleted_convs 使 save_conversation 入口对该 conv 一律跳过, + // 覆盖 stream_round 异常分支兜底 save 等所有既有写库路径(见模块头文档); + // conv_origins 记来源,统一出口摘除拦截落库时随首插写 ai_conversations.origin。 // 先于 conv() 调用(insert 需可变借 session,避免与 conv 借用冲突) session.deleted_convs.insert(child_conv_id.clone()); + session.conv_origins.insert(child_conv_id.clone(), SUB_AGENT_ORIGIN.to_string()); let conv = session.conv(&child_conv_id); conv.messages.push(ChatMessage::user(&task_prompt)); conv.stop_flag.store(false, Ordering::SeqCst); @@ -442,6 +519,8 @@ async fn run_sub_agent_loop_inner( let mut status = SubAgentStatus::Truncated; let mut stop_reason = String::new(); let mut report = String::new(); + // S2 报告协议:被拒工具累计(工具名→次数,BTreeMap 序确定性),出口汇总进报告尾+outcome + let mut rejected_tools: BTreeMap = BTreeMap::new(); for round in 0..max_rounds { // ── 停止检查点 + 预算检查(均在 prepare 段前)── @@ -547,10 +626,17 @@ async fn run_sub_agent_loop_inner( break; } // ── 工具执行(SubAgent 策略:NeedsAuth/中高风险自动 deny,零待审批挂起)── + // 先快照本轮 (tool_call_id, 工具名):process 按值消费 tool_calls,之后取不到 + let round_calls: Vec<(String, String)> = tool_calls + .values() + .map(|d| (d.id.clone(), d.name.clone())) + .collect(); let pending = process_tool_calls( &session_arc, tool_calls, &tools_arc, &db, &app_handle, &child_conv_id, ToolExecPolicy::SubAgent, ).await; + // S2 报告协议:收集本轮被拒清单(deny 单点已写入结构化 envelope,此处只读解析) + collect_rejected_tools(&session_arc, &child_conv_id, &round_calls, &mut rejected_tools).await; if pending > 0 { // 防御:SubAgent 策略下不应产生挂起;出现即终止(子代理无审批恢复通道) status = SubAgentStatus::Failed; @@ -626,6 +712,11 @@ async fn run_sub_agent_loop_inner( } }; + // S2 报告协议:报告尾部追加「被拒工具」段落(机制兜底,不依赖子 LLM 自觉申报); + // 清单同时随 outcome 交付父 LLM,并经 AiAgentFinished.rejected_tools 透出前端。 + let rejected: Vec<(String, u32)> = rejected_tools.into_iter().collect(); + let summary = append_rejected_section(summary, &rejected); + tracing::info!( parent = %parent_conv_id, agent = %child_conv_id, @@ -642,6 +733,7 @@ async fn run_sub_agent_loop_inner( summary, rounds: rounds_done, tokens: total_tokens, + rejected_tools: rejected, } } @@ -678,6 +770,37 @@ async fn last_assistant_text( .filter(|c| !c.trim().is_empty()) } +/// 收集本轮被拒工具调用(S2 报告协议数据源):按本轮 tool_call id 匹配消息流中的 +/// tool_result,解析结构化 envelope 判定子代理拒绝,按工具名累计次数。deny 单点 +/// 已把 envelope 写入消息流,此处只读解析,不反向侵入 audit 层接口。短锁段纯读。 +async fn collect_rejected_tools( + session_arc: &Arc>, + child_id: &str, + round_calls: &[(String, String)], // (tool_call_id, tool_name),本轮快照 + rejected: &mut BTreeMap, +) { + if round_calls.is_empty() { + return; + } + let session = session_arc.lock().await; + let Some(conv) = session.conv_read(child_id) else { + return; + }; + for m in conv.messages.all_messages_clone() { + if !matches!(m.role, MessageRole::Tool) { + continue; + } + let Some(tc_id) = m.tool_call_id.as_deref() else { + continue; + }; + if let Some((_, name)) = round_calls.iter().find(|(id, _)| id == tc_id) { + if is_sub_agent_deny_envelope(&m.content) { + *rejected.entry(name.clone()).or_insert(0) += 1; + } + } + } +} + // ============================================================ // 单元测试:白名单钳制 / max_rounds 熔断 / registry 级联注册移除 // ============================================================ @@ -797,5 +920,44 @@ mod tests { assert!(p.contains("子代理")); assert!(p.contains("最终回复")); assert!(p.contains("审批通道"), "应说明无审批通道(防挂死设计的前提告知)"); + assert!(p.contains("工具权限受限"), "S2 报告协议:应要求申报因工具受限未完成的子任务"); + } + + /// S2 报告协议:deny envelope 识别——结构化 JSON 判定,非 JSON/正常结果/其他拒绝不误计。 + #[test] + fn deny_envelope_recognition() { + let env = serde_json::json!({ + "status": "rejected", + "reason": "sub_agent_denied", + "tool_name": "write_file", + "risk_level": "medium", + "denial_layer": "medium-high-policy", + "message": "子代理无权执行该工具" + }).to_string(); + assert!(is_sub_agent_deny_envelope(&env)); + // 工具正常成功结果(JSON 但非拒绝 envelope)不误计 + assert!(!is_sub_agent_deny_envelope(r#"{"status":"ok","data":1}"#)); + // 裸文本输出(老格式/普通内容)不误计 + assert!(!is_sub_agent_deny_envelope("普通工具输出文本")); + // 其他 reason 的拒绝(如路径黑名单)不算子代理策略拒绝 + let other = serde_json::json!({"status":"rejected","reason":"path_blacklist"}).to_string(); + assert!(!is_sub_agent_deny_envelope(&other)); + } + + /// S2 报告协议:被拒段落——有拒绝才追加(总次数/逐项计数/接手提示),无拒绝零打扰。 + #[test] + fn rejected_section_appended_only_when_nonempty() { + let base = "结论:已完成检索".to_string(); + assert_eq!(append_rejected_section(base.clone(), &[]), base, "无拒绝应原文返回"); + let rejected = vec![("write_file".to_string(), 3u32), ("run_command".to_string(), 1u32)]; + let out = append_rejected_section(base, &rejected); + assert!(out.starts_with("结论:已完成检索"), "报告本体须在前: {}", out); + assert!(out.contains("被拒工具(共 4 次)"), "总次数统计: {}", out); + assert!(out.contains("write_file×3") && out.contains("run_command×1"), "逐项计数: {}", out); + assert!(out.contains("父会话"), "接手提示: {}", out); + assert_eq!( + format_rejected_tools(&rejected), + vec!["write_file×3".to_string(), "run_command×1".to_string()] + ); } } diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 9e08203..18145e3 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -114,20 +114,40 @@ pub(crate) enum ToolExecPolicy { /// SubAgent 策略的就地拒绝单点:失败 tool_result(结构化 envelope,断路器可判)+ /// emit AiToolCallCompleted + 审计留痕(decided_by=sub_agent_denied)。 /// 不 insert pending_approvals(子代理无审批通道,挂起即永久停摆)。 +/// +/// S2 拒绝可见性:envelope 增 tool_name/risk_level/denial_layer 结构化字段——拒绝不只 +/// 「说不」,子 LLM 须能从结构化信息推断可改尝试方向(换 Low 工具/换路径/转交主会话)。 +/// denial_layer 区分四层拒绝源,各层行动提示不同: +/// - whitelist-clamp:工具不在白名单(spawn 入口裁剪或幻觉调用) +/// - needs-auth:路径待授权(主会话可授权放行) +/// - path-blacklist:系统黑名单(主会话也放不了行,从 needs-auth 拆出防误导授权尝试) +/// - medium-high-policy:中高风险策略(拆低风险步骤或转交主会话审批) async fn deny_sub_agent_tool( session_arc: &Arc>, conv_id: &str, draft: &ToolCallDraft, tools_arc: &Arc, audit_repo: &AiToolExecutionRepo, + layer: &str, reason: &str, current_message_id: Option<&str>, app_handle: &AppHandle, ) { + // 各拒绝层的可改尝试方向(行动提示,让子 LLM 下一步有路可走而非反复重试同一拒绝) + let hint = match layer { + "whitelist-clamp" => "该工具不在你的白名单(spawn 时被裁剪或不存在),改用白名单内的低风险工具完成等价操作", + "needs-auth" => "该路径主会话可经授权放行,在报告未尽事项中说明,由主会话授权后执行", + "path-blacklist" => "该路径系统级禁用,主会话授权也无法放行,须改用其他路径或放弃该操作", + _ => "中高风险工具子代理不可执行,可拆解为低风险步骤,或在报告未尽事项中说明由主会话审批执行", + }; + let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High); let msg = serde_json::json!({ "status": "rejected", "reason": "sub_agent_denied", - "message": format!("子代理无权执行该工具({}):仅低风险只读且路径已授权的工具可在子代理中自动执行,如确需该操作请写入报告交由主会话完成", reason) + "tool_name": draft.name, + "risk_level": risk_level, + "denial_layer": layer, + "message": format!("子代理无权执行该工具({}):仅低风险只读且路径已授权的工具可在子代理中自动执行。{}", reason, hint) }).to_string(); // 短 lock 段:push tool_result(纯写,无 await) { @@ -142,7 +162,6 @@ async fn deny_sub_agent_tool( conversation_id: Some(conv_id.to_string()), }; emit_ai_chat_event(app_handle, ev); - let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High); audit_tool_call(audit_repo, conv_id, &draft.id, &draft.name, &draft.args, "rejected", risk_level, Some(msg), Some("sub_agent_denied"), current_message_id).await; } @@ -157,7 +176,12 @@ async fn deny_sub_agent_tool( /// 全盘 search 等开发长命令——前端 watchdog 误判断流，抛"工具已执行完成后续中断"误报 /// （实测：用户报"一边流一边抛"，前一轮 delta 文本在屏 + 当前轮工具执行静默 > 130s）。 /// 补工具执行阶段的心跳缺口，前端 watchdog 在静默期也能收到 reset，不再误杀长工具。 -async fn execute_with_heartbeat( +/// +/// 审批执行窗口(ai_approve / ai_authorize_dir)复用本函数:审批暂停时主 loop 已 return、 +/// 其 heartbeat_loop 随作用域退出,慢工具(大文件 patch / spawn_agent / run_command 长构建) +/// 整窗无心跳 → 前端看门狗误杀。复用同时获得 run_command 实时流式 sink、spawn_agent 父会话 +/// task-local 注入与 outer_secs 分档超时(spawn_agent=挂钟预算+60 / run_command=timeout_secs clamp)。 +pub(crate) async fn execute_with_heartbeat( tools: &AiToolRegistry, name: &str, args: serde_json::Value, @@ -242,9 +266,44 @@ async fn execute_with_heartbeat( conv_id.to_string(), exec_fut, ); - match tokio::time::timeout(Duration::from_secs(outer_secs), exec_fut).await { - Ok(result) => result, - Err(_elapsed) => { + + // ── 提前退出监视(防空转):会话已删 / 审批窗口失效时中止工具执行 ── + // 进入快照:该 tool_call_id 是否仍在 pending_approvals。现网 4 个调用点(chat.rs ai_approve / + // ai_authorize_dir 两个审批执行窗口均「先 remove 后 execute」,本文件 low/trust 两条自动执行 + // 路径从未 insert),进入时 pending 必无此 id——若按字面「不在 pending 即退出」实现,会把 + // 所有已批准执行立即腰斩。故审批移除监视仅在「进入时仍在 pending」的调用形态下武装 + // (现网不触发,为未来调用形态兜底);会话已删监视对所有调用点生效:执行中删除会话, + // 长工具最多再跑 2s 即中止,不再空转到 outer_secs 上限。 + let tc_watch = tool_call_id.to_string(); + let was_pending_at_entry = { + let st = app.state::(); + let session = st.ai_session.lock().await; + session.pending_approvals.contains_key(&tc_watch) + }; + let watch_app = app.clone(); + let watch_conv = conv_id.to_string(); + let abort_watch = async move { + loop { + // 2s 轮询:短锁纯读(锁不跨 await 持有);中止延迟 ≤2s 与锁竞争频率的权衡 + tokio::time::sleep(Duration::from_secs(2)).await; + let st = watch_app.state::(); + let session = st.ai_session.lock().await; + if session.deleted_convs.contains(&watch_conv) { + return format!("会话 {} 已删除", watch_conv); + } + if was_pending_at_entry && !session.pending_approvals.contains_key(&tc_watch) { + return format!("审批 {} 已不在挂起表", tc_watch); + } + } + }; + + tokio::pin!(exec_fut); + // 三路 select:exec 正常完成 / outer_secs 兜底超时(既有语义,drop exec 腰斩) / + // 监视判死提前退出(同样 drop exec,防审批被删或会话已删后空转到超时上限)。 + // exec 先完成时 select 丢弃 abort_watch,监视零泄漏(非 spawn,随 select 作用域消亡)。 + tokio::select! { + result = &mut exec_fut => result, + _ = tokio::time::sleep(Duration::from_secs(outer_secs)) => { tracing::error!( conv_id = %conv_id, tool = %name, @@ -252,11 +311,178 @@ async fn execute_with_heartbeat( "[ai] 工具执行超时({}s),返回错误 tool_result(防 process_tool_calls 持 session lock 永久卡死)", outer_secs ); - Err(anyhow::anyhow!("工具执行超时({}s),已中止(防死锁)", outer_secs)) + Err(anyhow::anyhow!("工具 {} 执行超时({}s),已中止(防死锁)", name, outer_secs)) + } + reason = abort_watch => { + tracing::warn!( + conv_id = %conv_id, + tool = %name, + tool_call_id = %tool_call_id, + "[ai] execute_with_heartbeat 提前退出: {reason}(防空转)" + ); + Err(anyhow::anyhow!("工具 {} 提前中止: {}", name, reason)) } } } +// ============================================================ +// 审批超时主动收敛(定时器) — 治「审批过期后用户不操作则永挂」 +// ============================================================ +// +// 原状:cleanup_expired_approvals(私有于 agentic 模块)仅在 try_continue_agent_loop 入口惰性 +// 触发,用户不操作则审批永挂(前端无终态、审批弹窗吊死)。此处补登记侧定时器:审批插入后 +// (process_tool_calls 收尾统一扫描武装)spawn tokio 定时任务,到点主动执行与 +// cleanup_expired_approvals 语义一致的取消动作 + emit AiApprovalResult 终态。 +// +// 模块可见性说明:approval_timeout 模块私有于 agentic(纯 `mod` 声明),audit 不可达; +// 文件约束(不动 agentic/mod.rs,他代理在改)下定时器与到点收敛落调用侧, +// cancel_expired_approval 语义镜像 approval_timeout::cleanup_expired_approvals, +// 两端需同步维护(改超时取消语义时两处同改)。 + +/// 已武装超时定时器的审批 tool_call_id 集合(进程级,防重复 spawn)。 +/// 挂起审批在 pending 期间会被每轮 process_tool_calls 的收尾扫描反复见到(启动恢复的积压审批 +/// 尤其跨多轮存留),靠此集合保证同 id 只 spawn 一次。只增不清:tool_call_id 一轮唯一, +/// 审批移除后残留条目仅少量内存无行为影响,进程退出即释放。 +static ARMED_TIMEOUT_IDS: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + +/// 为当前所有待审批武装「到点主动收敛」定时任务。 +/// +/// 幂等:ARMED_TIMEOUT_IDS 去重,同一 tool_call_id 只 spawn 一次。 +/// 全量扫描(不按 conv 过滤):启动恢复的积压审批也能在任意一轮 process_tool_calls 被补武装。 +/// 跳过:conversation_id=None(无主审批,cleanup 按 conv 路由无法定位)与 created_at=None +/// (异常数据,超时判定恒 false,定时任务纯浪费)。 +pub(crate) async fn arm_approval_timeout_timers(app: &AppHandle) { + let timeout_minutes = app + .state::() + .approval_timeout_minutes + .load(std::sync::atomic::Ordering::SeqCst); + if timeout_minutes == 0 { + return; // 0=禁用超时:不武装,与 cleanup_expired_approvals 禁用语义一致 + } + let timeout_dur = std::time::Duration::from_secs(timeout_minutes * 60); + + // 短锁快照 + 标武装(锁内纯读写,无 await 持锁跨 Await) + let to_arm: Vec<(String, String, std::time::SystemTime)> = { + let state = app.state::(); + let session = state.ai_session.lock().await; + let mut armed = ARMED_TIMEOUT_IDS.get_or_init(|| std::sync::Mutex::new(HashSet::new())) + .lock() + .unwrap_or_else(|e| e.into_inner()); + session + .pending_approvals + .values() + .filter_map(|a| { + let cid = a.conversation_id.as_deref()?; + let created = a.created_at?; + // insert 返 false = 已武装过,跳过(不重复 spawn) + if !armed.insert(a.tool_call_id.clone()) { + return None; + } + Some((cid.to_string(), a.tool_call_id.clone(), created)) + }) + .collect() + }; + + for (conv_id, tc_id, created) in to_arm { + let app_task = app.clone(); + tokio::spawn(async move { + // 首睡 = 登记时点距过期的差额 + 2s 缓冲(created_at 读取早于 spawn, + // 补缓冲保到点时 elapsed 判定必过线,免得到点差几毫秒被判未过期而空转)。 + let mut sleep_for = timeout_dur + .saturating_sub(created.elapsed().unwrap_or_default()) + + std::time::Duration::from_secs(2); + loop { + tokio::time::sleep(sleep_for).await; + let state = app_task.state::(); + // 幂等/防泄漏闸门(短锁纯读): + // 会话已删或 per_conv 已不在 → 静默退场,不复活会话、不 emit 幽灵终态; + // 审批已不在 pending(用户已处理/他路已清理)→ 终态已发,静默返回防重复 emit。 + { + let session = state.ai_session.lock().await; + if session.deleted_convs.contains(&conv_id) + || !session.per_conv.contains_key(&conv_id) + { + return; + } + if !session.pending_approvals.contains_key(&tc_id) { + return; + } + } + // 到点精判读最新配置:兼容运行中调大超时(未过期则补睡差额重查;调小/不变必已过期) + let cur_minutes = state + .approval_timeout_minutes + .load(std::sync::atomic::Ordering::SeqCst); + if cur_minutes == 0 { + return; // 中途禁用超时:放弃主动收敛(禁用语义=永不超时) + } + let created_at = { + let session = state.ai_session.lock().await; + session.pending_approvals.get(&tc_id).and_then(|a| a.created_at) + }; + let Some(created_at) = created_at else { + return; // 异常:条目存在却无 created_at,超时判定恒 false,退场交回惰性路径 + }; + let elapsed = std::time::SystemTime::now() + .duration_since(created_at) + .unwrap_or_default(); + let cur_dur = std::time::Duration::from_secs(cur_minutes * 60); + if elapsed >= cur_dur { + cancel_expired_approval(&app_task, state.inner(), &conv_id, &tc_id, cur_minutes).await; + return; + } + sleep_for = cur_dur - elapsed + std::time::Duration::from_secs(2); + } + }); + } +} + +/// 到点取消单个超时审批:语义镜像 approval_timeout::cleanup_expired_approvals 的单条处理路径 +/// (移除 pending → 替换占位 tool_result → emit AiApprovalResult 终态 → 审计 rejected → 落库)。 +/// 入参均为已判定过期(调用方闸门后),失败不抛错仅随各步内部降级,避免影响其他审批。 +async fn cancel_expired_approval( + app: &AppHandle, + state: &crate::state::AppState, + conv_id: &str, + tc_id: &str, + timeout_minutes: u64, +) { + tracing::info!( + conv_id = %conv_id, + tool_call_id = %tc_id, + timeout_minutes, + "[APPROVAL-TIMEOUT] 定时器到点,主动取消挂起审批" + ); + // 1+2. 短锁:从 pending 移除 + 替换占位 tool_result(让 LLM 看到回执不再盲猜) + { + let mut session = state.ai_session.lock().await; + session.pending_approvals.remove(tc_id); + session + .conv(conv_id) + .messages + .replace_tool_result_content(tc_id, &format!("审批超时({}分钟)已自动取消,请改换其他方案", timeout_minutes)); + } + // 3. emit AiApprovalResult(前端清除 pending UI,emit+publish 双写经正典 helper) + emit_ai_chat_event( + app, + AiChatEvent::AiApprovalResult { + id: tc_id.to_string(), + approved: false, + conversation_id: Some(conv_id.to_string()), + }, + ); + // 4. 审计落 rejected(审计表有终态,ai_approve 幂等分支可据此收敛前端残留) + audit_finalize( + state, + tc_id, + "rejected", + Some(format!("审批超时({}分钟)自动取消", timeout_minutes)), + ) + .await; + // 5. 落库会话状态变更(占位消息替换后需持久化,防重启后丢失) + super::conversation::save_conversation(&state.ai_session, &state.db, conv_id, None, None, true).await; +} + pub(crate) async fn process_tool_calls( session_arc: &Arc>, tool_calls_acc: HashMap, @@ -400,8 +626,10 @@ pub(crate) async fn process_tool_calls( // 收集 NeedsAuth/Denied drafts（从 drafts 移除，不进下方 RiskLevel 循环） let mut path_auth_pending: Vec<(ToolCallDraft, serde_json::Value, PathAuthRequest)> = Vec::new(); let mut path_denied: Vec<(ToolCallDraft, String)> = Vec::new(); - // SubAgent 策略:须授权/黑名单拒绝统一收集,后续就地拒绝(绝不挂起) - let mut sub_agent_path_denied: Vec<(ToolCallDraft, String)> = Vec::new(); + // SubAgent 策略:须授权/黑名单拒绝统一收集,后续就地拒绝(绝不挂起)。 + // 第三元=denial_layer:needs-auth(待授权)与 path-blacklist(系统黑名单)分开标注—— + // 黑名单主会话也放不了行,混标会误导子 LLM 走授权通道。 + let mut sub_agent_path_denied: Vec<(ToolCallDraft, String, &'static str)> = Vec::new(); let mut authorized_drafts: Vec<(u32, ToolCallDraft, serde_json::Value)> = Vec::new(); for (idx, draft, args) in drafts { // 会话级临时授权目录在 allowed_guard.session 内（进程级，随 active 会话切换清空）， @@ -411,7 +639,7 @@ pub(crate) async fn process_tool_calls( FileToolAuthOutcome::NeedsAuth(req) => { // SubAgent 策略:子代理无授权弹窗通道,挂起即永久停摆 → 就地拒绝。 if is_sub_agent { - sub_agent_path_denied.push((draft, "路径未授权且子代理无授权弹窗通道".to_string())); + sub_agent_path_denied.push((draft, "路径未授权且子代理无授权弹窗通道".to_string(), "needs-auth")); continue; } // all 档全自动(用户拍板):路径授权弹窗也不打断,与 RiskLevel 门同语义。 @@ -425,7 +653,7 @@ pub(crate) async fn process_tool_calls( } FileToolAuthOutcome::Denied(reason) => { if is_sub_agent { - sub_agent_path_denied.push((draft, reason)); + sub_agent_path_denied.push((draft, reason, "path-blacklist")); } else { path_denied.push((draft, reason)); } @@ -467,10 +695,10 @@ pub(crate) async fn process_tool_calls( } // SubAgent 策略:须授权/黑名单拒绝就地产出失败 tool_result(零挂起,防子循环永久停摆)。 - for (draft, reason) in sub_agent_path_denied { + for (draft, reason, layer) in sub_agent_path_denied { deny_sub_agent_tool( session_arc, conv_id, &draft, tools_arc, &audit_repo, - &reason, current_message_id, app_handle, + layer, &reason, current_message_id, app_handle, ).await; } @@ -588,9 +816,12 @@ pub(crate) async fn process_tool_calls( // SubAgent 策略:Medium/High 就地拒绝(无审批通道,挂起即子循环永久停摆)。 // 白名单钳制已使子代理 LLM 只见 Low 工具定义,此处拦的是幻觉调用非白名单工具的残面。 if is_sub_agent && risk_level != RiskLevel::Low { + // 已注册工具=中高风险策略拒绝;未注册名(risk 兜底 High)=白名单外幻觉调用, + // 标 whitelist-clamp 让子 LLM 知道该工具不在其可见定义里,而非风险不让用。 + let layer = if tools_arc.get(&draft.name).is_some() { "medium-high-policy" } else { "whitelist-clamp" }; deny_sub_agent_tool( session_arc, conv_id, &draft, tools_arc, &audit_repo, - "工具风险等级超出子代理权限", current_message_id, app_handle, + layer, "工具风险等级超出子代理权限", current_message_id, app_handle, ).await; continue; } @@ -611,6 +842,13 @@ pub(crate) async fn process_tool_calls( ).await; } + // ── 审批超时主动收敛:为本轮登记(含历史遗留)的挂起武装到点定时任务 ── + // 本函数是审批插入的唯一漏斗(路径授权挂起 + Med/High 风险审批都在上方循环落 pending), + // 收尾统一扫描武装,覆盖两类登记。惰性清理只在用户下次交互时生效,用户不操作则审批 + // 永挂(前端无终态、弹窗吊死);到点由定时任务主动 emit AiApprovalResult 终态收敛前端。 + // 内部幂等(armed 集合去重 + 会话已删/审批已处理静默跳过),重复调用无副作用。 + arm_approval_timeout_timers(app_handle).await; + // trust-hit 并行执行：execute + 即时 emit 在闭包内（闭包不访问 session, // 调用方不再持锁,本段所有 await 完全在锁外）, // push tool_result / audit 在 join_all 后串行回填（短 lock push + 锁外 audit）。对齐 Low risk 并行模式。 diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 3ad128e..20f55a6 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -34,7 +34,7 @@ use super::super::agentic::GOAL_MAX_CHARS; // ConvState 枚举变体引用(Idle 等),用于 B-Phase2 写侧 conv_states.transition 调用。 // 读侧拦截(can_accept_request)走 state.conv_states 方法,不直接引用 ConvState 类型。 use super::super::agentic::conv_state::ConvState; -use super::super::audit::{audit_finalize, emit_data_changed}; +use super::super::audit::{audit_finalize, emit_data_changed, execute_with_heartbeat}; use super::super::augmentation::build_augmentation_segment; use super::super::augmentation::sanitize; use super::super::conversation::save_conversation; @@ -806,8 +806,11 @@ pub(crate) async fn spawn_agentic_session( } else { conv.messages.push(ChatMessage::user(&user_content)); } - // 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级) - conv.working_context.begin_turn(&user_content, 200); + // 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级)。 + // 喂原文 message 而非落库变换文本 user_content:begin_turn 契约是"本条用户请求", + // 降级提示尾缀与 /{skill} 空文本标记是落库与标题素材专用,当目标会污染钉扎文本; + // 纯技能调用原文为空,靠 begin_turn 自身的空文本跳过语义不记目标。 + conv.working_context.begin_turn(&message, 200); // F-260619-04 P1:push 后取末条 user 消息 id(本轮 user,原供知识注入 referenced 溯源)。 // DRY(B):知识注入已收敛至 inject_knowledge_into_prompt(helper 内部同消息取 text+id), // 此处 user_msg_id 不再透传到注入逻辑,保留下划线占用(锁内 push 已发生,语义不变)。 @@ -1093,43 +1096,16 @@ pub async fn ai_approve( let exec_result: anyhow::Result = if approval.tool_name == "run_workflow" { super::super::execute_run_workflow_for_tool(&app, &state, &args, conv_id.as_deref()).await } else { - // F-260620 对齐 ai_authorize_dir(:658):ai_approve 执行同样包 60s 超时。 - // 防 list_directory 大目录/run_command 慢命令卡死 → 到不了 audit_finalize/emit/try_continue - // → IPC 永挂 → per_conv.generating 永真 → 前端 130s 看门狗静默吞消息(F-260620 同型故障, - // 之前只给 ai_authorize_dir 加超时,ai_approve 漏)。 - // run_command 实时流式:经 task-local sink 注入(AppHandle + tool_call_id + conv_id), - // handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。非 run_command 不注入。 - let sink = if approval.tool_name == "run_command" { - Some(super::super::command_stream::CommandSink::new( - app.clone(), id.clone(), conv_id.clone(), - )) - } else { None }; - let exec_fut = super::super::command_stream::scope(sink, state.ai_tools.execute(&approval.tool_name, args.clone())); - // spawn_agent 父会话定位:审批执行路径同样经 task-local 注入父 conv - // (handler 兜底 active_conversation_id 不可靠:审批可能发生在用户已切会话后)。 - let exec_fut = super::super::tools::spawn_agent::scope_caller_conv( - conv_id.clone().unwrap_or_default(), - exec_fut, - ); - // spawn_agent 外层超时对齐其挂钟预算加缓冲:子循环自身三重熔断先触发, - // 以截断报告优雅收尾,不被 60s 裸超时腰斩(报告蒸发)。 - let approve_outer_secs: u64 = if approval.tool_name == "spawn_agent" { - crate::commands::ai::agentic::sub_agent::SUB_AGENT_WALL_TIMEOUT_SECS + 60 - } else { - 60 - }; - match tokio::time::timeout( - std::time::Duration::from_secs(approve_outer_secs), - exec_fut, - ).await { - Ok(r) => r, - // 任务2: ai_approve 路径同步 run_command 失败提示(超时为最常见失败场景)。 - // 仅对 run_command 追加(其余工具无 shell 适配问题),避免污染其他工具错误语义。 - Err(_) => Err(anyhow::anyhow!("工具执行超时({}s): {}{}", approve_outer_secs, approval.tool_name, - if approval.tool_name == "run_command" && cfg!(windows) { - "\n提示: PowerShell 下路径用正斜杠或双引号包裹,命令间用 `;` 而非 `&&`(PS5)" - } else { "" })), - } + // 审批执行窗口复用 execute_with_heartbeat:审批暂停时主 loop 已 return、其 heartbeat_loop + // 随作用域退出,慢工具(大文件 patch / spawn_agent / run_command 长构建)整窗无心跳 + // → 前端 130s 看门狗误判断流。复用同时获得:30s AiHeartbeat 喂狗 + run_command 实时 + // 流式 sink + spawn_agent 父会话 task-local 注入 + outer_secs 分档(spawn_agent=挂钟 + // 预算+60 / run_command=args.timeout_secs clamp [60,600],此前恒 60s 会腰斩长构建, + // 现 run_command 内部超时先触发返带语义的 tool_result)。 + execute_with_heartbeat( + &state.ai_tools, &approval.tool_name, args.clone(), + &app, conv_id.as_deref().unwrap_or_default(), &id, + ).await }; // 工具失败不 return Err：把错误包成 tool_result，落库 + emit completed + 续循环全走通。 // 否则前端 approveToolCall 的 catch 会回滚 pending_approval，审批按钮卡死无法消除。 @@ -1451,20 +1427,18 @@ pub async fn ai_authorize_dir( // ── 执行工具(路径现已授权,handler 内 resolve_workspace_path_with_allowed 放行) ── // 复用 ai_approve 执行分支:run_workflow 特殊处理 / 其余走 ai_tools.execute tracing::debug!(tool = %approval.tool_name, "[AI-AUTHZ] execute 前"); - // F-260620 防工具卡死 ai_authorize_dir(致 per_conv.generating 永真 + 前端看门狗 130s 超时后 - // 静默吞消息):list_directory 等大目录工具可能 read_dir+metadata 海量文件卡住,execute 不返回 - // → 到不了 audit_finalize/emit/try_continue → IPC 永挂 → generating 永真。包 60s 超时(对齐 - // run_command),超时返 Err → 走 failed 分支 finalize+emit+try_continue,IPC 必返回不卡。 + // 防工具卡死:超时兜底已收敛进 execute_with_heartbeat 的 outer_secs 分档, + // execute 不返回时超时返 Err → 走 failed 分支 finalize+emit+try_continue,IPC 必返回不卡。 let exec_result: anyhow::Result = if approval.tool_name == "run_workflow" { super::super::execute_run_workflow_for_tool(&app, &state, &args, conv_id.as_deref()).await } else { - match tokio::time::timeout( - std::time::Duration::from_secs(60), - state.ai_tools.execute(&approval.tool_name, args.clone()), - ).await { - Ok(r) => r, - Err(_) => Err(anyhow::anyhow!("工具执行超时(60s): {}", approval.tool_name)), - } + // 授权执行窗口复用 execute_with_heartbeat(对齐 ai_approve,详见其执行段注释): + // 授权挂起期主 loop 心跳已停,慢文件工具(如跨盘 copy+remove 降级)整窗静默 + // → 前端看门狗误杀;附带补齐 run_command 实时流式与超时 clamp(原恒 60s)。 + execute_with_heartbeat( + &state.ai_tools, &approval.tool_name, args.clone(), + &app, conv_id.as_deref().unwrap_or_default(), &tool_call_id, + ).await }; tracing::debug!( ok = exec_result.is_ok(), @@ -1742,7 +1716,9 @@ pub async fn ai_chat_compress_context( if protect_start == 0 { // 空会话/全在保护区 → 无可压缩消息,set_compressing(false) 后返回 Ok conv.messages.set_compressing(false); - let cid = session.active_conversation_id.clone(); + // noop 事件归属 IPC 参数 conversation_id(与上方 set_compressing(false) 同一 conv), + // 不读 active_conversation_id:后台压缩非 active 会话时事件须路由回发起压缩的会话。 + let cid = Some(conversation_id.clone()); drop(session); // 治 Task#1:手动 IPC 路径用 AiManualCompressed(前端弹 toast 告知"无可压缩/已完成")。 let ev = AiChatEvent::AiManualCompressed { @@ -1760,7 +1736,11 @@ pub async fn ai_chat_compress_context( .filter(|t| t.message.is_active()) .map(|t| t.message.clone()) .collect(); - let cid = session.active_conversation_id.clone().unwrap_or_else(|| conversation_id.clone()); + // 压缩目标一律取 IPC 参数 conversation_id:上方 conv()/set_compressing(true)/active 克隆 + // 全部落在此 conv,后续突变(复位 compressing/标 compressed/插摘要/落库)必须同 conv, + // 否则后台压缩非 active 会话时目标 conv 的 compressing 标志永久卡死,active 会话反被 + // 插入他人摘要。AiCompressing/AiManualCompressed 事件本就按 conversation_id 路由,归属一致。 + let cid = conversation_id.clone(); log_slow_lock(__lock_t1205, "ai_chat_compress_context:1205"); (cid, active_msgs, protect_start) }; @@ -2137,8 +2117,11 @@ pub async fn ai_chat_force_send( } else { conv.messages.push(ChatMessage::user(&user_content)); } - // 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级) - conv.working_context.begin_turn(&user_content, 200); + // 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级)。 + // 喂原文 message 而非落库变换文本 user_content:begin_turn 契约是"本条用户请求", + // 降级提示尾缀与 /{skill} 空文本标记是落库与标题素材专用,当目标会污染钉扎文本; + // 纯技能调用原文为空,靠 begin_turn 自身的空文本跳过语义不记目标。 + conv.working_context.begin_turn(&message, 200); // F-260619-04 P1:push 后立即取末条 user 消息 id(供知识注入 referenced 溯源)。 let user_msg_id = conv.messages.last_user_message_id(); // G1 目标钉扎已迁移至 run_agentic_loop 工具调用推理(提取方式同 ai_chat_send) @@ -2169,6 +2152,7 @@ pub async fn ai_chat_force_send( // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, incomplete: None, + incomplete_reason: None, conversation_id: Some(cid.clone()), pinned_goals: old_pinned_goals.clone(), }; @@ -2294,6 +2278,7 @@ pub async fn ai_chat_stop( // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, incomplete: None, + incomplete_reason: None, conversation_id: Some(target), pinned_goals, }; @@ -2405,6 +2390,7 @@ pub async fn ai_chat_stop( // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, incomplete: None, + incomplete_reason: None, conversation_id: conv_id.clone(), pinned_goals, }; @@ -2458,6 +2444,78 @@ pub async fn ai_continue_loop( Ok("ok".to_string()) } +/// 流中断自动续跑探测:前端看门狗超时后调用,按后端真实状态三分支返回。 +/// 返回 "generating":loop 仍在跑(前端误判),前端重挂看门狗即可; +/// 返回 "resumed":检测到中断态(尾部无 assistant 终答),已复位生成态并重新 spawn loop; +/// 返回 "finished":会话已正常收敛(尾部是 assistant 文本终答),前端兜底收敛即可。 +#[tauri::command] +pub async fn ai_resume_interrupted( + app: AppHandle, + state: State<'_, AppState>, + conversation_id: String, +) -> Result { + { + let __lock_t = std::time::Instant::now(); + let mut session = state.ai_session.lock().await; + // loop 仍在跑 → 前端误判超时,不 spawn(防双跑),前端重挂看门狗即可。 + // 检查必须在 session lock 内:主窗口+分离窗口双探针并发探测时,锁外检查与 + // 后续 spawn 之间存在 TOCTOU 窗口(两探针都通过检查后各自续跑双 spawn); + // 锁内「查 active→transition Generating」对其他探针原子,只放行一方。 + if state.conv_states.is_active(&conversation_id) { + log_slow_lock(__lock_t, "ai_resume_interrupted"); + return Ok("generating".to_string()); + } + // conv_read 只读(存在性 + 尾部判定);突变走下面 session.conv(&mut)。 + let interrupted = match session.conv_read(&conversation_id) { + Some(conv) => { + // 尾部特征判定:最后一条是 Tool 结果或 assistant 带工具调用 → loop 死在 + // 工具轮中途(未 emit 终止事件);尾部是 assistant 纯文本 → 已正常收敛。 + match conv.messages.iter().last() { + Some(m) => match m.role { + df_ai::provider::MessageRole::Tool => true, + df_ai::provider::MessageRole::Assistant => { + m.tool_calls.as_ref().map_or(false, |t| !t.is_empty()) + } + // 尾部是 user(loop 死在首 token 前,LLM 未产出任何内容)也属中断, + // 走 resumed 续跑;归 finished 会误判收敛,前端补假「已收敛」信号。 + df_ai::provider::MessageRole::User => true, + _ => false, + }, + None => false, + } + } + None => { + log_slow_lock(__lock_t, "ai_resume_interrupted"); + return Err("会话不存在".to_string()); + } + }; + if !interrupted { + log_slow_lock(__lock_t, "ai_resume_interrupted"); + return Ok("finished".to_string()); + } + // 中断态复位:stop_flag 清残留(防续跑入口即退出),iteration 重计 + // (对齐 ai_continue_loop 用户点继续=授权重计语义)。 + let conv = session.conv(&conversation_id); + conv.stop_flag.store(false, Ordering::SeqCst); + conv.iteration_used = 0; + log_slow_lock(__lock_t, "ai_resume_interrupted"); + // conv_state 写侧 transition 失败 = 状态机不在可续跑态(如已 Idle=另有方在收尾), + // warn 且不续跑:续跑前提是本探针独占置起 Generating,叠加 spawn 会双跑。 + if let Err(e) = state.conv_states.transition(&conversation_id, ConvState::Generating) { + tracing::warn!( + conv_id = %conversation_id, + error = %e, + "[ai] conv_states ai_resume_interrupted→Generating 非法(不续跑)" + ); + log_slow_lock(__lock_t, "ai_resume_interrupted"); + return Ok("generating".to_string()); + } + } + // 锁已释放(独立块作用域)再续跑,复用审批恢复入口,内部 spawn run_agentic_loop。 + try_continue_agent_loop(&app, &state, &conversation_id, 0).await; + Ok("resumed".to_string()) +} + /// 停止 agentic 循环并走完成流程达 max_iterations 暂停态用户点「停止」） /// /// 场景：run_agentic_loop 达 max_iterations 未收敛 → emit AiMaxRoundsReached + 保持 @@ -2508,6 +2566,7 @@ pub async fn ai_stop_loop( // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, incomplete: None, + incomplete_reason: None, conversation_id: Some(conversation_id), pinned_goals, }; diff --git a/src-tauri/src/commands/ai/commands/conversation.rs b/src-tauri/src/commands/ai/commands/conversation.rs index fe6c627..267ea67 100644 --- a/src-tauri/src/commands/ai/commands/conversation.rs +++ b/src-tauri/src/commands/ai/commands/conversation.rs @@ -227,9 +227,10 @@ pub async fn ai_conversation_list( let limit = limit.unwrap_or(50); let include_archived = include_archived.unwrap_or(false); let records = state.ai_conversations.list_all().await.map_err(err_str)?; - // list_all 已按 created_at DESC(最新在前）；默认排除归档 + 截断 limit + // list_all 已按 created_at DESC(最新在前）；默认排除归档 + 截断 limit。 + // origin 非空(子代理轨迹,V44)不进侧栏列表——轨迹在库可追溯,展示面只留主会话。 let summaries: Vec = records.iter() - .filter(|r| include_archived || !r.archived) + .filter(|r| (include_archived || !r.archived) && r.origin.is_none()) .take(limit) .map(|r| { // 修复 models 字段类型 bug:r.models 是 JSON 字符串,前端期望数组 @@ -606,16 +607,33 @@ pub async fn ai_conversation_load_more( /// 删除对话 #[tauri::command] pub async fn ai_conversation_delete( + app: AppHandle, state: State<'_, AppState>, conversation_id: String, ) -> Result<(), String> { // 记录已删除 conv_id,必须先于 DB 删除写入:save_conversation 入口检测到即跳过, // 封死「loop 在途 save 复活」窗口(不惰性重建 per_conv、不 Ok(None) INSERT 空壳)。 // 删除语义:删了就删了,后续 save 不应复活。conv_id 是 ULID 不重用,标记长期保留。 - { + // + // 同锁内停 in-flight agentic loop(范式对齐 ai_chat_force_send / ai_chat_clear 的锁内写法): + // stop_flag 置位让 loop 在下一轮闸门退出;notify_one 唤醒阻塞在 stream select! 的 loop + // 不必等 chunk;epoch 递增使旧 loop 立即失去 owner 身份,其全部退出路径的 + // guard.reset/emit/save 因 epoch 不匹配静默跳过,不会在已删 conv 上复活生成态。 + // 用 conv_read 而非 conv():conv() 对缺失条目惰性重建空壳,恰是删除场景要避免的; + // 条目已不存在(重复删除/本就无 loop)则无事可停。 + let loop_was_alive = { let mut session = state.ai_session.lock().await; session.deleted_convs.insert(conversation_id.clone()); - } + match session.conv_read(&conversation_id) { + Some(conv) => { + conv.stop_flag.store(true, Ordering::SeqCst); + conv.notify.notify_one(); + conv.loop_epoch.fetch_add(1, Ordering::SeqCst); + true + } + None => false, + } + }; // 单事务删 ai_messages 子行 + ai_conversations 主行(防孤儿消息)。 // 顺序:先删数据再摘内存(下方 per_conv.remove)。 @@ -655,6 +673,24 @@ pub async fn ai_conversation_delete( if was_active { state.clear_session_allowed_dirs_if_idle().await; } + + // 终态事件单点补发:上方 epoch 递增后,被停 loop 的全部退出路径(loop 入口已删 / stop 闸门 + // teardown / 流后 stale / MidStream 保文跳过 teardown / push 后删除检查)都因 epoch 不匹配 + // 静默退出,不再发任何 AiCompleted,前端会永久卡 Generating。故由删除侧单点补发,一次覆盖 + // 所有退出路径,天然防重复(loop 侧 emit 全部带 epoch owner 守卫,删除后必不触发)。 + // 走正典 emit_ai_completed_once(零 token 先例见 agentic/helpers.rs),incomplete 语义对齐 + // D 线契约:emit 端 Some(true) + reason "deleted"(前端 types.ts 仅特判 max_rounds,其余 + // 值走通用不完整提示),publish 端 None 对齐 MidStream 分叉(总线消费方无需删除标记)。 + // notify=false:会话已删除,不发"任务完成"系统通知。仅 loop_was_alive 时补发,重复删除 + // 或无 loop 的 conv 不产生多余事件。 + if loop_was_alive { + let zero_usage = df_ai::provider::TokenUsage::default(); + crate::commands::ai::agentic::emit_ai_completed_once( + &app, &conversation_id, &zero_usage, false, + Some(true), None, Some("deleted"), true, + &[], false, + ).await; + } Ok(()) } diff --git a/src-tauri/src/commands/ai/conversation.rs b/src-tauri/src/commands/ai/conversation.rs index 8af4e68..4411049 100644 --- a/src-tauri/src/commands/ai/conversation.rs +++ b/src-tauri/src/commands/ai/conversation.rs @@ -253,7 +253,7 @@ async fn save_conversation_inner( // 避免同时持有 session 的 mut 和 immut 借用(E0502)。 // BE1无变化捷径提前到克隆之前——同一 lock 段内先判「消息数未增且非 dirty」, // 无变化则不 clone(每轮无变化 save 省一次全量 clone + 后续 truncate/records 构建)。 - let (provider_id, created_at, pinned_goals, msg_len, persisted_count, needs_full_rewrite, mut persist_msgs) = { + let (provider_id, created_at, pinned_goals, msg_len, persisted_count, needs_full_rewrite, mut persist_msgs, conv_origin) = { let mut session = session_arc.lock().await; // 已删除对话直接跳过写入(loop 在「push 后、save 前」被删除的幽灵复活根治): // 此处不惰性重建 per_conv、不触发 INSERT 空壳 + insert_batch 复活。 @@ -270,6 +270,8 @@ async fn save_conversation_inner( // 避免同时持有 session 的 mut 和 immut 借用(E0502)。 let provider_id = session.active_provider_id.clone(); let created_at = session.active_conv_created_at.clone(); + // 会话来源标记(V44 子轨迹落库):子代理 conv 首插时随 record 落 origin 列,主会话 None + let conv_origin = session.conv_origins.get(conv_id).cloned(); let conv = session.conv(conv_id); let msg_len = conv.messages.len(); let persisted_count = conv.messages.persisted_msg_count(); @@ -291,6 +293,7 @@ async fn save_conversation_inner( persisted_count, needs_full_rewrite, cloned, + conv_origin, ) }; let __lock_total = __lock_start.elapsed(); @@ -469,6 +472,7 @@ async fn save_conversation_inner( pending_approvals: Some(pending_approvals_json.clone()), created_at: conv_created.clone(), updated_at: now, + origin: conv_origin, }; if let Err(e) = conv_repo.insert(rec).await { tracing::warn!("落库对话失败 {conv_id}: {e}"); diff --git a/src-tauri/src/commands/ai/mod.rs b/src-tauri/src/commands/ai/mod.rs index 6de4114..785bc0d 100644 --- a/src-tauri/src/commands/ai/mod.rs +++ b/src-tauri/src/commands/ai/mod.rs @@ -181,6 +181,10 @@ pub enum AiChatEvent { is_estimated: bool, /// 不完整标记(可选):Some(true)=网络中断保文,None 或 Some(false)=完整回复 incomplete: Option, + /// 不完整原因(与 incomplete 配合):Some("max_rounds")=达最大轮次自动收尾, + /// Some("network")=流中途网络中断保文,None=完整回复/未分类。前端据此选文案: + /// "max_rounds" → 「已达最大轮次自动收尾」,其他/缺省 → 网络中断文案。 + incomplete_reason: Option, conversation_id: Option, /// 当前会话 pinned_goals(G1 目标钉扎持久化,前端直接读取刷新) pinned_goals: Vec, @@ -231,6 +235,7 @@ pub enum AiChatEvent { /// - `conversation_id`:父会话 id(路由键,挂父不挂子,子会话不在侧栏) /// - `status`:"completed" | "truncated" | "failed" /// - `rounds`/`tokens`:实际执行轮数与累计 token(启动失败为 0) + /// - `rejected_tools`:被拒工具清单("工具名×次数",如 "write_file×3"),None=本次零拒绝 AiAgentFinished { agent_id: String, conversation_id: Option, @@ -238,6 +243,9 @@ pub enum AiChatEvent { status: String, rounds: u32, tokens: u64, + /// S2 报告协议:子循环运行期被拒工具汇总,让父会话与用户看见子代理被拒了什么。 + /// 可选字段:前端 S1 完成卡不消费该字段也不破坏(事件多字段向后兼容),消费留二期。 + rejected_tools: Option>, }, /// 流式调用自动重试提示流前失败重试中） /// @@ -643,6 +651,10 @@ pub struct AiSession { /// 删除语义:删了就删了,loop 后续 save 不应复活。conv_id 是 ULID 不重用,条目长期保留 /// (单条 ~40B,删除量级下可忽略)。 pub deleted_convs: HashSet, + /// 会话来源标记(V44 子轨迹落库):conv_id → origin(现仅 "sub_agent")。 + /// save 首插时随 record 落 origin 列,桌面侧栏/miniapp 列表按此过滤子会话。 + /// 子代理启动时写入,出口清理时随 per_conv 一并移除;主会话无条目(None)。 + pub conv_origins: HashMap, /// 每会话 save 串行锁:同会话并发 save(loop 与 IPC)串行化,防全量重写互踩吞新行 pub save_locks: HashMap>>, /// T2 工具命名空间存储（运行时缓存，大工具结果不进主队列） @@ -658,6 +670,7 @@ impl AiSession { pending_approvals: HashMap::new(), per_conv: HashMap::new(), deleted_convs: HashSet::new(), + conv_origins: HashMap::new(), save_locks: HashMap::new(), namespace_store: df_ai::namespace_store::NamespaceStore::default(), } diff --git a/src-tauri/src/commands/ai/remote_bridge.rs b/src-tauri/src/commands/ai/remote_bridge.rs index 70f80d2..e7854d6 100644 --- a/src-tauri/src/commands/ai/remote_bridge.rs +++ b/src-tauri/src/commands/ai/remote_bridge.rs @@ -422,8 +422,10 @@ async fn route_list_conversations(state: &State<'_, AppState>) { }; // 映射摘要 + parse 时间戳。updated_at 缺失/脏 → None(排序兜底 0)。 + // origin 非空(子代理轨迹,V44)不推 miniapp——与桌面侧栏同一过滤口径。 let mut summaries: Vec = records .into_iter() + .filter(|r| r.origin.is_none()) .map(|r| ConvSummary { id: r.id, title: r.title.unwrap_or_default(), @@ -809,8 +811,8 @@ async fn route_send_message(app: &AppHandle, state: &State<'_, AppState>, args: conv_id = ?reject_conv_id, "[remote_bridge][R1] 同 conv 正在生成中,拒绝远程 send_message(emit AiError 回 miniapp)" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: "正在生成中，请等待完成".to_string(), // 并发拒绝非错误源分类(auth/network/timeout/provider_config),用 Unknown 兜底。 @@ -868,8 +870,8 @@ async fn route_send_message(app: &AppHandle, state: &State<'_, AppState>, args: conv_id = ?conversation_id, "[remote_bridge] ai_chat_send 返回 Err,转 AiError emit 回 miniapp" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: err_msg, // ai_chat_send 的 Err 多为 generating 拦截 / Provider 配置缺失, @@ -948,8 +950,8 @@ async fn route_force_send(app: &AppHandle, state: &State<'_, AppState>, args: Va conv_id = ?conversation_id, "[remote_bridge] ai_chat_force_send 返回 Err,转 AiError emit 回 miniapp" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: err_msg, error_type: Some(crate::commands::ai::ErrorType::Unknown), @@ -997,8 +999,8 @@ async fn route_edit_message(app: &AppHandle, state: &State<'_, AppState>, args: conv_id = %conversation_id, "[remote_bridge] ai_chat_edit 返回 Err,转 AiError emit 回 miniapp" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: err_msg, error_type: Some(crate::commands::ai::ErrorType::Unknown), @@ -1031,8 +1033,8 @@ async fn route_compress_context(app: &AppHandle, state: &State<'_, AppState>, ar conv_id = %conversation_id, "[remote_bridge] ai_chat_compress_context 返回 Err,转 AiError emit 回 miniapp" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: err_msg, error_type: Some(crate::commands::ai::ErrorType::Unknown), @@ -1063,8 +1065,8 @@ async fn route_clear_context(app: &AppHandle, state: &State<'_, AppState>, args: conv_id = %conversation_id, "[remote_bridge] ai_chat_clear_context 返回 Err,转 AiError emit 回 miniapp" ); - let _ = app.emit( - "ai-chat-event", + emit_ai_chat_event( + app, AiChatEvent::AiError { error: err_msg, error_type: Some(crate::commands::ai::ErrorType::Unknown), diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index fe9e466..bc1a32d 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -428,6 +428,8 @@ pub fn run() { // 达 max_iterations 暂停态续/停（消费 AiMaxRoundsReached，前端操作卡留 batch45） commands::ai::ai_continue_loop, commands::ai::ai_stop_loop, + // 流中断自动续跑探测(前端看门狗超时后调,按后端真实状态续跑/兜底收敛) + commands::ai::ai_resume_interrupted, commands::ai::ai_pending_tool_calls, commands::ai::ai_chat_clear, commands::ai::ai_is_generating,