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

修复: 飞书细项批: 取消审批通路+多chat路由+命令容错

Browse Source
This commit is contained in:
lxy
2026-10-07 18:33:41 +08:00
parent 8c06076873
commit 76ace953be
1 changed files with 513 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/feishu_channel.rs
+486 -30
View File
@@ -36,12 +36,30 @@ const QUEUE_BUSY_MSG: &str =
/// OnceLock 语义匹配(重启应用才可能重配)。
static ACTIVE_CLIENT: std::sync::OnceLock<&'static LarkClient> = std::sync::OnceLock::new();
/// 渠道客户端只读访问(37 监控告警等外部推送方用):None=渠道未启动
pub(crate) fn active_client() -> Option<&'static LarkClient> {
ACTIVE_CLIENT.get().copied()
}
/// 34 滚动汇总卡句柄:chat_id 对 (在途事务卡 message_id, 已累计正文)。同 chat 多步
/// 进度推送原地追加为一卡,防逐条直推刷屏;内存句柄重启即失(重启后首推另起新卡)。
/// std Mutex 只做句柄与字符串存取,不跨 await 持有,短锁语义
static ROLLING_CARDS: std::sync::OnceLock<std::sync::Mutex<HashMap<String, (String, String)>>> =
std::sync::OnceLock::new();
/// 24 取消收口旁路:conv_id 对完成订阅任务的取消发送端。引擎停止路径(ai_stop_loop 置
/// stop_flag 后 loop 于轮边界收尾)只 emit AiCompleted,watcher 消费不到;/cancel 后须旁路
/// 唤醒 watcher 投「已停止」终版,防空等至双超时弃守期间反复重刷过期提示。oneshot 消费
/// 一次即摘除,watcher 任意路径退出亦摘除,陈旧条目最多导致一次空发,无泄漏
static CANCEL_WATCHERS: std::sync::OnceLock<
tokio::sync::Mutex<HashMap<String, tokio::sync::oneshot::Sender<()>>>,
> = std::sync::OnceLock::new();
fn cancel_watchers(
) -> &'static tokio::sync::Mutex<HashMap<String, tokio::sync::oneshot::Sender<()>>> {
CANCEL_WATCHERS.get_or_init(|| tokio::sync::Mutex::new(HashMap::new()))
}
#[derive(Debug, Clone, Deserialize)]
struct FeishuConfig {
#[serde(default)]
@@ -139,6 +157,9 @@ pub fn start_if_enabled(app: tauri::AppHandle) {
let leaked: &'static LarkClient = Box::leak(Box::new(client));
// 19 outbound 统一出口:全局句柄就绪,push_to_chat 自此可用
let _ = ACTIVE_CLIENT.set(leaked);
// 24 审批可见性:渠道生命周期订阅任务,审批挂起事件推卡到绑定 chat(跨重连常驻,
// 与 run_loop 的按连接重建无关)
tauri::async_runtime::spawn(run_approval_notifier(ctx.clone()));
// 断线自动重连:网络波动/服务端断开导致 start 退出时 backoff 重试,防 bot 静默死
let mut backoff = 2u64;
loop {
@@ -245,6 +266,36 @@ fn now_secs() -> i64 {
.unwrap_or(0)
}
/// 命令解析前归一(30 半全角容错):全角 ASCII 变体与全角空格转半角。仅用于命令识别,
/// 任务原文不归一(不动用户内容)
fn normalize_halfwidth(s: &str) -> String {
s.chars()
.map(|c| match c {
'\u{3000}' => ' ',
c if ('\u{FF01}'..='\u{FF5E}').contains(&c) => {
char::from_u32(c as u32 - 0xFEE0).unwrap_or(c)
}
_ => c,
})
.collect()
}
/// 命令识别(30 大小写/附言容错):`/cmd` 或 `/cmd 附言` 皆命中,命令词小写化,
/// 返回 (命令词, 附言)。非命令(无 / 前缀或空命令词)返回 None,落常规任务路径;
/// 未知命令词由调用方落回常规路径
fn parse_command(text: &str) -> Option<(String, String)> {
let t = normalize_halfwidth(text);
let t = t.trim();
let rest = t.strip_prefix('/')?;
let mut parts = rest.splitn(2, char::is_whitespace);
let cmd = parts.next()?.to_ascii_lowercase();
if cmd.is_empty() {
return None;
}
let tail = parts.next().unwrap_or("").trim().to_string();
Some((cmd, tail))
}
/// 入站文本:群聊需 @bot;白名单校验;/new 重置;其余走 ai-chat 会话链路
async fn handle_message(
ctx: Arc<Ctx>,
@@ -296,19 +347,44 @@ async fn handle_message(
return;
}
// /new 重置该飞书会话映射(下次消息开新 conv)
if text == "/new" {
// 命令通路(#30 容错):全角/大小写/附言归一后识别,仅命令词走控制面,任务原文不动。
// 顺序约束:命令与审批决议须先于注入旁路——conv 活跃时这两类文本否则会被当补充
// 指令注入任务;未知命令词落回常规任务路径,零行为变更
if let Some((cmd, _tail)) = parse_command(&text) {
match cmd.as_str() {
// /new 重置该飞书会话映射(下次消息开新 conv)。附言容错忽略不烧 LLM;
// 旧 conv 任务在跑时重置只换绑定不杀任务,明示去向与出路(30)
"new" => {
let prev = ctx.chat_map.lock().await.get(&chat_id).cloned();
ctx.chat_map.lock().await.remove(&chat_id);
// 旧 conv 滚动汇总卡句柄一并清:绑定已换,推送不再续旧卡防上下文串卡
if let Some(m) = ROLLING_CARDS.get() {
if let Ok(mut h) = m.lock() {
h.remove(&chat_id);
}
}
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let snapshot = ctx.chat_map.lock().await.clone();
save_chat_map(&db, &snapshot).await;
reply_message(&ctx.client, &message_id, &chat_id, "✅ 已重置会话").await;
let state = ctx.app.state::<crate::state::AppState>();
let busy_note = match prev {
Some(cid) if state.conv_states.is_active(&cid) => {
"\n⚠️ 原会话任务仍在后台执行,本群不再接收其推送;如需停止请发 /cancel。"
}
_ => "",
};
reply_message(
&ctx.client,
&message_id,
&chat_id,
&format!("✅ 已重置会话{busy_note}"),
)
.await;
return;
}
// #21 /status 查询当前会话任务状态(话术曾引用不存在的命令,用户发 /status 被当任务
// 烧一轮 LLM)。只读不 spawn:查 conv 绑定与活跃态给进度摘要。
if text == "/status" {
"status" => {
let state = ctx.app.state::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let reply = match conv_id {
@@ -333,6 +409,60 @@ async fn handle_message(
reply_message(&ctx.client, &message_id, &chat_id, &reply).await;
return;
}
// 24 取消通路:飞书侧此前零 stop 通道,跑偏任务只能干等。对接引擎停止原语
// ai_stop_loop(置 stop_flag,loop 于轮边界收尾),并旁路唤醒完成订阅任务收口;
// 无在途任务如实回告,不误报已停止
"cancel" => {
let state = ctx.app.state::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let reply = match conv_id {
Some(cid) if state.conv_states.is_active(&cid) => {
feishu_log(&format!("/cancel 停止 conv={cid} chat={chat_id}"));
match crate::commands::ai::ai_stop_loop(ctx.app.clone(), state, cid.clone())
.await
{
Ok(_) => {
if let Some(tx) = cancel_watchers().lock().await.remove(&cid) {
let _ = tx.send(());
}
"🛑 已发送停止指令,任务收尾中。".to_string()
}
Err(e) => format!("⚠️ 停止指令未生效:{}", truncate(&e, 120)),
}
}
_ => "当前无在途任务。".to_string(),
};
reply_message(&ctx.client, &message_id, &chat_id, &reply).await;
return;
}
_ => {}
}
}
// 24 审批决议通路:该会话有挂起审批时,「批准/拒绝」直调审批原语不入对话链路
// (决议文本若落常规路径会被注入运行中任务,审批永远无人拍板)。决议取该会话最新
// 一条挂起;无挂起不拦截,文本照常落常规路径,零行为变更
if text == "批准" || text == "拒绝" {
let state = ctx.app.state::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let pending = match conv_id {
Some(ref cid) => latest_pending_approval(&state, cid).await,
None => None,
};
if let Some(tc_id) = pending {
let approved = text == "批准";
feishu_log(&format!("审批决议 chat={chat_id} tc={tc_id} approved={approved}"));
let reply =
match crate::commands::ai::ai_approve(ctx.app.clone(), state, tc_id, approved).await
{
Ok(_) if approved => "✅ 已批准,任务继续执行。".to_string(),
Ok(_) => "🚫 已拒绝,该工具调用终止,任务按拒绝结果继续。".to_string(),
Err(e) => format!("⚠️ 审批处理失败:{}", truncate(&e, 160)),
};
reply_message(&ctx.client, &message_id, &chat_id, &reply).await;
return;
}
}
// ── 纠偏注入旁路(todo.md 16,2026-10-06):会话运行中 → 直接注入运行中 conv,不排队 ──
// 排队语义的痛点:per-chat 锁把补充指令排到 agent 完成后才处理,长任务期间纠偏无效。
@@ -442,7 +572,7 @@ async fn handle_message(
// #26 占用期话术改实况:委托制无真排队,禁承诺「任务完成后处理」
QUEUE_BUSY_MSG.to_string()
} else {
format!("❌ 会话启动失败:{}`,请稍后重试", truncate(&e, 120))
format!("❌ 会话启动失败:{},请稍后重试", truncate(&e, 120))
};
deliver_result(
&ctx.client,
@@ -476,9 +606,24 @@ async fn handle_message(
// 返回即 drop,受理后任务还在后台跑计数却归零,探活误报窗口重开;任务终局(watcher
// 各退出路径)guard drop 自然回落。与本函数前置计数的短暂重叠对看门狗无影响(判据 >0)
let inflight_guard = InflightGuard(ctx.inflight.clone());
// 24 取消收口:注册本 conv 的取消发送端,/cancel 触发后旁路唤醒 watcher 投终版
let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
cancel_watchers().lock().await.insert(conv_id.clone(), cancel_tx);
let conv_watched = conv_id.clone();
tauri::async_runtime::spawn(async move {
run_completion_watcher(ctx, conv_id, placeholder_id, message_id, chat_id, ack, inflight_guard)
run_completion_watcher(
ctx,
conv_id,
placeholder_id,
message_id,
chat_id,
ack,
inflight_guard,
cancel_rx,
)
.await;
// watcher 任意路径退出都摘除取消句柄,防陈旧条目驻留(后续 /cancel 空发无害)
cancel_watchers().lock().await.remove(&conv_watched);
});
}
@@ -503,6 +648,8 @@ async fn run_completion_watcher(
ack: String,
// #20:计数生命周期跟任务走,guard 持有至 watcher 退出
_inflight: InflightGuard,
// 24 取消收口:/cancel 旁路唤醒(引擎停止路径只 emit AiCompleted,本任务消费不到)
mut cancel_rx: tokio::sync::oneshot::Receiver<()>,
) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
@@ -532,6 +679,19 @@ async fn run_completion_watcher(
let mut timeouts = 0usize;
loop {
let text = 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;
}
item = rx.recv() => match item {
Ok(v) => {
if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) {
@@ -583,6 +743,73 @@ async fn run_completion_watcher(
}
}
/// 24 审批可见性:渠道生命周期订阅任务。消费 ai_event_bus 的 AiApprovalRequired,反查
/// conv 绑定的飞书 chat 推卡提示(回复「批准」或「拒绝」);无绑定(纯桌面会话)不推,
/// 桌面 ApprovalOverlay 既有通路处理。审批挂起时流式卡静止无解释,本推送补齐可见性,
/// 决议闭环由 handle_message 的「批准/拒绝」拦截承接
async fn run_approval_notifier(ctx: Arc<Ctx>) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
loop {
match rx.recv().await {
Ok(v) => {
if v.get("type").and_then(|t| t.as_str()) != Some("AiApprovalRequired") {
continue;
}
let Some(cid) = v.get("conversation_id").and_then(|c| c.as_str()) else {
continue;
};
let chat_id = ctx
.chat_map
.lock()
.await
.iter()
.find(|(_, c)| c.as_str() == cid)
.map(|(k, _)| k.clone());
let Some(chat_id) = chat_id else {
continue;
};
let name = v.get("name").and_then(|n| n.as_str()).unwrap_or("工具");
let reason = v
.get("reason")
.and_then(|r| r.as_str())
.unwrap_or("")
.replace('\n', " ");
let brief = truncate(&reason, 80);
let card = if brief.is_empty() {
format!("⏸ 需审批: `{name}`。回复「批准」或「拒绝」继续")
} else {
format!("⏸ 需审批: `{name}` {brief}。回复「批准」或「拒绝」继续")
};
send_markdown_card(&ctx.client, "", &chat_id, &card).await;
feishu_log(&format!("审批挂起推送 chat={chat_id} conv={cid} tool={name}"));
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
}
}
}
/// 该 conv 最新一条挂起审批的 tool_call_id(决议原语的消费目标);无挂起返回 None。
/// created_at 缺失视为最旧(宁取有时序的),同刻并列取遍历后者(HashMap 无序,可接受)
async fn latest_pending_approval(
state: &crate::state::AppState,
conv_id: &str,
) -> Option<String> {
let session = state.ai_session.lock().await;
session
.pending_approvals
.iter()
.filter(|(_, a)| a.conversation_id.as_deref() == Some(conv_id))
.max_by_key(|(_, a)| {
a.created_at
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0)
})
.map(|(k, _)| k.clone())
}
/// D2 白名单热加载:每消息现读 KV,设置面板改白名单即生效免重启;读取/解析失败沿用
/// 上次缓存——拒收口径随缓存,不因 DB 抖动放大为误放行;成功时回写缓存供降级使用
async fn read_whitelist(ctx: &Ctx) -> Vec<String> {
@@ -605,6 +832,10 @@ 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 强制刷心跳行。
/// 60s 取「长工具常见执行时长」与「卡片冻结感阈值」之间——再短会与常规刷新混叠,
/// 再长用户已开始疑心卡死。心跳行此后每静默分钟刷一次(桶去重),事件恢复即自然停
const HEARTBEAT_SILENT_THRESHOLD: u64 = 60;
/// 流式预览任务:消费 ai_event_bus 的 AiTextDelta,节流把累计文本 edit_card 进占位卡片;
/// 终版文本由 deliver_tx 单点传入,复用 deliver_result 投递后退出。
@@ -631,8 +862,20 @@ async fn run_stream_refresher(
// 提炼原则:决策价值过滤——只留「第 N 轮 + 当前工具名」,历史工具折叠计数,不刷屏不顶正文。
let mut round: Option<u32> = None;
let mut current_tool: Option<String> = None;
// 13a-2 最近动作:工具完成后 current_tool 清空,但「最近在干什么」仍有展示价值(心跳行数据源)
let mut last_tool: Option<String> = None;
let mut tool_steps: usize = 0usize;
let mut painted_sig = String::new();
// 13a-3 阶段进度:最近一次 AiTaskProgress 快照(round/max/tokens),心跳行与头行数据源
let mut progress_round: Option<u32> = None;
let mut progress_max: Option<u32> = None;
let mut progress_tokens: Option<u64> = None;
// 13a-1 刷新可观测:流式刷失败连败计数(成功归零;==阈值告警一次,继续累计防重复告警)
let mut refresh_failures: u32 = 0;
// 13a-2 静默期心跳:最后活动时刻(本 conv 任意事件都算活着);静默超阈值强制刷心跳行,
// 防长工具/LLM 长往返期间卡片冻结在旧帧,用户误判卡死(todo.md 13a-2)
let mut last_activity = tokio::time::Instant::now();
let mut painted_hb_bucket: Option<u64> = None;
let mut ticker = tokio::time::interval(STREAM_FLUSH_INTERVAL);
// 刷新落后只保最新帧不补帧:预览语义是最新输出才有价值
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -670,6 +913,7 @@ async fn run_stream_refresher(
Some("AiToolCallStarted") => {
if let Some(name) = v.get("name").and_then(|n| n.as_str()) {
current_tool = Some(name.to_string());
last_tool = Some(name.to_string());
tool_steps += 1;
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
@@ -678,8 +922,20 @@ async fn run_stream_refresher(
current_tool = None;
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
// 13a-3 阶段进度:每 5 轮一次的里程碑(mod.rs PROGRESS_EVERY_N_ROUNDS),
// 立即标脏强制下一 tick 把「N/max 轮 · 累计 token」刷进卡片(长任务阶段可见)
Some("AiTaskProgress") => {
if let Some(r) = v.get("round").and_then(|r| r.as_u64()) {
progress_round = Some(r as u32);
progress_max = v.get("max_rounds").and_then(|m| m.as_u64()).map(|m| m as u32);
progress_tokens = v.get("tokens_used").and_then(|t| t.as_u64());
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
}
_ => {}
}
// 13a-2 活动时钟:本 conv 任意事件(delta/轮次/工具/进度)都算活着,重置静默计时
last_activity = tokio::time::Instant::now();
}
// 预览允许丢帧:终版事实源在通知桥 oneshot,消费落后只降级预览不断链路
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
@@ -693,25 +949,55 @@ async fn run_stream_refresher(
return;
}
},
_ = ticker.tick() => {
_ = ticker.tick() => {
// 占位缺失(首张卡片 reply 失败)时无从原地刷,保持静默等终版走新发回退
let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) else {
continue;
};
// 信号标脏:过程事件(轮次/工具)已把 painted_chars 回拨 1,此处字符数不等即刷
if acc_chars == painted_chars {
// ── 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; // 本静默分钟已刷过心跳,去重
}
// 常规信号标脏:过程事件已回拨 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);
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 });
if sig == painted_sig {
continue;
}
if update_card(&ctx.client, pid, &stream_preview(&acc, round, current_tool.as_deref(), tool_steps)).await {
// 13a-1 刷新可观测:刷失败不再静默——feishu_log 落盘+连败计数,≥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));
if update_card(&ctx.client, pid, &stream_preview(&acc, round, current_tool.as_deref(), last_tool.as_deref(), tool_steps, silent_for_preview, progress_snapshot)).await {
if refresh_failures > 0 {
feishu_log(&format!("[refresher] 刷新恢复(此前连败 {refresh_failures} 次)"));
}
refresh_failures = 0;
painted_chars = acc_chars;
painted_sig = sig;
if hb_needed {
painted_hb_bucket = Some(hb_bucket); // 心跳桶已刷,本静默分钟不再刷
}
} else {
refresh_failures += 1;
feishu_log(&format!(
"[refresher] update_card 失败(连败 {refresh_failures}) chat={chat_id} msg={pid} chars={acc_chars}"
));
if refresh_failures == REFRESH_FAIL_ALERT_THRESHOLD {
feishu_log(&format!(
"[refresher] ⚠️ 连续 {refresh_failures} 次刷新失败,卡片可能冻结——流式预览降级,终版仍走通知桥(不受影响)"
));
}
}
// 刷失败静默跳过,下个 tick 自然重试
}
}
}
@@ -721,22 +1007,50 @@ async fn run_stream_refresher(
///
/// 提炼原则(todo.md 13「决策价值过滤」):历史步骤折叠为计数,只透出「当前在干什么」——
/// 读者要的是"活着没卡死+进行到哪",不是 20 个工具名的流水墙;正文仍以流式文本为主,
/// 纯工具轮(无文本)时头部行是唯一进度证据。
/// 纯工具轮(无文本)时头部行是唯一进度证据。13a-2:静默超阈值时追加心跳行
/// 「💓 仍在跑:第 X 轮 · 最近动作 · 静默 Ys」——长工具/LLM 长往返期间卡片冻结疑似
/// 卡死,心跳行是「引擎活着」的用户可见证据。
fn stream_preview(
acc: &str,
round: Option<u32>,
current_tool: Option<&str>,
last_tool: Option<&str>,
tool_steps: usize,
silent_secs: Option<u64>,
progress: Option<(u32, Option<u32>, Option<u64>)>,
) -> String {
// 头部摘要行:轮次 + 当前工具(执行中)/已执行步数;全部缺省(首轮文本前)不显头行
// 头部摘要行:轮次 + 当前工具(执行中)/已执行步数;全部缺省(首轮文本前)不显头行。
// 13a-3:round 缺省时回退 AiTaskProgress 的 round(里程碑轮快照),并附累计 token
let mut head = String::new();
if round.is_some() || tool_steps > 0 {
let r = round.map(|n| format!("第 {n} 轮")).unwrap_or_default();
let (disp_round, prog_suffix) = match (&round, &progress) {
(Some(r), _) => (Some(*r), String::new()),
(None, Some((pr, max, tokens))) => {
let max_part = match max.filter(|m| *m > 0) {
Some(m) => format!("/{m} 轮"),
None => String::new(),
};
let tok_part = tokens.map(|t| format!(" · 累计 {t} tokens")).unwrap_or_default();
(Some(*pr), format!("{max_part}{tok_part}"))
}
(None, None) => (None, String::new()),
};
if disp_round.is_some() || tool_steps > 0 {
let r = disp_round.map(|n| format!("第 {n} 轮")).unwrap_or_default();
let t = match current_tool {
Some(name) => format!("⚙ `{name}` 执行中"),
None => format!("已完成 {tool_steps} 步工具调用"),
};
head = format!("**{r} · {t}**\n\n");
head = format!("**{r}{prog_suffix} · {t}**\n\n");
}
// 13a-2 心跳行:静默超阈值(refresher 判定后传入)时头部追加,数据源=轮次+最近工具。
// 动作优先级:执行中工具名 > 最近工具名 > 步数/思考中(last_tool_fallback)
if let Some(secs) = silent_secs {
let r = disp_round.map(|n| format!("第 {n} 轮")).unwrap_or_else(|| "启动".to_string());
let action = match current_tool.or(last_tool) {
Some(name) => format!("`{name}` 执行中"),
None => format!("最近动作 `{}`", last_tool_fallback(tool_steps)),
};
head = format!("💓 **仍在跑:{r} · {action} · 静默 {secs}s**\n\n{head}");
}
// 正文:超上限保尾截断(对齐原口径)
let total = acc.chars().count();
@@ -749,10 +1063,20 @@ fn stream_preview(
format!("{head}{body}\n▌")
}
/// markdown 回复:interactive 卡片承载(lark_md 原生渲染标题/列表/粗体,text 类型会符号裸露)。
/// 13a-2 心跳行最近动作回退:current_tool 已清空(工具完成、等 LLM 响应)时的展示文案。
/// tool_steps>0 用步数表述「刚干完 N 步」,零步(纯 LLM 长思考)退「LLM 思考中」
fn last_tool_fallback(tool_steps: usize) -> String {
if tool_steps > 0 {
format!("{tool_steps} 步工具调用")
} else {
"LLM 思考中".to_string()
}
}
/// markdown 回复:interactive 卡片承载
/// 出站先过 38a 净化(md_to_lark_friendly),成功返回新卡片 message_id(34 滚动汇总卡
/// 绑定句柄用,取 id 方式与 reply_markdown_card 同款);彻底失败回退纯文本保可达,返 None
async fn send_markdown_card(
pub(crate) async fn send_markdown_card(
client: &LarkClient,
message_id: &str,
chat_id: &str,
@@ -883,10 +1207,11 @@ async fn deliver_result(
/// 11:23 实证:绕过链的回复带完整 md 表格未降级,踩回表格渲染不稳坑)。
/// 34 滚动汇总卡:同 chat 在途事务卡存在时原地追加(句柄见 ROLLING_CARDS),✅/❌ 终态
/// 更新后清句柄下条另起新卡;原地更新失败清句柄回退新发并重绑(比照 deliver_result 兜底)。
/// 推送目标:chat_map 最近映射的会话;渠道未启动(无 client)或无映射时静默跳过。
/// 29 推送路由:conv_hint 按绑定关系精确反查 chat;缺省/未命中时单绑定直用(单绑定行为
/// 不变),多绑定取最近活跃,不再取 KV 首条盲投。渠道未启动(无 client)或无映射时静默跳过。
/// 19 正典出口预留给将来推送方(挂机进度汇总/巡检通知),当前无调用方
#[allow(dead_code)]
pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str) {
pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str, conv_hint: Option<&str>) {
let Some(client) = ACTIVE_CLIENT.get() else {
return; // 渠道未启动:推送无落点,静默跳过(调用方自行决定是否记日志)
};
@@ -897,8 +1222,7 @@ pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str) {
else {
return;
};
// 最近会话:取任一映射(KV HashMap 无序,取首条;多 chat 精准路由属将来形态)
let Some(chat_id) = map.keys().next().cloned() else {
let Some(chat_id) = resolve_push_chat(&db, &map, conv_hint).await else {
return;
};
// 34 终态判据:✅/❌ 开头即事务收尾,更新后不再续卡
@@ -943,6 +1267,54 @@ pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str) {
feishu_log(&format!("push_to_chat chat={chat_id} len={} terminal={terminal}", processed.len()));
}
/// 29 推送路由裁决:conv_hint 有绑定命中即用(多命中取最近活跃);缺省/未命中时单绑定
/// 直用(行为与旧首条取法一致),多绑定取最近活跃。无绑定返回 None 静默跳过
async fn resolve_push_chat(
db: &df_storage::db::Database,
map: &HashMap<String, String>,
conv_hint: Option<&str>,
) -> Option<String> {
if let Some(hint) = conv_hint {
let hits: Vec<&String> = map
.iter()
.filter(|(_, c)| c.as_str() == hint)
.map(|(k, _)| k)
.collect();
match hits.len() {
1 => return Some(hits[0].clone()),
n if n > 1 => return most_recent_chat(db, &hits).await,
_ => {} // hint 未命中:绑定可能已重建,落多绑定裁决
}
}
match map.len() {
0 => None,
1 => map.keys().next().cloned(),
_ => {
let all: Vec<&String> = map.keys().collect();
most_recent_chat(db, &all).await
}
}
}
/// 多绑定裁决:逐 conv 取最后一条消息时间戳,最新者胜;全部无消息取遍历首个(稳定退化)
async fn most_recent_chat(db: &df_storage::db::Database, chats: &[&String]) -> Option<String> {
use df_storage::crud::AiMessageRepo;
let repo = AiMessageRepo::new(db);
let mut best: Option<(&String, i64)> = None;
for cid in chats {
let ts = repo
.list_by_conversation(cid)
.await
.ok()
.and_then(|msgs| msgs.last().and_then(|m| m.timestamp))
.unwrap_or(i64::MIN);
if best.map_or(true, |(_, bts)| ts > bts) {
best = Some((cid, ts));
}
}
best.map(|(k, _)| (*k).clone())
}
/// 卡片正文长度防御:飞书卡片 JSON 约 30KB 上限,超长会让原地更新与新发双双失败,
/// 一路穿透到纯文本兜底仍不可读;入口统一裁剪,尾注指回桌面端(u-claw archiveIfClipped 同口径)
fn clip_for_card(text: &str, max_chars: usize) -> String {
@@ -1124,8 +1496,11 @@ fn table_line_to_list(line: &str, bold: bool) -> String {
format!("- {}", cells.join("："))
}
/// 13a-1 刷新可观测:流式刷连败告警阈值(== 触发一次,继续累计不重复告警)
const REFRESH_FAIL_ALERT_THRESHOLD: u32 = 3;
/// 关键事件日志落盘(零依赖 fs append;Roaming 下 feishu-channel.log,排查空回复等复现凭据)
fn feishu_log(line: &str) {
pub(crate) fn feishu_log(line: &str) {
use std::io::Write;
if let Ok(dir) = std::env::var("APPDATA") {
let p = std::path::Path::new(&dir).join("top.1216.devflow").join("feishu-channel.log");
@@ -1176,7 +1551,43 @@ 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, strip_mention_placeholders, stream_preview};
use super::{
is_terminal_push, md_to_lark_friendly, msg_after_acceptance, parse_command,
strip_mention_placeholders, stream_preview,
};
#[test]
fn command_case_and_tail_tolerance() {
// 30:/new 大小写与附言容错,附言不烧 LLM
assert_eq!(parse_command("/new"), Some(("new".to_string(), String::new())));
assert_eq!(
parse_command("/New 帮我开新会话"),
Some(("new".to_string(), "帮我开新会话".to_string()))
);
assert_eq!(parse_command("/STATUS"), Some(("status".to_string(), String::new())));
}
#[test]
fn command_fullwidth_normalized() {
// 30:半全角混排,全角斜杠/字母/空格归一后识别
assert_eq!(parse_command("／ｎｅｗ"), Some(("new".to_string(), String::new())));
assert_eq!(
parse_command("／ＳＴＡＴＵＳ　附加"),
Some(("status".to_string(), "附加".to_string()))
);
}
#[test]
fn command_none_and_unknown_fallthrough() {
// 纯任务文本与裸斜杠不是命令
assert!(parse_command("帮我走查代码").is_none());
assert!(parse_command("/").is_none());
// 未知命令词解析得出,由调用方落回常规任务路径(零行为变更);无空格路径整体为命令词
assert_eq!(
parse_command("/usr/bin 定位").map(|(c, _)| c),
Some("usr/bin".to_string())
);
}
#[test]
fn mention_cn_no_space_not_swallowed() {
@@ -1205,14 +1616,14 @@ mod tests {
#[test]
fn preview_first_text_no_head() {
// 首轮纯文本(无轮次无工具):不显头行,原文+光标
let out = stream_preview("你好", None, None, 0);
let out = stream_preview("你好", None, None, None, 0, None, None);
assert_eq!(out, "你好\n▌");
}
#[test]
fn preview_tool_running_head() {
// 工具执行中:头行 = 轮次 + ⚙ 工具名 执行中
let out = stream_preview("分析中", Some(3), Some("read_file"), 5);
let out = stream_preview("分析中", Some(3), Some("read_file"), None, 5, None, None);
assert!(out.starts_with("**第 3 轮 · ⚙ `read_file` 执行中**\n\n"));
assert!(out.contains("分析中"));
assert!(out.ends_with("▌"));
@@ -1221,7 +1632,7 @@ mod tests {
#[test]
fn preview_tool_done_head() {
// 工具轮间空档(当前无工具):头行转「已完成 N 步」——纯工具轮的"活着"证据
let out = stream_preview("", Some(4), None, 7);
let out = stream_preview("", Some(4), None, None, 7, None, None);
assert!(out.starts_with("**第 4 轮 · 已完成 7 步工具调用**\n\n"));
}
@@ -1229,12 +1640,57 @@ mod tests {
fn preview_tail_keep_on_overflow() {
// 超上限保尾截断 + 前缀标注;头行不受截断影响
let long = "x".repeat(9000);
let out = stream_preview(&long, Some(1), None, 1);
let out = stream_preview(&long, Some(1), None, None, 1, None, None);
assert!(out.starts_with("**第 1 轮 · 已完成 1 步工具调用**"));
assert!(out.contains("…(前文已略)"));
assert!(out.chars().count() < 8500); // 头行 + 8000 尾 + 标注
}
#[test]
fn preview_heartbeat_line_on_silent() {
// 13a-2 静默超阈值:心跳行置顶(在常规头行之前),含轮次/动作/静默秒数
let out = stream_preview("分析中", Some(3), Some("run_command"), None, 5, Some(67), None);
assert!(out.starts_with("💓 **仍在跑:第 3 轮 · `run_command` 执行中 · 静默 67s**\n\n"));
assert!(out.contains("**第 3 轮 · ⚙ `run_command` 执行中**")); // 常规头行保留
}
#[test]
fn preview_heartbeat_tool_done_fallback() {
// 13a-2 静默且工具间空档:动作优先用最近工具名,无则回退步数表述
let out = stream_preview("", Some(4), None, Some("grep"), 9, Some(95), None);
assert!(out.starts_with("💓 **仍在跑:第 4 轮 · `grep` 执行中 · 静默 95s**\n\n"));
// 无最近工具(纯 LLM 长思考)回退
let out2 = stream_preview("", Some(4), None, None, 9, Some(95), None);
assert!(out2.starts_with("💓 **仍在跑:第 4 轮 · 最近动作 `9 步工具调用` · 静默 95s**\n\n"));
}
#[test]
#[test]
fn preview_heartbeat_no_round_no_steps() {
// 13a-2 静默且零步零轮(纯 LLM 长思考):轮次回退「启动」,动作回退「LLM 思考中」
let out = stream_preview("开头文本", None, None, None, 0, Some(80), None);
assert!(out.starts_with("💓 **仍在跑:启动 · 最近动作 `LLM 思考中` · 静默 80s**\n\n"));
assert!(out.contains("开头文本\n▌"));
}
#[test]
fn preview_progress_fallback_round_when_no_agent_round() {
// 13a-3:AiAgentRound 缺席(纯进度事件场景)时,头行回退 AiTaskProgress 快照
// 并附 max 轮与累计 token;max=0(不限轮)不显上限
let out = stream_preview("干活中", None, None, None, 2, None, Some((10, Some(50), Some(123456))));
assert!(out.starts_with("**第 10 轮/50 轮 · 累计 123456 tokens · 已完成 2 步工具调用**"));
// max_rounds=0 = 不限轮次,只显 token
let out2 = stream_preview("干活中", None, None, None, 2, None, Some((10, Some(0), Some(999))));
assert!(out2.starts_with("**第 10 轮 · 累计 999 tokens · 已完成 2 步工具调用**"));
}
#[test]
fn preview_heartbeat_uses_progress_round() {
// 13a-3×13a-2 组合:静默心跳的轮次也吃 AiTaskProgress 快照
let out = stream_preview("", None, None, None, 3, Some(72), Some((15, None, Some(42))));
assert!(out.starts_with("💓 **仍在跑:第 15 轮 · 最近动作 `3 步工具调用` · 静默 72s**\n\n"));
}
// ── 38a 出站 md 净化 ──
#[test]