新增: 连发消息防抖与长命令后台化

- 飞书连发消息 0.5s 窗(拍板值)合并为一次任务:文本空行拼接+图片到达序合并
- 命令/审批/注入即时路径不进窗;追加消息并入即静默返回;窗满统一受理,守卫拒绝落待发箱
- run_command 超 grace 窗(默认 90s,可配)转后台:LLM 收非终态结果,不再整轮阻塞
- 后台任务自持心跳+删除自中止+失败记忆+完成通知+回填续跑;占用闸单点保会话态自洽
- 开关默认关=同步路径逐字节不变;新增 13 测
This commit is contained in:
lxy
2026-10-11 12:37:23 +08:00
parent fda991cbe1
commit c7d0ed939c
4 changed files with 894 additions and 90 deletions
@@ -83,6 +83,15 @@ impl GeneratingGuard {
/// 迁移 ConvState + emit AiConvStateChanged(经无锁 ConvStateStore,同步无 await)。 /// 迁移 ConvState + emit AiConvStateChanged(经无锁 ConvStateStore,同步无 await)。
fn transition(&self, target: ConvState) { 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 let conv_states = self
.app_handle .app_handle
.state::<crate::state::AppState>() .state::<crate::state::AppState>()
+539
View File
@@ -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<u64> {
if name != "run_command" {
return None;
}
let state = app.state::<AppState>();
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<String>) -> u64 {
raw.and_then(|s| s.parse::<u64>().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<DashMap<String, BgHold>> = OnceLock::new();
fn holds() -> &'static DashMap<String, BgHold> {
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<R> {
/// 窗口内完成:与同步原路径结果完全一致
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<R, W>(
exec: Pin<Box<dyn Future<Output = R> + Send>>,
grace: Duration,
watch: W,
) -> (GraceRace<R>, Pin<Box<dyn Future<Output = R> + Send>>)
where
W: Future<Output = String>,
{
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<Box<dyn Future<Output = anyhow::Result<serde_json::Value>> + Send>>,
name: String,
args_for_memory: serde_json::Value,
app: AppHandle,
conv_id: String,
tool_call_id: String,
last_heartbeat: Option<Arc<AtomicU64>>,
) {
// 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<Box<dyn Future<Output = anyhow::Result<serde_json::Value>> + Send>>,
name: String,
args_for_memory: serde_json::Value,
app: AppHandle,
conv_id: String,
tool_call_id: String,
last_heartbeat: Arc<AtomicU64>,
) {
// 接管 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::<AppState>();
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::<AppState>();
// 先注入后释放名额:占用期 is_active 恒真(Generating),注入守卫全场景可过;
// 先释放会在「名额清零且 loop 已退」窗口露出非活跃态,注入误走降级分支
match inject_runtime_message(&state, &conv_id, &note).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<Box<dyn Future<Output = ()> + 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, &note,
"后台命令完成回填(会话非活跃,注入通道已关)",
).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::<AppState>();
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::Value, anyhow::Error>(serde_json::json!({"exit": 0, "out": "done"}))
});
let (raced, _fut) = race_grace(
exec,
Duration::from_secs(5),
std::future::pending::<String>(),
).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::Value, anyhow::Error>(serde_json::json!({"late": true}))
});
let (raced, mut fut) = race_grace(
exec,
Duration::from_millis(30),
std::future::pending::<String>(),
).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::<Result<serde_json::Value, anyhow::Error>>());
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>(r: &GraceRace<R>) -> &'static str {
match r {
GraceRace::Done(_) => "Done",
GraceRace::Expired => "Expired",
GraceRace::Aborted(_) => "Aborted",
}
}
}
+127 -32
View File
@@ -24,6 +24,9 @@ use crate::commands::ai::is_destructive_command;
mod utils; mod utils;
pub(crate) use utils::{risk_from_str, risk_str, truncate_chars_ellipsis}; 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 生成。 // diff(audit/diff.rs):write_file 审批预览 diff 生成。
// 第五批从本文件抽离,行为零变更。 // 第五批从本文件抽离,行为零变更。
mod 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 长构建) /// 其 heartbeat_loop 随作用域退出,慢工具(大文件 patch / spawn_agent / run_command 长构建)
/// 整窗无心跳 → 前端看门狗误杀。复用同时获得 run_command 实时流式 sink、spawn_agent 父会话 /// 整窗无心跳 → 前端看门狗误杀。复用同时获得 run_command 实时流式 sink、spawn_agent 父会话
/// task-local 注入与 outer_secs 分档超时(spawn_agent=挂钟预算+60 / run_command=timeout_secs clamp)。 /// task-local 注入与 outer_secs 分档超时(spawn_agent=挂钟预算+60 / run_command=timeout_secs clamp)。
/// `tools` 形参为 `&Arc<AiToolRegistry>`:后台化移交需把工具 future 移入
/// detached task('static),调用点本就持有 Arc(process_tool_calls 三处 clone 局部 +
/// chat.rs 审批/授权两窗口 `&state.ai_tools`),签名升级对调用方零改动。
pub(crate) async fn execute_with_heartbeat( pub(crate) async fn execute_with_heartbeat(
tools: &AiToolRegistry, tools: &Arc<AiToolRegistry>,
name: &str, name: &str,
args: serde_json::Value, args: serde_json::Value,
app: &AppHandle, app: &AppHandle,
@@ -358,6 +364,92 @@ pub(crate) async fn execute_with_heartbeat(
} else { } else {
60 60
}; };
// ── 后台化资格(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 /
// ai_authorize_dir 两个审批执行窗口均「先 remove 后 execute」,本文件 low/trust 两条自动执行
// 路径从未 insert),进入时 pending 必无此 id——若按字面「不在 pending 即退出」实现,会把
// 所有已批准执行立即腰斩。故审批移除监视仅在「进入时仍在 pending」的调用形态下武装
// (现网不触发,为未来调用形态兜底);会话已删监视对所有调用点生效:执行中删除会话,
// 长工具最多再跑 2s 即中止,不再空转到超时上限。
let tc_watch = tool_call_id.to_string();
let was_pending_at_entry = {
let st = app.state::<crate::state::AppState>();
let session = st.ai_session.lock().await;
session.pending_approvals.contains_key(&tc_watch)
};
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::<crate::state::AppState>();
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), // run_command 实时流式:经 task-local sink 注入(AppHandle + tool_call_id + conv_id),
// handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。 // handler 内 execute_streaming 每行 emit AiCommandOutput(治执行中黑盒)。
// 非 run_command 不注入 sink(handler 内 emit_output 静默 noop),零行为变更。 // 非 run_command 不注入 sink(handler 内 emit_output 静默 noop),零行为变更。
@@ -394,37 +486,7 @@ pub(crate) async fn execute_with_heartbeat(
conv_id.to_string(), conv_id.to_string(),
exec_fut, exec_fut,
); );
let abort_watch = abort_when_invalid(app.clone(), conv_id.to_string(), tc_watch, was_pending_at_entry);
// ── 提前退出监视(防空转):会话已删 / 审批窗口失效时中止工具执行 ──
// 进入快照:该 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::<crate::state::AppState>();
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::<crate::state::AppState>();
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); tokio::pin!(exec_fut);
// 三路 select:exec 正常完成 / outer_secs 兜底超时(既有语义,drop exec 腰斩) / // 三路 select:exec 正常完成 / outer_secs 兜底超时(既有语义,drop exec 腰斩) /
// 监视判死提前退出(同样 drop exec,防审批被删或会话已删后空转到超时上限)。 // 监视判死提前退出(同样 drop exec,防审批被删或会话已删后空转到超时上限)。
@@ -451,6 +513,8 @@ pub(crate) async fn execute_with_heartbeat(
Err(anyhow::anyhow!("工具 {} 提前中止: {}", name, reason)) Err(anyhow::anyhow!("工具 {} 提前中止: {}", name, reason))
} }
}; };
exec_result
};
// ── B5d 工具失败记忆:执行后回写(成功清零 / 失败递增并记最近错误) ── // ── B5d 工具失败记忆:执行后回写(成功清零 / 失败递增并记最近错误) ──
// 与入口 precheck 配对:成功即删记忆(知识保鲜,防误杀);失败累加计数, // 与入口 precheck 配对:成功即删记忆(知识保鲜,防误杀);失败累加计数,
@@ -463,6 +527,37 @@ pub(crate) async fn execute_with_heartbeat(
exec_result 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<Output = String> {
async move {
loop {
// 2s 轮询:短锁纯读(锁不跨 await 持有);中止延迟 ≤2s 与锁竞争频率的权衡
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let st = app.state::<crate::state::AppState>();
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);
}
}
}
}
// ============================================================ // ============================================================
// 审批超时主动收敛(定时器) — 治「审批过期后用户不操作则永挂」 // 审批超时主动收敛(定时器) — 治「审批过期后用户不操作则永挂」
// ============================================================ // ============================================================
+173 -12
View File
@@ -117,6 +117,9 @@ struct Ctx {
chat_map: Arc<tokio::sync::Mutex<HashMap<String, String>>>, chat_map: Arc<tokio::sync::Mutex<HashMap<String, String>>>,
/// per-chat 串行锁:同会话消息排队处理,防通知桥单槽跨消息串扰(P1-1) /// per-chat 串行锁:同会话消息排队处理,防通知桥单槽跨消息串扰(P1-1)
chat_locks: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>, chat_locks: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
/// 防抖聚合窗:chat_id → 窗内追加消息蓄积。首条登记,追加并入即返回,
/// 窗满首条取走统一受理;生命周期同 chat_locks(渠道重启随 Ctx 重建)
debounce: Arc<tokio::sync::Mutex<HashMap<String, DebounceEntry>>>,
/// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流 /// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流
last_alive: Arc<AtomicI64>, last_alive: Arc<AtomicI64>,
/// D3 在途会话数:消息受理(handle_message 白名单通过)到任务投递完成期间 >0, /// 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)), whitelist: Arc::new(tokio::sync::RwLock::new(cfg.whitelist)),
chat_map: Arc::new(tokio::sync::Mutex::new(restored)), chat_map: Arc::new(tokio::sync::Mutex::new(restored)),
chat_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())), 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())), last_alive: Arc::new(AtomicI64::new(now_secs())),
inflight: Arc::new(AtomicUsize::new(0)), 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<Vec<ContentPart>>,
/// 已追加条数(不含首条)
appended: usize,
}
impl DebounceEntry {
/// 追加一条进窗:文本空行拼接、图片片到达序合并、计数累加
fn append(&mut self, text: &str, parts: Option<Vec<ContentPart>>) {
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<Vec<ContentPart>>,
next: Option<Vec<ContentPart>>,
) -> Option<Vec<ContentPart>> {
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<Vec<ContentPart>>,
placeholder_id: Option<String>,
}
/// 连发消息防抖聚合(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<Vec<ContentPart>>,
) -> Option<MergedIncoming> {
{
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( async fn handle_message(
ctx: Arc<Ctx>, ctx: Arc<Ctx>,
msg: larksuite_oapi_sdk_rs::channel::NormalizedMessage, msg: larksuite_oapi_sdk_rs::channel::NormalizedMessage,
@@ -740,14 +852,15 @@ async fn handle_message(
// chat_id → conv_id 映射(内存命中优先,未命中新建) // chat_id → conv_id 映射(内存命中优先,未命中新建)
feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80))); feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80)));
// 立即确认(交互丝滑):占位改用单行卡片并记下其 message_id,完成后原地更新为结果,
// 占位与结果合一不刷屏;占位 reply 失败则置空,完成路径回退现状(新发卡片) // ── 连续消息防抖聚合(u-claw 同款):同 chat 0.5s 窗内连发合并为一次任务 ──
// 2026-10-08 占位卡带任务原文回显:原固定「🤔 已接收,处理中…」用户不知在做什么, // 追加消息在此收尾(None),窗满首条携合并载荷继续下方原流程(per-chat 锁→conv
// 到受理卡之间(per-chat 锁排队 + spawn)有一段无内容可对证期;回显摘要让用户即时确认 // 映射→受理守卫→spawn/959)
// 消息已被正确接收。text 此后会被 move 进 spawn_agentic_session,此处先取摘要 let Some(merged) = debounce_collect(&ctx, &chat_id, &message_id, text, parts).await else {
let placeholder_text = format!("🤔 已接收,处理中…\n任务:{}", truncate(&text, 60)); return;
let placeholder_id = };
reply_markdown_card(&ctx.client, &message_id, &placeholder_text).await; let MergedIncoming { text, parts, placeholder_id } = merged;
// per-chat 串行:同会话消息排队(防通知桥单槽被后条覆盖致前条串扰) // per-chat 串行:同会话消息排队(防通知桥单槽被后条覆盖致前条串扰)
let chat_lock = { let chat_lock = {
let mut locks = ctx.chat_locks.lock().await; let mut locks = ctx.chat_locks.lock().await;
@@ -2288,9 +2401,10 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap<String, Stri
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{ use super::{
delivery_blocked, extract_delivery_paths, feishu_file_type, heartbeat_due, ContentPart, DebounceEntry, delivery_blocked, extract_delivery_paths, feishu_file_type,
is_terminal_push, md_to_lark_friendly, msg_after_acceptance, parse_command, heartbeat_due, is_terminal_push, md_to_lark_friendly, merge_debounce_parts,
sniff_image_mime, strip_mention_placeholders, stream_preview, merge_debounce_text, msg_after_acceptance, parse_command, sniff_image_mime,
strip_mention_placeholders, stream_preview,
}; };
#[test] #[test]
@@ -2542,4 +2656,51 @@ mod tests {
assert!(!is_terminal_push("⏳ 进行中")); assert!(!is_terminal_push("⏳ 进行中"));
assert!(!is_terminal_push("普通进度行")); assert!(!is_terminal_push("普通进度行"));
} }
// ── 防抖合并 ──
#[test]
fn debounce_text_merges_with_blank_line() {
// 多文本空行拼接;空侧直取另一侧不产首尾空行;追加路径逐条 append 的链式形态
assert_eq!(merge_debounce_text("第一段", "第二段"), "第一段\n\n第二段");
assert_eq!(merge_debounce_text("", "第二段"), "第二段");
assert_eq!(merge_debounce_text("第一段", ""), "第一段");
let mut merged = merge_debounce_text("a", "b");
merged = merge_debounce_text(&merged, "c");
assert_eq!(merged, "a\n\nb\n\nc");
}
#[test]
fn debounce_parts_merge_in_arrival_order() {
// 图片片按消息到达序合并;单侧 None 直取另一侧;双 None 保持 None
let img1 = ContentPart::image_url("https://x/1.png");
let img2 = ContentPart::image_url("https://x/2.png");
let merged = merge_debounce_parts(
Some(vec![ContentPart::text("图1说明"), img1.clone()]),
Some(vec![img2.clone()]),
);
assert_eq!(merged, Some(vec![ContentPart::text("图1说明"), img1, img2.clone()]));
assert_eq!(
merge_debounce_parts(None, Some(vec![img2.clone()])),
Some(vec![img2.clone()])
);
assert_eq!(merge_debounce_parts(Some(vec![]), None), Some(vec![]));
assert_eq!(merge_debounce_parts(None, None), None);
}
#[test]
fn debounce_entry_counts_appends() {
// N 计数只含追加(首条不入 entry):3 条追加=合并 4 条受理;文本与图片同步蓄积
let mut entry = DebounceEntry::default();
assert_eq!(entry.appended, 0);
entry.append("第二条", None);
entry.append("第三条", Some(vec![ContentPart::image_url("https://x/2.png")]));
entry.append("第四条", None);
assert_eq!(entry.appended, 3);
assert_eq!(entry.text, "第二条\n\n第三条\n\n第四条");
assert_eq!(
entry.parts,
Some(vec![ContentPart::image_url("https://x/2.png")])
);
}
} }