lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复: 模型停用无效根因 + 上下文归位/压缩守卫

Browse Source 
- 模型停用无效根因: get_active_provider 两条路径均不校验 enabled(显式活跃直返/默认只挑 is_default) → 收敛为启用集合解析,无启用 provider 明确报错
- 压缩/标题兜底模型不校验 enabled → 新增 resolve_fallback_model 纯函数(title 复用同一实现,单一源无双轨)
- 上下文多夹层归位修复: gather_tool_responses 聚拢该头全部响应(此前第二个夹层之后的响应仍被判未闭合剥除)
- 压缩后台任务加内容版本哨兵(取首条消息 id,否掉 len:压缩期间 loop 正常 append 会致守卫永不生效) + panic 兜底复位标志
- 902 超时类失败与其它 Fatal 区分停靠文案(点继续可续跑)
- 子代理事件按 conv 改挂(route_scope) + 运行期事件推父会话视图
- 飞书 13a 真根因: run_stream_refresher 死代码(提前 drop)+静默钟被心跳重置 → 帧通道+仅内容事件重置
- 项目详情文件查看: F22 目录不存在误显非 Git 仓库 / F24 模块为仓库子目录时 git 徽标全失效(scope_git_path 按 show-prefix 剥前缀)
- 更新徽章通道错配(Tauri 事件 vs window CustomEvent 同名不同通道,徽章此前永不触发)
- ProjectCard 卡片根 keydown 顶替内嵌按钮(补 e.target 守卫)
- i18n 坏键 ai.aiTool.rejectedHint → aiTool.rejectedHint(双缺会渲染裸 key)
- miniapp 三处: 会话列表刷新漏网/看门狗节流致新一轮无兜底/newConversation 漏清看门狗
This commit is contained in:
lxy
2026-10-10 08:08:48 +08:00
parent 2a3a6cd25b
commit 9c68f98d89
24 changed files with 1023 additions and 3644 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/feishu_channel.rs
+134 -97
View File
@@ -630,11 +630,10 @@ async fn handle_message(
let (tx, _rx) = tokio::sync::oneshot::channel::<String>();
state.register_agent_completion(conv_id.clone(), tx);
// 流式预览:订阅既有 ai_event_bus 拿本 conv 的 AiTextDelta 增量(emit_ai_chat_event
// 双写总线已接入,渠道侧纯消费,零改动接线)。spawn 前订阅防早期增量丢失;
// 终版文本经 deliver_tx 也交给本任务单点投递,占位卡片自 spawn 起只被这一个任务写,
// 中间态与终版写卡天然串行,无迟到中间态覆盖终版的竞态
let (deliver_tx, deliver_rx) = tokio::sync::oneshot::channel::<String>();
// 过程可见性(13a):预览任务订阅既有 ai_event_bus 拿本 conv 的增量(delta/轮次/工具/
// 进度)节流刷新占位卡;受理卡与终版由完成订阅任务经帧通道投递。本任务为占位卡唯一
// 写者——中间态与终版同任务串行,无迟到中间态覆盖终版的竞态。spawn 前订阅防早期增量丢失
let (frame_tx, frame_rx) = tokio::sync::mpsc::channel::<CardFrame>(8);
let delta_rx = state.ai_event_bus.subscribe();
let stream_task = tauri::async_runtime::spawn(run_stream_refresher(
ctx.clone(),
@@ -643,7 +642,7 @@ async fn handle_message(
message_id.clone(),
chat_id.clone(),
delta_rx,
deliver_rx,
frame_rx,
));
let task_brief = truncate(&text, 60); // move 前先取展示摘要(spawn 会 move text)
@@ -669,9 +668,9 @@ async fn handle_message(
Err(e) => {
// 启动失败:通知桥条目清掉防死条目(编排节点同款)
state.notify_agent_completion(&conv_id, String::new());
// 关终版通道让流式任务退出(此路无任何增量产出,卡片未被流式写过),
// 关帧通道让预览任务退出(此路无任何帧产出,卡片未被写过),
// 待其退出后再走失败投递,保卡片写入单点次序
drop(deliver_tx);
drop(frame_tx);
let _ = stream_task.await;
feishu_log(&format!("spawn 失败 conv={conv_id}: {e}"));
// 18 渠道错误话术透传:并发护栏拒绝(生成中)≠真失败——固定「启动失败」话术
@@ -701,11 +700,10 @@ async fn handle_message(
// 消息处理 30 分钟,后续消息(含注入旁路)即时响应。终版投递改由 completion_watcher
// 消费 ai_event_bus 的 AiAgentFinished 广播(引擎层 deliver_completion_report_if_waited
// 统一发出,单一事实源),原地 edit_card 占位卡为结果。
// 卡片所有权移交(走查🔴#1 修正):占位卡自此只被 watcher 任务写——受理卡与终版
// 都在 watcher 单任务内按序落卡(受理卡先写,快任务兜底命中时跳过受理卡直投终版),
// 彻底消除「受理卡与终版两个 HTTP 请求网络乱序,终版被受理卡覆盖丢失」的竞态窗口。
drop(deliver_tx);
let _ = stream_task.await; // 等流式任务退出(快,只走通道关闭分支)
// 卡片所有权(13a 修订):占位卡唯一写者是预览任务(run_stream_refresher),受理卡与
// 终版都由本完成订阅任务经帧通道交给它按序落卡;预览任务内部串行,消除「受理卡与终版
// 两个 HTTP 请求网络乱序,终版被受理卡覆盖丢失」的竞态窗口。13a 前独立预览任务在受理
// 即被关停(drop 帧通道),心跳/阶段进展形同虚设;此处交回帧通道驱动使其真正落到卡片。
let ack = format!("🚀 已受理,后台执行中\n任务:{task_brief}");
feishu_log(&format!("受理返回 chat={chat_id} conv={conv_id}"));
@@ -723,12 +721,11 @@ async fn handle_message(
run_completion_watcher(
ctx,
conv_id,
placeholder_id,
message_id,
chat_id,
ack,
inflight_guard,
cancel_rx,
frame_tx,
stream_task,
)
.await;
// watcher 任意路径退出都摘除取消句柄,防陈旧条目驻留(后续 /cancel 空发无害)
@@ -741,9 +738,16 @@ async fn handle_message(
/// 判定秒级任务已完成。8s 覆盖秒级任务;长任务仅推迟主循环进入,不影响终版到达时刻
const FAST_TASK_GRACE: std::time::Duration = std::time::Duration::from_secs(8);
/// 完成事件订阅任务(#14 两段式):占位卡唯一写者。先投受理卡,再消费 ai_event_bus 的
/// AiAgentFinished(本 conv)原地 edit_card 为终版。引擎层 deliver_completion_report_if_waited
/// 是广播单一发出点,watcher 纯消费无状态。
/// 卡片帧:占位卡唯一写者(预览任务)与裁决者(完成订阅任务)之间的投递契约。
/// terminal=true 的帧(终版/取消/失败)写入后预览任务收工退出,保证卡片最后写入即终版。
struct CardFrame {
text: String,
terminal: bool,
}
/// 完成事件订阅任务(#14 两段式):只做裁决——消费 ai_event_bus 的 AiAgentFinished(本 conv)
/// 决定终版文本,经帧通道交给预览任务原地落卡(占位卡唯一写者是预览任务,受理卡/终版/阶段
/// 提示全部经帧通道按序落卡)。引擎层 deliver_completion_report_if_waited 是广播单一发出点。
///
/// 超时语义:35min(> loop 30min 上限)首超时回查边界内产出,无果投降级提示但不退出
/// (任务后台不丢,晚到完成事件仍送达);二次超时真弃守(#23:两轮均无事件且无边界内
@@ -751,14 +755,15 @@ const FAST_TASK_GRACE: std::time::Duration = std::time::Duration::from_secs(8);
async fn run_completion_watcher(
ctx: Arc<Ctx>,
conv_id: String,
placeholder_id: Option<String>,
message_id: String,
chat_id: String,
ack: String,
// #20:计数生命周期跟任务走,guard 持有至 watcher 退出
_inflight: InflightGuard,
// 24 取消收口:/cancel 旁路唤醒(引擎停止路径只 emit AiCompleted,本任务消费不到)
mut cancel_rx: tokio::sync::oneshot::Receiver<()>,
// 13a:占位卡写权限交预览任务,本任务所有落卡经此投递(terminal 标记终版)
frame_tx: tokio::sync::mpsc::Sender<CardFrame>,
// 预览任务句柄:投终版后 await 其退出,保证落卡完成后才释放 inflight
stream_task: tauri::async_runtime::JoinHandle<()>,
) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
@@ -768,8 +773,8 @@ async fn run_completion_watcher(
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
// 落受理卡(占位缺失回退新发,deliver_result 内部链路)
deliver_result(&ctx.client, &placeholder_id.clone(), &message_id.clone(), &chat_id.clone(), &ack).await;
// 落受理卡(经预览任务原地更新占位卡,未收到终版前预览任务持续刷新过程进展)
let _ = frame_tx.send(CardFrame { text: ack, terminal: false }).await;
// #27 快任务兜底:受理卡发出后开宽限窗再判定,事件驱动替代恒假的时间比较。
// 秒级任务的完成事件可能先于本订阅已广播完(错过),窗结束时 conv 已归非活跃且
// 边界内有产出,即判快任务已完成,直投终版
@@ -778,7 +783,8 @@ async fn run_completion_watcher(
if let Some(t) = last_assistant_reply(&state.db, &conv_id, Some(accepted_ms)).await {
if !t.trim().is_empty() {
feishu_log(&format!("watcher 快任务兜底 conv={conv_id} len={}", t.len()));
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &t).await;
let _ = frame_tx.send(CardFrame { text: t, terminal: true }).await;
let _ = stream_task.await;
return;
}
}
@@ -787,19 +793,11 @@ async fn run_completion_watcher(
let timeout = std::time::Duration::from_secs(2100);
let mut timeouts = 0usize;
loop {
let text = tokio::select! {
let frame = tokio::select! {
// 24 取消收口:/cancel 旁路唤醒,投「已停止」终版后退出,不再等引擎事件
_ = &mut cancel_rx => {
feishu_log(&format!("watcher 收到取消指令 conv={conv_id}"));
deliver_result(
&ctx.client,
&placeholder_id,
&message_id,
&chat_id,
"🛑 任务已停止。可发 /status 查看会话状态,或直接发新任务。",
)
.await;
return;
CardFrame { text: "🛑 任务已停止。可发 /status 查看会话状态,或直接发新任务。".to_string(), terminal: true }
}
item = rx.recv() => match item {
Ok(v) => {
@@ -813,11 +811,17 @@ async fn run_completion_watcher(
Some("AiError") => {
let err = v.get("error").and_then(|s| s.as_str()).unwrap_or("未知错误");
feishu_log(&format!("watcher 收到 AiError conv={conv_id}"));
format!("❌ 任务执行失败:{},可稍后重试,或发 /status 查看会话状态。", truncate(err, 300))
CardFrame {
text: format!("❌ 任务执行失败:{},可稍后重试,或发 /status 查看会话状态。", truncate(err, 300)),
terminal: true,
}
}
Some("AiAgentFinished") => {
// 广播语义:agent_id==conv_id 时 summary 即本会话最终报告(引擎层构造)
v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string()
CardFrame {
text: v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string(),
terminal: true,
}
}
_ => continue,
}
@@ -829,25 +833,30 @@ async fn run_completion_watcher(
timeouts += 1;
// 二次超时弃守:两轮 35min 均无事件且无边界内产出,认任务失联,停止重刷
if timeouts >= 2 {
let give_up = "⏰ 任务超 70 分钟无任何结果(已停止等待;引擎仍在后台运行则结果稍后可在桌面端查看;可发 /status 查询)。".to_string();
feishu_log(&format!("watcher 二次超时弃守 conv={conv_id}"));
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &give_up).await;
return;
}
// 首超时:DB 回查兜底(任务可能刚完成事件已发完);回查有果照投终版,
// 无果投降级提示并继续等待(晚到事件仍送达)
match last_assistant_reply(&state.db, &conv_id, Some(accepted_ms)).await {
Some(t) if !t.trim().is_empty() => t,
_ => {
let notice = "⏰ 任务执行超 35 分钟仍未返回(后台仍在运行,完成后本卡片自动更新;或本会话 /status 查询)。".to_string();
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &notice).await;
continue;
CardFrame {
text: "⏰ 任务超 70 分钟无任何结果(已停止等待;引擎仍在后台运行则结果稍后可在桌面端查看;可发 /status 查询)。".to_string(),
terminal: true,
}
} else {
// 首超时:DB 回查兜底(任务可能刚完成事件已发完);回查有果照投终版,
// 无果投降级提示并继续等待(晚到事件仍送达)
match last_assistant_reply(&state.db, &conv_id, Some(accepted_ms)).await {
Some(t) if !t.trim().is_empty() => CardFrame { text: t, terminal: true },
_ => {
let _ = frame_tx.send(CardFrame {
text: "⏰ 任务执行超 35 分钟仍未返回(后台仍在运行,完成后本卡片自动更新;或本会话 /status 查询)。".to_string(),
terminal: false,
}).await;
continue;
}
}
}
}
};
feishu_log(&format!("watcher 终版 conv={conv_id} len={}", text.len()));
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await;
feishu_log(&format!("watcher 终版 conv={conv_id} len={}", frame.text.len()));
let _ = frame_tx.send(frame).await;
let _ = stream_task.await; // 等预览任务落卡终版并退出,再释放 inflight
return;
}
}
@@ -1199,18 +1208,33 @@ async fn read_whitelist(ctx: &Ctx) -> Vec<String> {
const STREAM_FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3);
/// 流式预览上限字符数:对齐终版 clip_for_card 口径;超限从头截断保尾部
const STREAM_PREVIEW_MAX_CHARS: usize = 8000;
/// 13a-2 静默期心跳阈值:超此秒数无本 conv 事件(delta/轮次/工具),tick 强制刷心跳行。
/// 13a-2 静默期心跳阈值:超此秒数无本 conv 内容进展(delta/轮次/工具/进度),tick 强制刷心跳行。
/// 60s 取「长工具常见执行时长」与「卡片冻结感阈值」之间——再短会与常规刷新混叠,
/// 再长用户已开始疑心卡死。心跳行此后每静默分钟刷一次(桶去重),事件恢复即自然停
/// 再长用户已开始疑心卡死。心跳行此后每静默分钟刷一次(桶去重),内容恢复即自然停。
/// 注意:AiHeartbeat 等引擎保活事件不计入活动钟(非内容进展),否则心跳节拍会把静默永远压在阈值内
const HEARTBEAT_SILENT_THRESHOLD: u64 = 60;
/// 流式预览任务:消费 ai_event_bus 的 AiTextDelta,节流把累计文本 edit_card 进占位卡片;
/// 终版文本由 deliver_tx 单点传入,复用 deliver_result 投递后退出。
/// 13a-2 心跳刷判定(纯函数,单测锁定):内容静默达阈值且进入新的静默分钟桶时返回该桶号
/// (应刷),否则 None(不刷)。桶去重保证每个静默分钟只刷一次,内容恢复(静默回落)即停。
fn heartbeat_due(silent_secs: u64, painted_bucket: Option<u64>, threshold: u64) -> Option<u64> {
if silent_secs < threshold {
return None;
}
let bucket = silent_secs / 60;
if Some(bucket) == painted_bucket {
None
} else {
Some(bucket)
}
}
/// 流式预览任务:订阅 ai_event_bus 的增量(delta/轮次/工具/进度)节流把累计内容 edit_card
/// 进占位卡;受理卡/阶段提示/终版经帧通道(frame_rx)传入,本任务是占位卡唯一写者。
///
/// 竞态设计:占位卡片在本任务存活期只有本任务写,中间态 tick 与终版投递同任务串行,
/// 上一次 edit_card 的 HTTP 响应返回后才可能发下一次,飞书服务端所见次序即代码次序,
/// 终版不会被迟到的中间态覆盖,故无需 AbortHandle。中间态刷失败不新发卡片回退
/// (不值得为预览刷屏),新发回退链仍归终版 deliver_result 独有,行为不受本任务影响。
/// 竞态设计:占位卡在本任务存活期只有本任务写,中间态 tick 与帧投递同任务串行,上一次
/// edit_card 的 HTTP 响应返回后才可能发下一次,飞书服务端所见次序即代码次序,终版不会
/// 被迟到的中间态覆盖,故无需 AbortHandle。中间态刷失败不新发卡片回退(不值得为预览刷屏),
/// 新发回退链归帧投递(deliver_result)独有。
async fn run_stream_refresher(
ctx: Arc<Ctx>,
conv_id: String,
@@ -1218,7 +1242,7 @@ async fn run_stream_refresher(
message_id: String,
chat_id: String,
mut delta_rx: tokio::sync::broadcast::Receiver<serde_json::Value>,
mut deliver_rx: tokio::sync::oneshot::Receiver<String>,
mut frame_rx: tokio::sync::mpsc::Receiver<CardFrame>,
) {
let mut acc = String::new();
// 累计字符数 O(1) 维护,与上次已刷数比对判"有无新增量",避免无变化空刷烧配额
@@ -1239,22 +1263,28 @@ async fn run_stream_refresher(
let mut progress_tokens: Option<u64> = None;
// 13a-1 刷新可观测:流式刷失败连败计数(成功归零;==阈值告警一次,继续累计防重复告警)
let mut refresh_failures: u32 = 0;
// 13a-2 静默期心跳:最后活动时刻(本 conv 任意事件都算活着);静默超阈值强制刷心跳行,
// 防长工具/LLM 长往返期间卡片冻结在旧帧,用户误判卡死(todo.md 13a-2)
// 13a-2 静默期心跳:最后内容进展时刻(delta/轮次/工具/进度;不含 AiHeartbeat);静默超阈值
// 强制刷心跳行,防长工具/LLM 长往返期间卡片冻结在旧帧,用户误判卡死(todo.md 13a-2)
let mut last_activity = tokio::time::Instant::now();
let mut painted_hb_bucket: Option<u64> = None;
// 受理卡(首帧)到达前不写卡:防心跳/预览帧先于受理卡落卡,次序错乱
let mut started = false;
let mut ticker = tokio::time::interval(STREAM_FLUSH_INTERVAL);
// 刷新落后只保最新帧不补帧:预览语义是最新输出才有价值
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
// 终版单点收口:投递后任务退出,卡片所有权移交终版内容
final_text = &mut deliver_rx => {
if let Ok(text) = final_text {
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await;
// 帧通道:受理卡/阶段提示/终版。走 deliver_result(净化+裁剪+新发兜底),与预览同任务串行
frame = frame_rx.recv() => {
let Some(f) = frame else {
// 发送端丢弃(watcher 退出/受理失败):无更多帧,收工
return;
};
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &f.text).await;
started = true;
if f.terminal {
return;
}
// 发送端被丢弃(spawn 失败路径)时不投递,由调用方兜底
return;
}
item = delta_rx.recv() => match item {
Ok(v) => {
@@ -1267,11 +1297,10 @@ async fn run_stream_refresher(
if let Some(delta) = v.get("delta").and_then(|d| d.as_str()) {
acc.push_str(delta);
acc_chars += delta.chars().count();
// 首帧提速(2026-10-08):首个 delta 到达即立即刷占位卡,
// 不等 3s ticker——短响应(简单问答/翻译)原需等满一个节流周期
// 才见首字,感知停顿明显。仅首帧走即时刷(acc_chars 首次>painted),
// 后续仍按 3s 节流守飞书 QPS 配额
if painted_chars == 0 && acc_chars > 0 {
// 首帧提速(2026-10-08):首个 delta 到达且受理卡已落地即立即刷占位卡,
// 不等 3s ticker——短响应(简单问答/翻译)原需等满一个节流周期才见首字,
// 感知停顿明显。仅首帧走即时刷,后续仍按 3s 节流守飞书 QPS 配额
if started && painted_chars == 0 && acc_chars > 0 {
if let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) {
let progress_snapshot = progress_round.map(|pr| (pr, progress_max, progress_tokens));
if update_card(&ctx.client, pid, &stream_preview(&acc, round, current_tool.as_deref(), last_tool.as_deref(), tool_steps, None, progress_snapshot)).await {
@@ -1311,48 +1340,46 @@ async fn run_stream_refresher(
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
}
_ => {}
// AiHeartbeat 等引擎保活事件不重置内容静默钟:否则 20s 心跳节拍会把静默
// 永远压在阈值内,13a-2 心跳行永不触发(静默指内容进展停顿,非事件停顿)
_ => continue,
}
// 13a-2 活动时钟:本 conv 任意事件(delta/轮次/工具/进度)都算活着,重置静默计时
// 13a-2 活动时钟:本 conv 内容事件(delta/轮次/工具/进度)算活着,重置静默计时
last_activity = tokio::time::Instant::now();
}
// 预览允许丢帧:终版事实源在通知桥 oneshot,消费落后只降级预览不断链路
// 预览允许丢帧:终版事实源在帧通道,消费落后只降级预览不断链路
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::warn!("[feishu] 流式预览消费落后,丢弃 {n} 条增量(预览降级,终版不受影响)");
}
// 总线随 AppState 常驻理论不关;防御性收口:只等终版,不再消费增量
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
if let Ok(text) = deliver_rx.await {
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await;
}
return;
}
// 总线随 AppState 常驻理论不关;防御性收口:总线关闭即收工(进程收尾)
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
},
_ = ticker.tick() => {
_ = ticker.tick() => {
// 受理卡未落地前不刷:防预览/心跳帧抢先覆盖受理卡
if !started {
continue;
}
// 占位缺失(首张卡片 reply 失败)时无从原地刷,保持静默等终版走新发回退
let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) else {
continue;
};
// ── 13a-2 静默期心跳:超阈值无事件强制刷心跳行,防卡片冻结疑似卡死 ──
// ── 13a-2 静默期心跳:内容静默达阈值且进入新静默分钟桶 → 刷心跳行(纯函数判定) ──
// 长工具/LLM 长往返期间无 delta 无轮次推进,卡片静止;心跳行给出「引擎活着+
// 进行到哪」证据。桶去重:每静默分钟只刷一次(silent_secs/60 桶号相等即同桶)
// 进行到哪」证据。桶去重保证每静默分钟只刷一次,内容恢复即自然停
let silent_secs = last_activity.elapsed().as_secs();
let hb_needed = silent_secs >= HEARTBEAT_SILENT_THRESHOLD;
let hb_bucket = silent_secs / 60;
if hb_needed && Some(hb_bucket) == painted_hb_bucket {
continue; // 本静默分钟已刷过心跳,去重
}
let hb_bucket = heartbeat_due(silent_secs, painted_hb_bucket, HEARTBEAT_SILENT_THRESHOLD);
let hb_needed = hb_bucket.is_some();
// 常规信号标脏:过程事件已回拨 painted_chars,字符数不等即刷
let dirty = acc_chars != painted_chars;
if !dirty && !hb_needed {
continue;
}
// 签名去重:正文长度与轨迹行均未变则跳过(防同 tick 内事件风暴空刷)
let sig = format!("{}|{}|{}|{}|{}", acc_chars, round.unwrap_or(0), current_tool.as_deref().unwrap_or(""), tool_steps, if hb_needed { hb_bucket } else { u64::MAX });
let sig = format!("{}|{}|{}|{}|{}", acc_chars, round.unwrap_or(0), current_tool.as_deref().unwrap_or(""), tool_steps, hb_bucket.unwrap_or(u64::MAX));
if sig == painted_sig {
continue;
}
// 13a-1 刷新可观测:刷失败不再静默——feishu_log 落盘+连败计数,≥3 连败告警一次(计数继续累计,
// 13a-1 刷新可观测:刷失败落盘+连败计数,≥3 连败告警一次(计数继续累计,
// 重置仅在成功后;防每 tick 重复告警刷屏)。排查凭据:刷了没刷/刷败原因全可查
let silent_for_preview = if hb_needed { Some(silent_secs) } else { None };
let progress_snapshot = progress_round.map(|pr| (pr, progress_max, progress_tokens));
@@ -1363,8 +1390,8 @@ _ = ticker.tick() => {
refresh_failures = 0;
painted_chars = acc_chars;
painted_sig = sig;
if hb_needed {
painted_hb_bucket = Some(hb_bucket); // 心跳桶已刷,本静默分钟不再刷
if let Some(b) = hb_bucket {
painted_hb_bucket = Some(b); // 心跳桶已刷,本静默分钟不再刷
}
} else {
refresh_failures += 1;
@@ -1373,7 +1400,7 @@ _ = ticker.tick() => {
));
if refresh_failures == REFRESH_FAIL_ALERT_THRESHOLD {
feishu_log(&format!(
"[refresher] ⚠️ 连续 {refresh_failures} 次刷新失败,卡片可能冻结——流式预览降级,终版仍走通知桥(不受影响)"
"[refresher] ⚠️ 连续 {refresh_failures} 次刷新失败,卡片可能冻结——流式预览降级,终版仍走帧通道(不受影响)"
));
}
}
@@ -1931,7 +1958,7 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap<String, Stri
#[cfg(test)]
mod tests {
use super::{
is_terminal_push, md_to_lark_friendly, msg_after_acceptance, parse_command,
heartbeat_due, is_terminal_push, md_to_lark_friendly, msg_after_acceptance, parse_command,
strip_mention_placeholders, stream_preview,
};
@@ -2069,6 +2096,16 @@ mod tests {
assert!(out.starts_with("💓 **仍在跑:第 15 轮 · 最近动作 `3 步工具调用` · 静默 72s**\n\n"));
}
#[test]
fn heartbeat_due_threshold_and_bucket_dedup() {
// 13a-2 判定纯函数:未达阈值不刷;达阈值且新桶刷;同静默分钟桶去重;进新桶再刷
assert_eq!(heartbeat_due(59, None, 60), None);
assert_eq!(heartbeat_due(60, None, 60), Some(1));
assert_eq!(heartbeat_due(60, Some(1), 60), None);
assert_eq!(heartbeat_due(119, Some(1), 60), None); // 119/60=1 同桶
assert_eq!(heartbeat_due(120, Some(1), 60), Some(2));
}
// ── 38a 出站 md 净化 ──
#[test]