diff --git a/src-tauri/src/feishu_channel.rs b/src-tauri/src/feishu_channel.rs index f510211..cbd61a4 100644 --- a/src-tauri/src/feishu_channel.rs +++ b/src-tauri/src/feishu_channel.rs @@ -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::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>>, +> = std::sync::OnceLock::new(); + +fn cancel_watchers( +) -> &'static tokio::sync::Mutex>> { + 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, @@ -296,42 +347,121 @@ async fn handle_message( return; } - // /new 重置该飞书会话映射(下次消息开新 conv) - if text == "/new" { - ctx.chat_map.lock().await.remove(&chat_id); - let db = ctx.app.state::().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; - return; + // 命令通路(#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::().db.clone(); + let snapshot = ctx.chat_map.lock().await.clone(); + save_chat_map(&db, &snapshot).await; + let state = ctx.app.state::(); + 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 绑定与活跃态给进度摘要。 + "status" => { + let state = ctx.app.state::(); + let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned(); + let reply = match conv_id { + None => "📭 当前无绑定会话,发送消息即自动创建。".to_string(), + Some(cid) => { + let short = &cid[..8.min(cid.len())]; + match state.conv_states.get(&cid) { + s if s.is_active() => format!("🚀 会话 `{short}` 任务执行中,完成后自动推送结果;可发消息注入补充指令(下一轮生效)。"), + crate::commands::ai::agentic::conv_state::ConvState::Compressed => { + format!("⏳ 会话 `{short}` 压缩暂存中,稍后自动续跑。") + } + crate::commands::ai::agentic::conv_state::ConvState::Stopping => { + format!("⏹ 会话 `{short}` 正在停止。") + } + crate::commands::ai::agentic::conv_state::ConvState::Error => { + format!("❌ 会话 `{short}` 上次任务出错,重发消息即可重试。") + } + _ => "✅ 会话空闲,无进行中任务。".to_string(), + } + } + }; + 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::(); + 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; + } + _ => {} + } } - // #21 /status 查询当前会话任务状态(话术曾引用不存在的命令,用户发 /status 被当任务 - // 烧一轮 LLM)。只读不 spawn:查 conv 绑定与活跃态给进度摘要。 - if text == "/status" { + // 24 审批决议通路:该会话有挂起审批时,「批准/拒绝」直调审批原语不入对话链路 + // (决议文本若落常规路径会被注入运行中任务,审批永远无人拍板)。决议取该会话最新 + // 一条挂起;无挂起不拦截,文本照常落常规路径,零行为变更 + if text == "批准" || text == "拒绝" { let state = ctx.app.state::(); let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned(); - let reply = match conv_id { - None => "📭 当前无绑定会话,发送消息即自动创建。".to_string(), - Some(cid) => { - let short = &cid[..8.min(cid.len())]; - match state.conv_states.get(&cid) { - s if s.is_active() => format!("🚀 会话 `{short}` 任务执行中,完成后自动推送结果;可发消息注入补充指令(下一轮生效)。"), - crate::commands::ai::agentic::conv_state::ConvState::Compressed => { - format!("⏳ 会话 `{short}` 压缩暂存中,稍后自动续跑。") - } - crate::commands::ai::agentic::conv_state::ConvState::Stopping => { - format!("⏹ 会话 `{short}` 正在停止。") - } - crate::commands::ai::agentic::conv_state::ConvState::Error => { - format!("❌ 会话 `{short}` 上次任务出错,重发消息即可重试。") - } - _ => "✅ 会话空闲,无进行中任务。".to_string(), - } - } + let pending = match conv_id { + Some(ref cid) => latest_pending_approval(&state, cid).await, + None => None, }; - reply_message(&ctx.client, &message_id, &chat_id, &reply).await; - return; + 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,不排队 ── @@ -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) - .await; + 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::(); 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) { + let state = ctx.app.state::(); + 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 { + 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 { @@ -605,6 +832,10 @@ async fn read_whitelist(ctx: &Ctx) -> Vec { 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 = None; let mut current_tool: Option = None; + // 13a-2 最近动作:工具完成后 current_tool 清空,但「最近在干什么」仍有展示价值(心跳行数据源) + let mut last_tool: Option = None; let mut tool_steps: usize = 0usize; let mut painted_sig = String::new(); + // 13a-3 阶段进度:最近一次 AiTaskProgress 快照(round/max/tokens),心跳行与头行数据源 + let mut progress_round: Option = None; + let mut progress_max: Option = None; + let mut progress_tokens: Option = 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 = 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, current_tool: Option<&str>, + last_tool: Option<&str>, tool_steps: usize, + silent_secs: Option, + progress: Option<(u32, Option, Option)>, ) -> 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, + conv_hint: Option<&str>, +) -> Option { + 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 { + 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