From c7d0ed939c205f35af661aa3f9987dcb07c6757e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sun, 11 Oct 2026 12:37:23 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20=E8=BF=9E=E5=8F=91?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E9=98=B2=E6=8A=96=E4=B8=8E=E9=95=BF=E5=91=BD?= =?UTF-8?q?=E4=BB=A4=E5=90=8E=E5=8F=B0=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 飞书连发消息 0.5s 窗(拍板值)合并为一次任务:文本空行拼接+图片到达序合并 - 命令/审批/注入即时路径不进窗;追加消息并入即静默返回;窗满统一受理,守卫拒绝落待发箱 - run_command 超 grace 窗(默认 90s,可配)转后台:LLM 收非终态结果,不再整轮阻塞 - 后台任务自持心跳+删除自中止+失败记忆+完成通知+回填续跑;占用闸单点保会话态自洽 - 开关默认关=同步路径逐字节不变;新增 13 测 --- src-tauri/src/commands/ai/agentic/guard.rs | 9 + src-tauri/src/commands/ai/audit/bg_run.rs | 539 +++++++++++++++++++++ src-tauri/src/commands/ai/audit/mod.rs | 251 +++++++--- src-tauri/src/feishu_channel.rs | 185 ++++++- 4 files changed, 894 insertions(+), 90 deletions(-) create mode 100644 src-tauri/src/commands/ai/audit/bg_run.rs diff --git a/src-tauri/src/commands/ai/agentic/guard.rs b/src-tauri/src/commands/ai/agentic/guard.rs index 07a5a46..55fec9d 100644 --- a/src-tauri/src/commands/ai/agentic/guard.rs +++ b/src-tauri/src/commands/ai/agentic/guard.rs @@ -83,6 +83,15 @@ impl GeneratingGuard { /// 迁移 ConvState + emit AiConvStateChanged(经无锁 ConvStateStore,同步无 await)。 fn transition(&self, target: ConvState) { + // 后台工具占用闸:后台命令在飞时拦截 loop 退出的 Idle 迁移, + // 会话保持 Generating(受理守卫判占用、前端不显示空闲),并记录「loop 已退」 + // 供 detached task 终局决策 try_continue 续跑;占用复位由最后一个后台任务 + // 终局时放行(audit/bg_run.rs)。用户停止/致命错误路径不经过此闸(仅拦 Idle)。 + if target == ConvState::Idle + && crate::commands::ai::audit::bg_run::hold_blocks_idle(&self.conv_id) + { + return; + } let conv_states = self .app_handle .state::() diff --git a/src-tauri/src/commands/ai/audit/bg_run.rs b/src-tauri/src/commands/ai/audit/bg_run.rs new file mode 100644 index 0000000..a8a9ac3 --- /dev/null +++ b/src-tauri/src/commands/ai/audit/bg_run.rs @@ -0,0 +1,539 @@ +//! run_command 超时转后台执行。 +//! +//! 长命令(构建/测试)整轮阻塞 agent——`run_command` 占 >90s 工具桶的 95%。 +//! 本模块把超 grace 窗口未完成的工具 future 移交 detached task:LLM 立即拿到 +//! 「已转后台」非终态 tool_result 继续轮次(或自然收尾);命令跑完后结果经 +//! inject_runtime_message 注入原会话,并按需 try_continue_agent_loop 续跑。 +//! +//! 三条红线(勘察定): +//! - 心跳接管:execute_with_heartbeat 的心跳 task 随函数返回即 abort,前端 130s +//! 看门狗靠它续命——detached task 自持 30s 心跳并刷新 per_conv.last_heartbeat +//! (loop 存活证据,与 agentic heartbeat_loop / ai_chat_stop 兜底判活同源)。 +//! - 会话占用:后台运行期会话必须维持 Generating(受理守卫判占用)。loop 收敛 +//! 退出时 GeneratingGuard 的 Idle 迁移被占用闸拦截(guard.rs),最后一个 +//! detached 终局时才放行复位——占用语义单点落在 guard/conv_state 层,loop +//! 主体零改动。用户停止(Stopping)与致命错误(Error)不拦截:前者终局时 +//! 结果落 system 通知后放行,后者保持 Error 由用户重试流程自管。 +//! - 回填禁楔:结果注入复用 inject_runtime_message(尾部追加,既有 system 锚点 +//! 避让保护),不自写裸 insert——楔进 assistant(tool_calls) 头与其 tool result +//! 之间会破坏「result 紧随其头」的上游不变式直接报 400;同 tool_call_id +//! 也不得二次回填 tool_result +//! (非终态 result 已占位,重复 id 会触发 provider 400)。 +//! +//! 开关:KV df-ai-run-command-background(默认关=同步路径逐字节不变)+ +//! df-ai-run-command-grace-secs(默认 90,解析失败/0 按默认)。仅主会话生效: +//! 子代理有独立循环(run_sub_agent_loop)与挂钟预算,try_continue_agent_loop 是 +//! 主循环恢复通道,子会话误续跑会绕开子代理全部约束,故 conv_origins 命中 +//! sub_agent 一律不转后台(超时仍走原 outer_secs 腰斩)。 + +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; + +use dashmap::DashMap; +use tauri::{AppHandle, Manager}; + +use crate::commands::ai::agentic::conv_state::ConvState; +use crate::commands::ai::agentic::helpers::insert_system_notice; +use crate::commands::ai::agentic::try_continue_agent_loop; +use crate::commands::ai::commands::chat::inject_runtime_message; +use crate::commands::ai::event_bus::emit_ai_chat_event; +use crate::commands::ai::AiChatEvent; +use crate::state::{AppState, SettingsNorm}; + +pub(crate) const BG_SWITCH_KEY: &str = "df-ai-run-command-background"; +pub(crate) const BG_GRACE_KEY: &str = "df-ai-run-command-grace-secs"; +const DEFAULT_GRACE_SECS: u64 = 90; +/// 与 sub_agent.rs SUB_AGENT_ORIGIN 同值(该 const 私有,此处按值比对 conv_origins) +const SUB_AGENT_ORIGIN: &str = "sub_agent"; + +// ============================================================ +// 资格判定 +// ============================================================ + +/// 后台化资格:开关开 + run_command + 主会话 → Some(grace 秒);否则 None +/// (调用方走同步原路径,行为逐字节不变)。每工具一次 KV 读(可接受成本), +/// 非 run_command 先短路不触 KV。 +pub(crate) async fn background_grace_secs( + app: &AppHandle, + name: &str, + conv_id: &str, +) -> Option { + if name != "run_command" { + return None; + } + let state = app.state::(); + if !state.settings.get_bool_norm(BG_SWITCH_KEY, false).await { + return None; + } + let is_sub = { + let session = state.ai_session.lock().await; + session + .conv_origins + .get(conv_id) + .map(|o| o == SUB_AGENT_ORIGIN) + .unwrap_or(false) + }; + if is_sub { + return None; + } + Some(parse_grace(state.settings.get_string_norm(BG_GRACE_KEY).await)) +} + +/// grace 解析:非法/0 归默认 90(0 会让命令瞬时转后台,视为配置错误不采纳) +fn parse_grace(raw: Option) -> u64 { + raw.and_then(|s| s.parse::().ok()) + .filter(|g| *g > 0) + .unwrap_or(DEFAULT_GRACE_SECS) +} + +// ============================================================ +// per-conv 后台占用登记(会话删除/停止感知的最小形态:per-conv 计数) +// ============================================================ + +/// in_flight = 在飞 detached 数;loop_exited = loop 已收敛退出但被占用闸拦下 +/// 未复位 Idle(guard.rs 置位,终局决策消费)。 +struct BgHold { + in_flight: usize, + loop_exited: bool, +} + +static BG_HOLDS: OnceLock> = OnceLock::new(); + +fn holds() -> &'static DashMap { + BG_HOLDS.get_or_init(DashMap::new) +} + +/// 占用闸(guard.rs 调用):后台任务在飞时拦截 loop 退出的 Idle 迁移,并记录 +/// 「loop 已退」供 detached 终局决策决定是否 try_continue 续跑。 +pub(crate) fn hold_blocks_idle(conv_id: &str) -> bool { + let Some(mut hold) = holds().get_mut(conv_id) else { + return false; + }; + if hold.in_flight == 0 { + return false; + } + hold.loop_exited = true; + true +} + +/// 只读在飞判定(终局放行用,不带置位副作用) +fn hold_in_flight(conv_id: &str) -> bool { + holds().get(conv_id).map(|h| h.in_flight > 0).unwrap_or(false) +} + +fn hold_acquire(conv_id: &str) { + holds() + .entry(conv_id.to_string()) + .or_insert(BgHold { in_flight: 0, loop_exited: false }) + .in_flight += 1; +} + +/// hold 名额 RAII:正常路径显式 [`release`](Self::release)(返回是否最后一个); +/// detached task 异常 unwind 时 Drop 兜底释放,防名额泄漏致会话永久卡 Generating。 +struct HoldGuard { + conv: String, + released: bool, +} + +impl HoldGuard { + /// 接管 spawn_detached 已登记的名额(不重复 acquire) + fn takeover(conv_id: &str) -> Self { + Self { conv: conv_id.to_string(), released: false } + } + + fn release(&mut self) -> bool { + if self.released { + return false; + } + self.released = true; + hold_release(&self.conv) + } +} + +impl Drop for HoldGuard { + fn drop(&mut self) { + self.release(); + } +} + +/// 释放一个在飞名额,返回是否最后一个(终局复位语义只在最后一个生效) +fn hold_release(conv_id: &str) -> bool { + let last = match holds().get_mut(conv_id) { + Some(mut h) => { + h.in_flight = h.in_flight.saturating_sub(1); + h.in_flight == 0 + } + None => true, // 条目缺失视为无在飞(防御,不该发生) + }; + if last { + holds().remove(conv_id); + } + last +} + +/// 读「loop 已退出」标记(只读不消费,须在 hold.release 前调——release 摘除条目 +/// 会连带吞掉标记):true=loop 已收敛退出,最后一个名额终局时需 try_continue 续跑 +/// 消费注入消息;false=loop 仍在飞,活跃 loop 下一轮自然消费,续跑反而抢断当前轮。 +fn peek_loop_exited(conv_id: &str) -> bool { + holds() + .get(conv_id) + .map(|h| h.loop_exited) + .unwrap_or(false) +} + +// ============================================================ +// grace 竞速 + 移交 +// ============================================================ + +/// grace 窗口三路竞速结果 +pub(crate) enum GraceRace { + /// 窗口内完成:与同步原路径结果完全一致 + Done(R), + /// 窗口超时:future 未被消费,归还调用方移交 detached task + Expired, + /// 监视判死(会话删除/审批失效):同原提前退出语义 + Aborted(String), +} + +/// exec 与 grace 计时、监视 future 三路 select。select 只借 `&mut *exec`, +/// 借用随 select 结束解除,超时分支把未消费的 future 原样归还不违反所有权 +/// (这正是「移交 detached」的机制基础:future 在竞速失败后仍可独立驱动)。 +pub(crate) async fn race_grace( + exec: Pin + Send>>, + grace: Duration, + watch: W, +) -> (GraceRace, Pin + Send>>) +where + W: Future, +{ + let mut exec = exec; + let raced = tokio::select! { + r = exec.as_mut() => GraceRace::Done(r), + _ = tokio::time::sleep(grace) => GraceRace::Expired, + reason = watch => GraceRace::Aborted(reason), + }; + (raced, exec) +} + +/// 非终态 tool_result:LLM 立即拿到明确「已转后台」信号,可继续轮或自然收尾; +/// 完成后结果自动回传本会话(ReAct 语义变更,开关默认关保回退)。 +pub(crate) fn background_payload(grace: u64) -> serde_json::Value { + serde_json::json!({ + "status": "background", + "message": format!("已转后台执行(超 {grace}s),完成后结果自动回传本会话"), + }) +} + +/// grace 超时移交:登记占用(先于返回,guard 占用闸随即生效)→ spawn detached。 +/// **必须是普通同步函数(内部无 await)**:调用方(execute_with_heartbeat)以普通调用 +/// 而非 await 进入,background_run 的 coroutine 不进调用方协程的 Send 验证链——否则 +/// background_run→try_continue→run_agentic_loop→execute_with_heartbeat 构成自动 trait +/// 验证环(rustc E0391)。 +pub(crate) fn spawn_detached( + exec: Pin> + Send>>, + name: String, + args_for_memory: serde_json::Value, + app: AppHandle, + conv_id: String, + tool_call_id: String, + last_heartbeat: Option>, +) { + // F2 存活心跳证据源:detached 续写 per_conv.last_heartbeat(ai_chat_stop 兜底判活同源); + // conv 已删时给孤儿 Arc(store 无消费者,仅保类型) + let hb = last_heartbeat.unwrap_or_else(|| Arc::new(AtomicU64::new(0))); + hold_acquire(&conv_id); + tokio::spawn(background_run(exec, name, args_for_memory, app, conv_id, tool_call_id, hb)); +} + +/// detached task 本体:自持心跳 + 会话删除自中止 + 失败记忆 + 完成通知 + 回填续跑。 +/// 后台失败同样有终局(注入失败消息),不静默蒸发。 +async fn background_run( + mut exec: Pin> + Send>>, + name: String, + args_for_memory: serde_json::Value, + app: AppHandle, + conv_id: String, + tool_call_id: String, + last_heartbeat: Arc, +) { + // 接管 spawn_detached 已登记的名额:panic unwind 时 Drop 兜底释放,防占用泄漏 + let mut hold = HoldGuard::takeover(&conv_id); + // 自持心跳:接管随 execute_with_heartbeat 返回被 abort 的原心跳,30s emit + + // 刷 last_heartbeat,后台运行期前端看门狗持续续命 + let hb_stop = Arc::new(AtomicBool::new(false)); + let heartbeat = { + let app = app.clone(); + let conv = conv_id.clone(); + let stop = hb_stop.clone(); + let hb_arc = last_heartbeat.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(30)); + interval.tick().await; // 弃首 tick(对齐原心跳,首 tick 立即返回) + loop { + interval.tick().await; + if stop.load(Ordering::SeqCst) { + break; + } + hb_arc.store(df_types::now_millis() as u64, Ordering::SeqCst); + emit_ai_chat_event(&app, AiChatEvent::AiHeartbeat { conversation_id: Some(conv.clone()) }); + } + }) + }; + + // 会话删除自中止(对齐 execute_with_heartbeat 提前退出监视):执行中删除会话, + // 后台命令最多再跑 2s 即中止,不再空转到命令自身超时。审批失效监视不武装 + // (detached 诞生于执行期,pending_approvals 必无此 id,按字面监视为死分支) + let watch_app = app.clone(); + let watch_conv = conv_id.clone(); + let deleted_watch = async move { + loop { + tokio::time::sleep(Duration::from_secs(2)).await; + let state = watch_app.state::(); + let session = state.ai_session.lock().await; + if session.deleted_convs.contains(&watch_conv) { + return format!("会话 {} 已删除", watch_conv); + } + } + }; + + let outcome = tokio::select! { + r = exec.as_mut() => Ok(r), + reason = deleted_watch => Err(reason), + }; + hb_stop.store(true, Ordering::SeqCst); + heartbeat.abort(); + + // 失败记忆(与同步路径 postlude 同源):成功清零 / 失败累加,后台失败也进 + // 断路统计,防 LLM 拿到失败后原样重调 + let (ok, result_text) = match outcome { + Ok(Ok(v)) => { + crate::commands::ai::tool_failure_memory::record(&name, &args_for_memory, true, None); + (true, v.to_string()) + } + Ok(Err(e)) => { + let msg = e.to_string(); + crate::commands::ai::tool_failure_memory::record(&name, &args_for_memory, false, Some(msg.clone())); + (false, format!("工具 {} 后台执行失败: {}", name, msg)) + } + Err(reason) => { + crate::commands::ai::tool_failure_memory::record(&name, &args_for_memory, false, Some(reason.clone())); + (false, format!("工具 {} 后台执行中止: {}", name, reason)) + } + }; + + // 完成通知:emit Completed 让前端工具卡从「已转后台」收敛为真实结果 + emit_ai_chat_event(&app, AiChatEvent::AiToolCallCompleted { + id: tool_call_id.clone(), + result: serde_json::Value::String(result_text.clone()), + conversation_id: Some(conv_id.clone()), + }); + + let note = if ok { + format!("后台命令「{name}」已执行完成,结果如下:\n{result_text}") + } else { + format!("后台命令「{name}」执行失败,详情如下:\n{result_text}") + }; + let state = app.state::(); + // 先注入后释放名额:占用期 is_active 恒真(Generating),注入守卫全场景可过; + // 先释放会在「名额清零且 loop 已退」窗口露出非活跃态,注入误走降级分支 + match inject_runtime_message(&state, &conv_id, ¬e).await { + Ok(true) => { + // 先读「loop 已退」再释放名额(release 摘除条目会吞标记);多名额时 + // 标记保持 true,由最后一个名额的终局决策消费(was_last 门控防重复续跑) + let loop_gone = peek_loop_exited(&conv_id); + let was_last = hold.release(); + if was_last && loop_gone { + // loop 已收敛退出:正典续跑通道消费注入消息(iteration 累计透传, + // 对齐 ai_approve 续跑口径,防反复续跑跑满轮次预算) + let start_iter = { + let session = state.ai_session.lock().await; + session.conv_read(&conv_id).map(|c| c.iteration_used).unwrap_or(0) + }; + // Box 擦除内层 coroutine 类型:background_run → try_continue → run_agentic_loop + // → execute_with_heartbeat → spawn(background_run) 构成 Send 自动 trait 验证环, + // trait object 在此边截断(行为不变,仅类型层面)。 + let continued = Box::pin(try_continue_agent_loop(&app, &state, &conv_id, start_iter)) + as Pin + Send>>; + continued.await; + } + // 其余情况:loop 仍在飞/名额未尽,活跃 loop 下轮自然消费注入消息 + } + Ok(false) => { + // conv 不存在(已删/未建):仅释放名额,状态随删除链路清理 + hold.release(); + } + Err(_) => { + // 非活跃(用户停止 Stopping / 致命错误 Error):结果落 system 通知留档 + // (下轮打开会话可见),最后一个名额终局时放行占用复位 + hold.release(); + insert_system_notice( + &state.ai_session.clone(), &conv_id, ¬e, + "后台命令完成回填(会话非活跃,注入通道已关)", + ).await; + if !hold_in_flight(&conv_id) { + release_occupancy(&app, &conv_id).await; + } + } + } +} + +/// 终局放行:Generating/Stopping → Idle + 状态事件(对齐 guard.transition 的 +/// emit + 托盘同步);Error 保持原态(用户重试流程自管,结果已落 system 通知)。 +async fn release_occupancy(app: &AppHandle, conv_id: &str) { + let state = app.state::(); + let cur = state.conv_states.get(conv_id); + if !matches!(cur, ConvState::Generating | ConvState::Stopping) { + return; + } + if let Ok(next) = state.conv_states.transition(conv_id, ConvState::Idle) { + emit_ai_chat_event(app, AiChatEvent::AiConvStateChanged { + conv_state: next, + conversation_id: Some(conv_id.to_string()), + }); + crate::tray::update_tray_runtime(app); + } +} + +// ============================================================ +// 单测(纯逻辑部分;execute_with_heartbeat 集成路径依赖 AppHandle 无测点, +// 开关关=同步路径逐字节不变,由存量测试全绿作回归证明) +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + + // ── 非终态 tool_result 话术(ReAct 语义变更的核心载体,锁定文案) ── + + #[test] + fn payload_states_background_and_promises_callback() { + let p = background_payload(90); + assert_eq!(p["status"], "background"); + let msg = p["message"].as_str().unwrap(); + assert!(msg.contains("已转后台执行(超 90s)"), "实际: {msg}"); + assert!(msg.contains("完成后结果自动回传本会话"), "实际: {msg}"); + } + + #[test] + fn payload_carries_configured_grace() { + let p = background_payload(30); + assert!(p["message"].as_str().unwrap().contains("(超 30s)")); + } + + // ── grace 配置解析(默认 90 / 非法与 0 归默认 / 合法值采纳) ── + + #[test] + fn grace_defaults_when_missing_or_invalid() { + assert_eq!(parse_grace(None), 90); + assert_eq!(parse_grace(Some("abc".to_string())), 90); + assert_eq!(parse_grace(Some("0".to_string())), 90, "0 视为配置错误归默认"); + assert_eq!(parse_grace(Some("-5".to_string())), 90); + } + + #[test] + fn grace_accepts_valid_value() { + assert_eq!(parse_grace(Some("30".to_string())), 30); + assert_eq!(parse_grace(Some("600".to_string())), 600); + } + + // ── per-conv 占用登记(会话占用红线的注册表语义) ── + // 静态表进程级共享:各测用例用独立 conv key + 结束清理,防并行互踩 + + #[test] + fn hold_blocks_idle_only_while_in_flight() { + let key = "test-hold-basic-1"; + assert!(!hold_blocks_idle(key), "未登记不拦截"); + hold_acquire(key); + assert!(hold_blocks_idle(key), "在飞时拦截 Idle 迁移"); + assert!(hold_release(key), "单名额释放即最后"); + assert!(!hold_in_flight(key), "释放后无在飞"); + assert!(!hold_blocks_idle(key), "释放后不再拦截"); + holds().remove(key); + } + + #[test] + fn hold_release_reports_last_only() { + let key = "test-hold-last-2"; + hold_acquire(key); + hold_acquire(key); + assert!(!hold_release(key), "仍有 1 个在飞,非最后"); + assert!(hold_release(key), "最后一个释放返 true"); + assert!(!hold_in_flight(key)); + } + + #[test] + fn loop_exited_set_on_block_and_survives_until_release() { + let key = "test-hold-exit-3"; + hold_acquire(key); + assert!(!peek_loop_exited(key), "未拦截前置位应为 false"); + // loop 收敛退出经 guard 占用闸 → loop_exited 置位 + assert!(hold_blocks_idle(key)); + // 只读不消费:终局决策在 release 前读,多名额时标记保持供最后一个消费 + assert!(peek_loop_exited(key)); + assert!(peek_loop_exited(key), "peek 不消费,重复读恒 true"); + holds().remove(key); + } + + // ── grace 竞速(机制核心:窗口内完成=原路径结果一致;超时=future 可移交) ── + + #[tokio::test] + async fn race_done_within_window_matches_direct_result() { + let exec = Box::pin(async { + Ok::(serde_json::json!({"exit": 0, "out": "done"})) + }); + let (raced, _fut) = race_grace( + exec, + Duration::from_secs(5), + std::future::pending::(), + ).await; + match raced { + GraceRace::Done(Ok(v)) => assert_eq!(v, serde_json::json!({"exit": 0, "out": "done"})), + other => panic!("grace 内完成应返 Done 且与同步结果一致,实际 {:?}", result_tag(&other)), + } + } + + #[tokio::test] + async fn race_expired_hands_future_to_detached() { + let exec = Box::pin(async { + tokio::time::sleep(Duration::from_millis(200)).await; + Ok::(serde_json::json!({"late": true})) + }); + let (raced, mut fut) = race_grace( + exec, + Duration::from_millis(30), + std::future::pending::(), + ).await; + assert!(matches!(raced, GraceRace::Expired), "超时应返 Expired"); + // 移交语义:future 未被消费,spawn 后仍跑到终局(detached 启动的机制本质) + let handle = tokio::spawn(async move { fut.as_mut().await }); + let v = tokio::time::timeout(Duration::from_secs(10), handle).await + .expect("detached 续跑应在时限内完成").unwrap().unwrap(); + assert_eq!(v, serde_json::json!({"late": true})); + } + + #[tokio::test] + async fn race_watch_abort_wins_over_pending_exec() { + let exec = Box::pin(std::future::pending::>()); + let (raced, _fut) = race_grace( + exec, + Duration::from_secs(60), + std::future::ready("会话 1 已删除".to_string()), + ).await; + match raced { + GraceRace::Aborted(reason) => assert!(reason.contains("已删除")), + other => panic!("监视判死应返 Aborted,实际 {:?}", result_tag(&other)), + } + } + + /// 测试辅助:竞速结果打标签(GraceRace 未要求 Debug, panic 消息用) + fn result_tag(r: &GraceRace) -> &'static str { + match r { + GraceRace::Done(_) => "Done", + GraceRace::Expired => "Expired", + GraceRace::Aborted(_) => "Aborted", + } + } +} diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 6872f01..24301bc 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -24,6 +24,9 @@ use crate::commands::ai::is_destructive_command; mod utils; pub(crate) use utils::{risk_from_str, risk_str, truncate_chars_ellipsis}; +// bg_run(audit/bg_run.rs):run_command 超时转后台(KV 默认关)。 +pub(crate) mod bg_run; + // diff(audit/diff.rs):write_file 审批预览 diff 生成。 // 第五批从本文件抽离，行为零变更。 mod diff; @@ -243,8 +246,11 @@ fn sub_agent_tool_clamped(allowed: Option<&[String]>, tool_name: &str) -> bool { /// 其 heartbeat_loop 随作用域退出,慢工具(大文件 patch / spawn_agent / run_command 长构建) /// 整窗无心跳 → 前端看门狗误杀。复用同时获得 run_command 实时流式 sink、spawn_agent 父会话 /// task-local 注入与 outer_secs 分档超时(spawn_agent=挂钟预算+60 / run_command=timeout_secs clamp)。 +/// `tools` 形参为 `&Arc`:后台化移交需把工具 future 移入 +/// detached task('static),调用点本就持有 Arc(process_tool_calls 三处 clone 局部 + +/// chat.rs 审批/授权两窗口 `&state.ai_tools`),签名升级对调用方零改动。 pub(crate) async fn execute_with_heartbeat( - tools: &AiToolRegistry, + tools: &Arc, name: &str, args: serde_json::Value, app: &AppHandle, @@ -358,42 +364,11 @@ pub(crate) async fn execute_with_heartbeat( } else { 60 }; - // run_command 实时流式:经 task-local sink 注入(AppHandle + tool_call_id + conv_id), - // handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。 - // 非 run_command 不注入 sink(handler 内 emit_output 静默 noop),零行为变更。 - let sink = if name == "run_command" { - Some(crate::commands::ai::command_stream::CommandSink::new( - app.clone(), - tool_call_id.to_string(), - Some(conv_id.to_string()), - )) - } else { - None - }; - let exec_fut = crate::commands::ai::command_stream::scope(sink, tools.execute(name, args)); - // ask_uclaw 进度流式:u-claw agent_progress 块级(2s 节流)经独立 task-local sink 注入, - // handler 内 emit_progress_line 每块 emit AiCommandOutput(前端工具卡实时展示)。 - // 非 ask_uclaw 不注入(handler 内 emit_progress_line 静默 noop),零行为变更。 - // 与上方 command_stream 的 SINK / spawn_agent 的 CALLER_CONV 三 task-local 嵌套互不干扰。 - // 两分支都走 scope(else 包 None):if/else 类型一致(future 输出 R 相同,scope 包装后 - // 具体类型不同会 E0308,统一 None/Some 走同一泛型路径)。 - let uclaw_sink = if name == "ask_uclaw" { - Some(crate::commands::ai::tools::uclaw::UclawProgressSink::new( - app.clone(), - tool_call_id.to_string(), - Some(conv_id.to_string()), - )) - } else { - None - }; - let exec_fut = - crate::commands::ai::tools::uclaw::scope_progress_sink(uclaw_sink, exec_fut); - // 父会话 task-local 注入:spawn_agent handler 经 caller_conv() 定位发起调用的父会话 - // (直调防御兜底用)。与上方 command_stream 的 SINK scope 嵌套,两个 task-local 互不干扰。 - let exec_fut = crate::commands::ai::tools::spawn_agent::scope_caller_conv( - conv_id.to_string(), - exec_fut, - ); + // ── 后台化资格(run_command 专属,KV 默认关):None=同步原路径 ── + // 资格判定含主会话排除(子代理独立循环不转后台),详见 bg_run 模块文档。 + // 前置于 exec_fut 构造:后台路径须以 Arc 克隆 + name 所有权重建工具 future + // (原构造借用 &tools/&name,非 'static,无法移交 detached task)。 + let bg_grace = bg_run::background_grace_secs(app, name, conv_id).await; // ── 提前退出监视(防空转):会话已删 / 审批窗口失效时中止工具执行 ── // 进入快照:该 tool_call_id 是否仍在 pending_approvals。现网 4 个调用点(chat.rs ai_approve / @@ -401,55 +376,144 @@ pub(crate) async fn execute_with_heartbeat( // 路径从未 insert),进入时 pending 必无此 id——若按字面「不在 pending 即退出」实现,会把 // 所有已批准执行立即腰斩。故审批移除监视仅在「进入时仍在 pending」的调用形态下武装 // (现网不触发,为未来调用形态兜底);会话已删监视对所有调用点生效:执行中删除会话, - // 长工具最多再跑 2s 即中止,不再空转到 outer_secs 上限。 + // 长工具最多再跑 2s 即中止,不再空转到超时上限。 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 作用域消亡)。 - let exec_result = tokio::select! { - result = &mut exec_fut => result, - _ = tokio::time::sleep(Duration::from_secs(outer_secs)) => { - tracing::error!( - conv_id = %conv_id, - tool = %name, - outer_secs, - "[ai] 工具执行超时({}s),返回错误 tool_result(防 process_tool_calls 持 session lock 永久卡死)", - 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)) + let exec_result = if let Some(grace) = bg_grace { + // ── 后台路径:grace 窗口三路 select(窗口内完成 / 超时移交 / 监视判死)── + // owned 'static future:Arc 克隆 + name/args 所有权重建(execute 语义不变: + // 查表 + handler 调用);command_stream sink 注入保留——后台期实时流式仍进原工具卡。 + // 窗口内完成与同步路径同语义(含监视);超时把未消费 future 移交 detached task, + // 本函数立即返非终态 tool_result(「已转后台」),loop 与前端不再被该命令阻塞。 + // outer_secs 在此被 grace 取代:后台期命令受工具内部 timeout_secs 约束,不再需要 + // 外层硬腰斩(移交后外层腰斩语义无从谈起);监视判死仍生效(会话删除即中止)。 + let tools_owned = tools.clone(); + let name_owned = name.to_string(); + let args_owned = args.clone(); + let sink = crate::commands::ai::command_stream::CommandSink::new( + app.clone(), + tool_call_id.to_string(), + Some(conv_id.to_string()), + ); + let exec = Box::pin(crate::commands::ai::tools::spawn_agent::scope_caller_conv( + conv_id.to_string(), + crate::commands::ai::command_stream::scope( + Some(sink), + async move { tools_owned.execute(&name_owned, args_owned).await }, + ), + )); + let watch = abort_when_invalid(app.clone(), conv_id.to_string(), tc_watch, was_pending_at_entry); + let (raced, exec) = + bg_run::race_grace(exec, Duration::from_secs(grace), watch).await; + match raced { + bg_run::GraceRace::Done(r) => r, + bg_run::GraceRace::Expired => { + tracing::info!( + conv_id = %conv_id, + tool = %name, + tool_call_id = %tool_call_id, + grace, + "[ai] run_command 超 grace 窗口,移交后台执行(LLM 收非终态 tool_result)" + ); + // 存活心跳证据源:detached 续写 per_conv.last_heartbeat(ai_chat_stop + // 兜底判活同源);conv 已删传 None(bg_run 给孤儿 Arc 仅保类型) + let detached_hb = { + let st = app.state::(); + let session = st.ai_session.lock().await; + session.conv_read(conv_id).map(|c| c.last_heartbeat.clone()) + }; + bg_run::spawn_detached( + exec, + name.to_string(), + args_for_memory, + app.clone(), + conv_id.to_string(), + tool_call_id.to_string(), + detached_hb, + ); + return Ok(bg_run::background_payload(grace)); + } + bg_run::GraceRace::Aborted(reason) => { + tracing::warn!( + conv_id = %conv_id, + tool = %name, + tool_call_id = %tool_call_id, + "[ai] execute_with_heartbeat 提前退出: {reason}(防空转)" + ); + Err(anyhow::anyhow!("工具 {} 提前中止: {}", name, reason)) + } } + } else { + // ── 同步原路径(构造段 + select 与历史版本逐字一致)── + // run_command 实时流式:经 task-local sink 注入(AppHandle + tool_call_id + conv_id), + // handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。 + // 非 run_command 不注入 sink(handler 内 emit_output 静默 noop),零行为变更。 + let sink = if name == "run_command" { + Some(crate::commands::ai::command_stream::CommandSink::new( + app.clone(), + tool_call_id.to_string(), + Some(conv_id.to_string()), + )) + } else { + None + }; + let exec_fut = crate::commands::ai::command_stream::scope(sink, tools.execute(name, args)); + // ask_uclaw 进度流式:u-claw agent_progress 块级(2s 节流)经独立 task-local sink 注入, + // handler 内 emit_progress_line 每块 emit AiCommandOutput(前端工具卡实时展示)。 + // 非 ask_uclaw 不注入(handler 内 emit_progress_line 静默 noop),零行为变更。 + // 与上方 command_stream 的 SINK / spawn_agent 的 CALLER_CONV 三 task-local 嵌套互不干扰。 + // 两分支都走 scope(else 包 None):if/else 类型一致(future 输出 R 相同,scope 包装后 + // 具体类型不同会 E0308,统一 None/Some 走同一泛型路径)。 + let uclaw_sink = if name == "ask_uclaw" { + Some(crate::commands::ai::tools::uclaw::UclawProgressSink::new( + app.clone(), + tool_call_id.to_string(), + Some(conv_id.to_string()), + )) + } else { + None + }; + let exec_fut = + crate::commands::ai::tools::uclaw::scope_progress_sink(uclaw_sink, exec_fut); + // 父会话 task-local 注入:spawn_agent handler 经 caller_conv() 定位发起调用的父会话 + // (直调防御兜底用)。与上方 command_stream 的 SINK scope 嵌套,两个 task-local 互不干扰。 + let exec_fut = crate::commands::ai::tools::spawn_agent::scope_caller_conv( + conv_id.to_string(), + exec_fut, + ); + let abort_watch = abort_when_invalid(app.clone(), conv_id.to_string(), tc_watch, was_pending_at_entry); + tokio::pin!(exec_fut); + // 三路 select:exec 正常完成 / outer_secs 兜底超时(既有语义,drop exec 腰斩) / + // 监视判死提前退出(同样 drop exec,防审批被删或会话已删后空转到超时上限)。 + // exec 先完成时 select 丢弃 abort_watch,监视零泄漏(非 spawn,随 select 作用域消亡)。 + let exec_result = tokio::select! { + result = &mut exec_fut => result, + _ = tokio::time::sleep(Duration::from_secs(outer_secs)) => { + tracing::error!( + conv_id = %conv_id, + tool = %name, + outer_secs, + "[ai] 工具执行超时({}s),返回错误 tool_result(防 process_tool_calls 持 session lock 永久卡死)", + 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)) + } + }; + exec_result }; // ── B5d 工具失败记忆:执行后回写(成功清零 / 失败递增并记最近错误) ── @@ -463,6 +527,37 @@ pub(crate) async fn execute_with_heartbeat( exec_result } +/// 提前退出监视(防空转):会话已删 / 审批窗口失效时中止工具执行。 +/// 工厂形态:同步路径与后台 grace 路径各消费一个实例(同一快照语义)。 +/// +/// 进入快照:该 tool_call_id 是否仍在 pending_approvals。现网 4 个调用点(chat.rs ai_approve / +/// ai_authorize_dir 两个审批执行窗口均「先 remove 后 execute」,本文件 low/trust 两条自动执行 +/// 路径从未 insert),进入时 pending 必无此 id——若按字面「不在 pending 即退出」实现,会把 +/// 所有已批准执行立即腰斩。故审批移除监视仅在「进入时仍在 pending」的调用形态下武装 +/// (现网不触发,为未来调用形态兜底);会话已删监视对所有调用点生效:执行中删除会话, +/// 长工具最多再跑 2s 即中止,不再空转到超时上限。 +fn abort_when_invalid( + app: AppHandle, + conv_id: String, + tool_call_id: String, + was_pending_at_entry: bool, +) -> impl std::future::Future { + async move { + loop { + // 2s 轮询:短锁纯读(锁不跨 await 持有);中止延迟 ≤2s 与锁竞争频率的权衡 + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + let st = app.state::(); + let session = st.ai_session.lock().await; + if session.deleted_convs.contains(&conv_id) { + return format!("会话 {} 已删除", conv_id); + } + if was_pending_at_entry && !session.pending_approvals.contains_key(&tool_call_id) { + return format!("审批 {} 已不在挂起表", tool_call_id); + } + } + } +} + // ============================================================ // 审批超时主动收敛(定时器) — 治「审批过期后用户不操作则永挂」 // ============================================================ diff --git a/src-tauri/src/feishu_channel.rs b/src-tauri/src/feishu_channel.rs index 51a6591..291bc16 100644 --- a/src-tauri/src/feishu_channel.rs +++ b/src-tauri/src/feishu_channel.rs @@ -117,6 +117,9 @@ struct Ctx { chat_map: Arc>>, /// per-chat 串行锁:同会话消息排队处理,防通知桥单槽跨消息串扰(P1-1) chat_locks: Arc>>>>, + /// 防抖聚合窗:chat_id → 窗内追加消息蓄积。首条登记,追加并入即返回, + /// 窗满首条取走统一受理;生命周期同 chat_locks(渠道重启随 Ctx 重建) + debounce: Arc>>, /// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流 last_alive: Arc, /// D3 在途会话数:消息受理(handle_message 白名单通过)到任务投递完成期间 >0, @@ -176,6 +179,7 @@ pub fn start_if_enabled(app: tauri::AppHandle) { whitelist: Arc::new(tokio::sync::RwLock::new(cfg.whitelist)), chat_map: Arc::new(tokio::sync::Mutex::new(restored)), chat_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + debounce: Arc::new(tokio::sync::Mutex::new(HashMap::new())), last_alive: Arc::new(AtomicI64::new(now_secs())), inflight: Arc::new(AtomicUsize::new(0)), }); @@ -469,7 +473,115 @@ fn sniff_image_mime(b: &[u8]) -> Option<&'static str> { } } -/// 入站文本:群聊需 @bot;白名单校验;/new 重置;其余走 ai-chat 会话链路 +/// 防抖窗口(拍板 0.5s):同 chat 连发消息的合并等待时长 +const DEBOUNCE_WINDOW: std::time::Duration = std::time::Duration::from_millis(500); + +/// 防抖窗内一条 chat 的追加消息蓄积。首条原文/图片不入此(留在本地,窗满时合并), +/// 结构上保证窗满取走必得 entry 且最坏退化为单条,消息不丢 +#[derive(Default)] +struct DebounceEntry { + /// 追加消息合并文本(空行拼接) + text: String, + /// 追加消息图片片(到达序合并) + parts: Option>, + /// 已追加条数(不含首条) + appended: usize, +} + +impl DebounceEntry { + /// 追加一条进窗:文本空行拼接、图片片到达序合并、计数累加 + fn append(&mut self, text: &str, parts: Option>) { + self.text = merge_debounce_text(&self.text, text); + self.parts = merge_debounce_parts(self.parts.take(), parts); + self.appended += 1; + } +} + +/// 防抖合并纯函数:文本空行拼接(任一侧为空直取另一侧,不产首尾空行) +fn merge_debounce_text(base: &str, next: &str) -> String { + if base.is_empty() { + next.to_string() + } else if next.is_empty() { + base.to_string() + } else { + format!("{base}\n\n{next}") + } +} + +/// 防抖合并纯函数:图片片按到达序拼接(单侧 None 直取另一侧) +fn merge_debounce_parts( + base: Option>, + next: Option>, +) -> Option> { + match (base, next) { + (Some(mut a), Some(b)) => { + a.extend(b); + Some(a) + } + (a, b) => a.or(b), + } +} + +/// 窗满合并结果(首条视角):合并后任务载荷+占位卡句柄,交回 handle_message 续原流程 +struct MergedIncoming { + text: String, + parts: Option>, + placeholder_id: Option, +} + +/// 连发消息防抖聚合(u-claw 同款)。防抖点在注入旁路之后:命令/审批/注入即时路径不进窗, +/// 白名单与图片下载已在前逐条完成。追加消息并入 entry 即返回 None(调用方直接收尾, +/// 不回卡防刷屏——首条占位卡承载合并提示,不取 per-chat 锁不 spawn,受理计数随 RAII +/// 回落);首条登记空 entry 后回占位卡(文案不变)等窗,窗满取走 entry 返回合并结果, +/// 有追加时占位卡原地 PATCH 合并提示。sleep 全程不持任何锁;取走与追加的互斥由 +/// DebounceMap 锁保证,窗满后到达的消息查 None 自成新窗,自洽 +async fn debounce_collect( + ctx: &Ctx, + chat_id: &str, + message_id: &str, + text: String, + parts: Option>, +) -> Option { + { + let mut map = ctx.debounce.lock().await; + if let Some(entry) = map.get_mut(chat_id) { + entry.append(&text, parts); + return None; + } + map.insert(chat_id.to_string(), DebounceEntry::default()); + } + // 立即确认(交互丝滑):占位单行卡片原地更新为结果,占位与结果合一不刷屏; + // reply 失败置空,完成路径回退新发卡片。带任务原文回显,让用户在到受理卡的 + // 无内容可对证期即时确认消息被正确接收;摘要先取(text 随后参与合并 move) + let placeholder_text = format!("🤔 已接收,处理中…\n任务:{}", truncate(&text, 60)); + let placeholder_id = + reply_markdown_card(&ctx.client, message_id, &placeholder_text).await; + tokio::time::sleep(DEBOUNCE_WINDOW).await; + // 取走窗内蓄积。entry 仅本首条插入、追加只改不摘,None 理论不可达,兜底退化为单条 + let entry = ctx + .debounce + .lock() + .await + .remove(chat_id) + .unwrap_or_default(); + if entry.appended > 0 { + deliver_result( + &ctx.client, + &placeholder_id, + message_id, + chat_id, + &format!("📥 已合并 {} 条消息,开始处理", entry.appended + 1), + ) + .await; + } + Some(MergedIncoming { + text: merge_debounce_text(&text, &entry.text), + parts: merge_debounce_parts(parts, entry.parts), + placeholder_id, + }) +} + +/// 入站文本:群聊需 @bot;白名单校验;/new 重置;连发消息防抖合并;其余走 ai-chat 会话链路 async fn handle_message( ctx: Arc, msg: larksuite_oapi_sdk_rs::channel::NormalizedMessage, @@ -740,14 +852,15 @@ async fn handle_message( // chat_id → conv_id 映射(内存命中优先,未命中新建) feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80))); - // 立即确认(交互丝滑):占位改用单行卡片并记下其 message_id,完成后原地更新为结果, - // 占位与结果合一不刷屏;占位 reply 失败则置空,完成路径回退现状(新发卡片) - // 2026-10-08 占位卡带任务原文回显:原固定「🤔 已接收,处理中…」用户不知在做什么, - // 到受理卡之间(per-chat 锁排队 + spawn)有一段无内容可对证期;回显摘要让用户即时确认 - // 消息已被正确接收。text 此后会被 move 进 spawn_agentic_session,此处先取摘要 - let placeholder_text = format!("🤔 已接收,处理中…\n任务:{}", truncate(&text, 60)); - let placeholder_id = - reply_markdown_card(&ctx.client, &message_id, &placeholder_text).await; + + // ── 连续消息防抖聚合(u-claw 同款):同 chat 0.5s 窗内连发合并为一次任务 ── + // 追加消息在此收尾(None),窗满首条携合并载荷继续下方原流程(per-chat 锁→conv + // 映射→受理守卫→spawn/959) + let Some(merged) = debounce_collect(&ctx, &chat_id, &message_id, text, parts).await else { + return; + }; + let MergedIncoming { text, parts, placeholder_id } = merged; + // per-chat 串行:同会话消息排队(防通知桥单槽被后条覆盖致前条串扰) let chat_lock = { let mut locks = ctx.chat_locks.lock().await; @@ -2288,9 +2401,10 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap