diff --git a/crates/df-im-gateway/src/feishu.rs b/crates/df-im-gateway/src/feishu.rs index 5cca301..6336c7e 100644 --- a/crates/df-im-gateway/src/feishu.rs +++ b/crates/df-im-gateway/src/feishu.rs @@ -260,7 +260,7 @@ async fn handle_card_action(events: Sender, action: larksuite_oapi /// 构造审批卡片(Card JSON 1.0):标题 + 详情 + 同意/拒绝按钮 fn build_approval_card(approval_id: &str, title: &str, detail: &str) -> Card { - let detail_md = format!("{}\n\n**审批 ID**: `{approval_id}`", escape_md(detail)); + let detail_md = format!("{}\n\n**审批 ID**: `{approval_id}`", md_to_lark_friendly(detail)); let to_json_value = |v: serde_json::Value| -> larksuite_oapi_sdk_rs::JsonValue { serde_json::from_value(serde_json::to_value(v).expect("serde_json::Value 可逆")) .expect("按钮 value 结构兼容") @@ -285,19 +285,54 @@ fn build_approval_card(approval_id: &str, title: &str, detail: &str) -> Card { .element(Element::Action(actions)) } -/// Markdown 转义(飞书卡片 lark_md 语法) -fn escape_md(s: &str) -> String { - s.replace('\\', "\\\\") - .replace('*', "\\*") - .replace('_', "\\_") - .replace('`', "\\`") - .replace('[', "\\[") - .replace(']', "\\]") - .replace('<', "\\<") - .replace('>', "\\>") - .replace('(', "\\(") - .replace(')', "\\)") - .replace('#', "\\#") +/// 出站 Markdown 净化(飞书卡片 lark_md 语法适配)。 +/// +/// AI 产出常带标准 Markdown 结构,lark_md 不支持标题与表格语法,原样注入按字面量渲染。 +/// 规则(与 src-tauri feishu_channel.rs 侧 md_to_lark_friendly 同规则互引,双端出站口径一致): +/// 标题行(行首一个以上 #)转粗体行;表格行(| 开头)转列表行,表头分隔行整行跳过; +/// 代码块围栏内外均保留原样(转义或改写破坏代码展示);其余行原样透传(lark_md 原生 +/// 支持粗体/链接/行内代码,不额外转义)。 +fn md_to_lark_friendly(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut in_code = false; + for line in s.lines() { + if line.trim_start().starts_with("```") { + in_code = !in_code; + out.push_str(line); + out.push('\n'); + continue; + } + if in_code { + out.push_str(line); + out.push('\n'); + continue; + } + let trimmed = line.trim_start(); + if trimmed.starts_with('#') { + let title = trimmed.trim_start_matches('#').trim(); + if !title.is_empty() { + out.push_str(&format!("**{title}**\n")); + } + continue; + } + if trimmed.starts_with('|') { + let cells: Vec<&str> = trimmed.trim_matches('|').split('|').map(str::trim).collect(); + // 表头分隔行(|---|:---|)无内容语义,整行跳过 + if cells.iter().all(|c| c.is_empty() || c.chars().all(|ch| ch == '-' || ch == ':')) { + continue; + } + let shown: Vec<&str> = cells.into_iter().filter(|c| !c.is_empty()).collect(); + out.push_str(&format!("- {}\n", shown.join(" · "))); + continue; + } + out.push_str(line); + out.push('\n'); + } + // lines() 逐行补换行,源文本不以换行结尾时尾部多出一个,与源对齐 + if !s.ends_with('\n') { + out.pop(); + } + out } #[cfg(test)] @@ -349,11 +384,27 @@ mod tests { assert!(approved_flags.contains(&true) && approved_flags.contains(&false)); } - /// Markdown 转义:卡片详情中的 markdown 语法字符被转义 + /// 出站净化:标题转粗体行,表格行转列表行,表头分隔行跳过 #[test] - fn escape_md_works() { - assert_eq!(escape_md("a*b_c`d\\e"), "a\\*b\\_c\\`d\\\\e"); - // 走查 #7:链接/标题/引用语法字符也须转义,防用户内容注入卡片结构 - assert_eq!(escape_md("[x](http://e)#h (y)"), "\\[x\\]\\(http://e\\)\\#h \\\\(y\\)"); + fn lark_friendly_converts_headings_and_tables() { + let md = "# 标题\n| a | b |\n|---|---|\n| 1 | 2 |"; + assert_eq!(md_to_lark_friendly(md), "**标题**\n- a · b\n- 1 · 2"); + } + + /// 出站净化:代码围栏内外均保留原样,围栏外的标题照常改写 + #[test] + fn lark_friendly_preserves_code_fence() { + let md = "前文\n```\n# 不是标题\n| 不 | 是表格 |\n```\n## 后文"; + assert_eq!( + md_to_lark_friendly(md), + "前文\n```\n# 不是标题\n| 不 | 是表格 |\n```\n**后文**" + ); + } + + /// 出站净化:普通行原样透传(lark_md 原生支持粗体/链接,不额外转义) + #[test] + fn lark_friendly_passes_plain_lines() { + let md = "**粗** 与 [链接](http://e)\n普通行"; + assert_eq!(md_to_lark_friendly(md), md); } } diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index c3265d1..3db5601 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -654,6 +654,14 @@ pub async fn ai_chat_send( // None/空 → 无 @ mention(向后兼容,仅 / 技能走 skill 参数);非空 → 按 kind 投影成 Augmentation。 mention_spans: Option>, ) -> Result { + // /status 进度查询命令(2026-10-06):双端话术引用 /status 但命令此前不存在, + // 发送即被当任务烧一轮 LLM。置于技能解析之前拦截,零 LLM 零 spawn。飞书渠道在 + // feishu_channel.rs 受理前有同命令拦截,本分支补齐桌面入口。 + if message.trim() == "/status" { + return reply_status_summary(&app, &state, conversation_id.clone()) + .await + .map(|_| "ok".to_string()); + } // H1 根修:前导 `/技能名` 解析。此前技能唯一触发路是前端浮层选中(skill 参数), // 用户直接打字 `/name xxx` 时整串当普通消息,技能注入零发生。 // 此处对照 skills_cached() 已扫技能名(含 commands/*.md 来源)匹配前导斜杠令牌, @@ -692,6 +700,87 @@ pub async fn ai_chat_send( .map(|_| "ok".to_string()) } +/// /status 进度查询(2026-10-06):命令拦截应答,零 LLM 零 spawn。 +/// +/// 读 conv_states 生命周期态 + per_conv 轮次/最近输出,拼几行摘要作为 assistant 消息落库。 +/// 事件序列与常规轮收尾同契约(AiAgentRound 建泡 → AiTextDelta 填文 → AiCompleted 复位), +/// 缺完成事件前端 doSend 已置的 streaming 无法复位即悬挂。不迁移 ConvState(只读诊断), +/// 不写 working_context(/status 是探针不是任务目标)。 +/// +/// 会话定位与 spawn_agentic_session 同序(入参优先 → active);无会话/会话已删时 Err 透出 +/// (前端 toast),零落库零事件。轮次/目标快照短锁读即释,不与 loop 写交叉。 +async fn reply_status_summary( + app: &AppHandle, + state: &AppState, + conversation_id: Option, +) -> Result<(), String> { + let target = match conversation_id.filter(|s| !s.is_empty()) { + Some(id) => id, + None => state + .ai_session + .lock() + .await + .active_conversation_id + .clone() + .ok_or_else(|| "📭 当前无会话,发送消息即自动创建。".to_string())?, + }; + let (state_line, pinned_goals) = { + let cs = state.conv_states.get(&target); + let session = state.ai_session.lock().await; + let conv = session + .conv_read(&target) + .ok_or_else(|| "📭 会话不存在或已删除。".to_string())?; + let line = match cs { + ConvState::Generating => { + format!("🚀 任务执行中(已 {} 轮),完成后自动回填本会话。", conv.iteration_used) + } + ConvState::Compressed => "⏳ 上下文压缩中,完成后自动续跑。".to_string(), + ConvState::Stopping => "⏹ 收到停止指令,正在收尾退出。".to_string(), + ConvState::Error => "❌ 上次任务出错,处理后重发消息即可重试。".to_string(), + ConvState::Idle => "✅ 会话空闲,无进行中任务。".to_string(), + }; + (line, conv.pinned_goals.clone()) + }; + // 最近输出预览(复用 agentic 收敛的末条 assistant 取数;空=尚无产出,首 token/工具执行中) + let recent = super::super::agentic::last_active_assistant_text(&state.ai_session, &target).await; + let recent_line = if recent.trim().is_empty() { + "最近输出:暂无".to_string() + } else { + let snippet: String = recent.chars().take(80).collect(); + format!("最近输出:{snippet}") + }; + let summary = format!("{state_line}\n{recent_line}"); + + // 问答成对落库(重开可见,LLM 历史连贯);save 增量路径与 loop 同 per-conv 锁串行 + { + let mut session = state.ai_session.lock().await; + let conv = session.conv(&target); + conv.messages.push(ChatMessage::user("/status")); + conv.messages.push(ChatMessage::assistant(summary.clone())); + } + save_conversation(&state.ai_session, &state.db, &target, None, None, true).await; + // pinned_goals 原样透传防前端目标区回退(与 agentic completed_event 同口径) + emit_ai_chat_event(app, AiChatEvent::AiAgentRound { round: 1, conversation_id: Some(target.clone()) }); + emit_ai_chat_event(app, AiChatEvent::AiTextDelta { delta: summary, conversation_id: Some(target.clone()) }); + emit_ai_chat_event( + app, + AiChatEvent::AiCompleted { + total_tokens: 0, + prompt_tokens: 0, + completion_tokens: 0, + prompt_cache_hit_tokens: 0, + prompt_cache_miss_tokens: 0, + reasoning_tokens: 0, + is_estimated: false, + incomplete: None, + incomplete_reason: None, + conversation_id: Some(target), + pinned_goals, + }, + ); + Ok(()) +} + /// 发送消息并启动 agentic 会话的共用核心(纯搬迁自 ai_chat_send,行为逐字等价)。 /// /// 调用方:ai_chat_send IPC(前端对话,pin_active=true)/ 工作流 AiAgentNode 编排会话 diff --git a/src-tauri/src/feishu_channel.rs b/src-tauri/src/feishu_channel.rs index 75ac38b..f510211 100644 --- a/src-tauri/src/feishu_channel.rs +++ b/src-tauri/src/feishu_channel.rs @@ -26,11 +26,22 @@ const KV_FEISHU_CONFIG: &str = "feishu-config"; /// app_settings 键:飞书 chat_id → conversation_id 映射(持久化,跨重启续会话) const KV_CHAT_MAP: &str = "feishu-chat-map"; +/// #26 占用期话术改实况:委托制下无真排队(不入队),话术禁承诺「任务完成后处理」, +/// 如实告知未受理并给出路。#20 受理守卫拒绝与 spawn 并发护栏拒绝共用同一口径 +const QUEUE_BUSY_MSG: &str = + "⏳ 该会话已有任务占用中,本条未受理(暂不入队);可稍后重发,或先 /status 查看进度。"; + /// 19 outbound 统一出口:渠道启动成功后的全局 client 句柄(once 填充,渠道停用为空)。 /// push_to_chat 主动推送复用长连接 client,不临时建连;进程生命周期内渠道只启一次, /// OnceLock 语义匹配(重启应用才可能重配)。 static ACTIVE_CLIENT: std::sync::OnceLock<&'static LarkClient> = std::sync::OnceLock::new(); +/// 34 滚动汇总卡句柄:chat_id 对 (在途事务卡 message_id, 已累计正文)。同 chat 多步 +/// 进度推送原地追加为一卡,防逐条直推刷屏;内存句柄重启即失(重启后首推另起新卡)。 +/// std Mutex 只做句柄与字符串存取,不跨 await 持有,短锁语义 +static ROLLING_CARDS: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + #[derive(Debug, Clone, Deserialize)] struct FeishuConfig { #[serde(default)] @@ -62,7 +73,8 @@ struct Ctx { chat_locks: Arc>>>>, /// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流 last_alive: Arc, - /// D3 在途会话数:spawn 到投递完成期间 >0,看门狗要求为 0 才告警,防有回复待投递时误报 + /// D3 在途会话数:消息受理(handle_message 白名单通过)到任务投递完成期间 >0, + /// 看门狗要求为 0 才告警,防有回复待投递时误报 inflight: Arc, } @@ -241,6 +253,13 @@ async fn handle_message( // D3:任何入站消息都是事件管道存活证据,先于一切过滤更新活性时间戳 ctx.last_alive.store(now_secs(), Ordering::SeqCst); if msg.message_type != "text" { + // #25 非文本消息静默丢修复:回提示卡(D7 媒体能力前的兜底),不再零反馈 + let _ = reply_markdown_card( + &ctx.client, + &msg.message_id, + "📎 暂仅支持文本消息,图片/文件等媒体能力规划中", + ) + .await; return; } if msg.chat_type == "group" && !msg.mentioned_bot { @@ -287,6 +306,34 @@ async fn handle_message( return; } + // #21 /status 查询当前会话任务状态(话术曾引用不存在的命令,用户发 /status 被当任务 + // 烧一轮 LLM)。只读不 spawn:查 conv 绑定与活跃态给进度摘要。 + if text == "/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; + } + // ── 纠偏注入旁路(todo.md 16,2026-10-06):会话运行中 → 直接注入运行中 conv,不排队 ── // 排队语义的痛点:per-chat 锁把补充指令排到 agent 完成后才处理,长任务期间纠偏无效。 // 注入语义:conv 活跃(Generating/Compressed)时直接写 per_conv.messages,loop 下一轮 @@ -329,6 +376,14 @@ async fn handle_message( // D3 在途计数已前移至 handle_message 白名单通过处(探活口径修正,18),此处不再重复计数 let state = ctx.app.state::(); + // #20 受理守卫:发票(受理卡)前先过引擎读侧闸。不可启动(前任务占用/压缩暂存/停止中) + // 直接回实况话术,不发票、不 spawn、不注册桥与流式任务,堵「受理卡已发后续零通知」 + // 的伪受理断层(13:38 实证形态) + if !state.conv_states.can_accept_request(&conv_id) { + feishu_log(&format!("受理守卫拒绝 chat={chat_id} conv={conv_id}")); + deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, QUEUE_BUSY_MSG).await; + return; + } // 通知桥注册(#14):oneshot 接收端有意 drop——两段式下终版走事件广播 // (deliver_completion_report_if_waited 的 AiAgentFinished publish),桥条目存在的 // 唯一作用是 has_agent_completion_waiter 为真(触发引擎层取报告+广播);rx.drop 触发 @@ -384,7 +439,8 @@ async fn handle_message( // 吞掉真实原因,用户误判 bot 坏了重试白撞。按引擎层错误文案分类: // 生成中→明示任务进行中不会丢;其余→附截断原始原因(不吞因)。 let user_msg = if e.contains("正在生成") || e.contains("生成中") { - "⏳ 该会话已有任务进行中,本条将在任务完成后处理;可稍后重发,或先 /status 查看进度。".to_string() + // #26 占用期话术改实况:委托制无真排队,禁承诺「任务完成后处理」 + QUEUE_BUSY_MSG.to_string() } else { format!("❌ 会话启动失败:{}`,请稍后重试", truncate(&e, 120)) }; @@ -411,28 +467,33 @@ async fn handle_message( // 彻底消除「受理卡与终版两个 HTTP 请求网络乱序,终版被受理卡覆盖丢失」的竞态窗口。 drop(deliver_tx); let _ = stream_task.await; // 等流式任务退出(快,只走通道关闭分支) - let ack = format!("✅ 已受理,后台执行中\n任务:{task_brief}"); + let ack = format!("🚀 已受理,后台执行中\n任务:{task_brief}"); feishu_log(&format!("受理返回 chat={chat_id} conv={conv_id}")); // 完成订阅任务:独立于消息处理生命周期,直到 AiAgentFinished 广播到达投递终版。 // (spawn Err 路径在上方已 return,永不到达此处——watcher 依赖引擎层正常完成广播) + // #20 附带:inflight 生命周期跟任务走,guard 移交 watcher 持有。原 guard 随本函数 + // 返回即 drop,受理后任务还在后台跑计数却归零,探活误报窗口重开;任务终局(watcher + // 各退出路径)guard drop 自然回落。与本函数前置计数的短暂重叠对看门狗无影响(判据 >0) + let inflight_guard = InflightGuard(ctx.inflight.clone()); tauri::async_runtime::spawn(async move { - run_completion_watcher(ctx, conv_id, placeholder_id, message_id, chat_id, ack).await; + run_completion_watcher(ctx, conv_id, placeholder_id, message_id, chat_id, ack, inflight_guard) + .await; }); } -/// 完成事件订阅任务(#14 两段式):占位卡唯一写者。入场先投受理卡,再消费 -/// ai_event_bus 的 AiAgentFinished(本 conv)原地 edit_card 为终版。引擎层 -/// deliver_completion_report_if_waited 是广播单一发出点,watcher 纯消费无状态。 +/// #27 快任务宽限窗:受理卡发出后的事件驱动判定窗(原入场 elapsed>=3s 恒假死分支)。 +/// 窗内到达的完成事件由广播 rx 缓冲零丢失;窗结束时以「conv 非活跃 + 边界内有产出」 +/// 判定秒级任务已完成。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 纯消费无状态。 /// -/// 快任务竞态兜底(走查🟡#1 加门槛):广播可能先于订阅(秒级任务)——立即回查 -/// 一次,但须同时满足 conv 非活跃 + 回查时间晚于受理(防 spawn 返回但 loop 尚未 -/// 置 active 的时序窗口误判) + 文本非空,三者齐才直投终版跳过受理卡(卡片写入 -/// 单任务串行,无网络乱序覆盖)。任一不满足则先落受理卡再进事件等待。 -/// -/// 超时语义(走查🟡#2 修正):35min(> loop 30min 上限)超时投降级提示后**不退出**, -/// 继续挂事件等待——任务后台不丢,晚到的完成事件仍会送到卡片;watcher 与任务 -/// 同生命周期,彻底无果(引擎侧异常未广播)才在二次超时后弃守。 +/// 超时语义:35min(> loop 30min 上限)首超时回查边界内产出,无果投降级提示但不退出 +/// (任务后台不丢,晚到完成事件仍送达);二次超时真弃守(#23:两轮均无事件且无边界内 +/// 产出,认任务失联,停止重刷)。 async fn run_completion_watcher( ctx: Arc, conv_id: String, @@ -440,47 +501,73 @@ async fn run_completion_watcher( message_id: String, chat_id: String, ack: String, + // #20:计数生命周期跟任务走,guard 持有至 watcher 退出 + _inflight: InflightGuard, ) { let state = ctx.app.state::(); let mut rx = state.ai_event_bus.subscribe(); - let accepted_at = tokio::time::Instant::now(); - // 快任务兜底:立即回查一次(事件可能已错过);三分量门槛防时序窗口误判 - let settled = !state.conv_states.is_active(&conv_id) - && accepted_at.elapsed() >= std::time::Duration::from_secs(3) - && matches!(last_assistant_reply(&state.db, &conv_id).await, - Some(t) if !t.trim().is_empty()); - if settled { - let text = last_assistant_reply(&state.db, &conv_id) - .await - .unwrap_or_default(); - feishu_log(&format!("watcher 快任务兜底 conv={conv_id} len={}", text.len())); - deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await; - return; - } + // #22 受理时刻(Unix 毫秒):快任务兜底与超时回查只认此后落库的 assistant 消息, + // 防把上一任务旧回答冒充本次结果(无 timestamp 的老消息视为不在边界内) + let accepted_ms = std::time::SystemTime::now() + .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; - // 事件等待循环:首超时 35 分钟投降级提示不退出,继续等晚到事件;二次超时弃守。 + // #27 快任务兜底:受理卡发出后开宽限窗再判定,事件驱动替代恒假的时间比较。 + // 秒级任务的完成事件可能先于本订阅已广播完(错过),窗结束时 conv 已归非活跃且 + // 边界内有产出,即判快任务已完成,直投终版 + tokio::time::sleep(FAST_TASK_GRACE).await; + if !state.conv_states.is_active(&conv_id) { + 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; + return; + } + } + } + // 事件等待循环:首超时投降级提示不退出;二次超时真弃守(#23) let timeout = std::time::Duration::from_secs(2100); + let mut timeouts = 0usize; loop { let text = tokio::select! { item = rx.recv() => match item { Ok(v) => { - if v.get("type").and_then(|t| t.as_str()) != Some("AiAgentFinished") { - continue; - } if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) { continue; } - // 广播语义:agent_id==conv_id 时 summary 即本会话最终报告(引擎层构造) - v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string() + match v.get("type").and_then(|t| t.as_str()) { + // #23 防任务泄漏:引擎错误路径只发 AiError 不发 AiAgentFinished(引擎侧 + // 补广播在 chat.rs 域,本文件域禁触),渠道侧消费 AiError 视同终局, + // 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)) + } + Some("AiAgentFinished") => { + // 广播语义:agent_id==conv_id 时 summary 即本会话最终报告(引擎层构造) + v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string() + } + _ => continue, + } } Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, Err(tokio::sync::broadcast::error::RecvError::Closed) => return, }, _ = tokio::time::sleep(timeout) => { + 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 回查兜底(任务可能刚完成事件已发完);回查有果照投终版, - // 无果投降级提示并**继续等待**(走查🟡#2:超时不弃投递,晚到事件仍送达) - match last_assistant_reply(&state.db, &conv_id).await { + // 无果投降级提示并继续等待(晚到事件仍送达) + 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(); @@ -662,15 +749,23 @@ fn stream_preview( format!("{head}{body}\n▌") } -/// markdown 回复:interactive 卡片承载(lark_md 原生渲染标题/列表/粗体,text 类型会符号裸露) -async fn send_markdown_card(client: &LarkClient, message_id: &str, chat_id: &str, text: &str) { +/// markdown 回复:interactive 卡片承载(lark_md 原生渲染标题/列表/粗体,text 类型会符号裸露)。 +/// 出站先过 38a 净化(md_to_lark_friendly),成功返回新卡片 message_id(34 滚动汇总卡 +/// 绑定句柄用,取 id 方式与 reply_markdown_card 同款);彻底失败回退纯文本保可达,返 None +async fn send_markdown_card( + client: &LarkClient, + message_id: &str, + chat_id: &str, + text: &str, +) -> Option { + let text = &md_to_lark_friendly(text); let card = Card::new().element(Element::Markdown(MdElement::new(text.to_string()))); let document = match CardDocument::new(card) { Ok(d) => d, Err(e) => { tracing::warn!("[feishu] 卡片校验失败,回退纯文本: {e}"); reply_message(client, message_id, chat_id, text).await; - return; + return None; } }; let content = serde_json::to_string(document.card()).unwrap_or_default(); @@ -680,16 +775,25 @@ async fn send_markdown_card(client: &LarkClient, message_id: &str, chat_id: &str ..Default::default() }; let opt = &RequestOption::default(); - let sent = if !message_id.is_empty() { - client.channel_messaging().reply(message_id, &input, opt).await.is_ok() - } else { - false - }; - if !sent { - if let Err(e) = client.channel_messaging().send(&input, opt).await { + let mut replied: Option = None; + if !message_id.is_empty() { + if let Ok(sent) = client.channel_messaging().reply(message_id, &input, opt).await { + if !sent.message_id.is_empty() { + replied = Some(sent.message_id); + } + } + } + if replied.is_some() { + return replied; + } + match client.channel_messaging().send(&input, opt).await { + Ok(sent) if !sent.message_id.is_empty() => Some(sent.message_id), + Ok(_) => None, + Err(e) => { tracing::error!("[feishu] 卡片发送失败: {e}"); // 卡片彻底失败仍回退纯文本,保可达 reply_message(client, message_id, chat_id, text).await; + None } } } @@ -754,8 +858,9 @@ async fn update_card(client: &LarkClient, message_id: &str, text: &str) -> bool } } -/// 结果投递统一入口:先做表格降级(md 表格在卡片 markdown 组件渲染不稳), -/// 优先原地更新占位卡片(占位与结果合一);占位缺失或更新失败回退新发卡片(内含纯文本兜底) +/// 结果投递统一入口:先做出站净化(38a,update_card 原地路径不经 send_markdown_card, +/// 须在此净化;回退新发路径二次进入幂等)与裁剪,优先原地更新占位卡片(占位与结果合一); +/// 占位缺失或更新失败回退新发卡片(内含纯文本兜底) async fn deliver_result( client: &LarkClient, placeholder_id: &Option, @@ -763,20 +868,24 @@ async fn deliver_result( chat_id: &str, text: &str, ) { - let text = clip_for_card(&downgrade_md_tables(text), 8000); + let text = clip_for_card(&md_to_lark_friendly(text), 8000); if let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) { if update_card(client, pid, &text).await { return; } feishu_log("占位卡片更新失败,回退新发卡片"); } - send_markdown_card(client, message_id, chat_id, &text).await; + let _ = send_markdown_card(client, message_id, chat_id, &text).await; } /// 19 outbound 统一出口:主动推送(无占位卡上下文的外部推送方——挂机进度汇总/巡检通知 /// 等将来形态)。降级/裁剪/卡片链与 deliver_result 同源(禁再开独立发送路径绕过降级链, /// 11:23 实证:绕过链的回复带完整 md 表格未降级,踩回表格渲染不稳坑)。 +/// 34 滚动汇总卡:同 chat 在途事务卡存在时原地追加(句柄见 ROLLING_CARDS),✅/❌ 终态 +/// 更新后清句柄下条另起新卡;原地更新失败清句柄回退新发并重绑(比照 deliver_result 兜底)。 /// 推送目标:chat_map 最近映射的会话;渠道未启动(无 client)或无映射时静默跳过。 +/// 19 正典出口预留给将来推送方(挂机进度汇总/巡检通知),当前无调用方 +#[allow(dead_code)] pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str) { let Some(client) = ACTIVE_CLIENT.get() else { return; // 渠道未启动:推送无落点,静默跳过(调用方自行决定是否记日志) @@ -792,10 +901,46 @@ pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str) { let Some(chat_id) = map.keys().next().cloned() else { return; }; - let processed = clip_for_card(&downgrade_md_tables(text), 8000); + // 34 终态判据:✅/❌ 开头即事务收尾,更新后不再续卡 + let terminal = is_terminal_push(text); + let rolling = ROLLING_CARDS.get_or_init(|| std::sync::Mutex::new(HashMap::new())); + let handle = rolling.lock().ok().and_then(|m| m.get(&chat_id).cloned()); + if let Some((card_id, mut acc)) = handle { + acc.push('\n'); + acc.push_str(text); + let rendered = clip_for_card(&md_to_lark_friendly(&acc), 8000); + if update_card(client, &card_id, &rendered).await { + if terminal { + if let Ok(mut m) = rolling.lock() { + m.remove(&chat_id); + } + } + feishu_log(&format!("push_to_chat 追加 chat={chat_id} len={}", rendered.len())); + return; + } + // 原地更新失败(卡片被撤回/JSON 超限等):清句柄,新发并重绑 + if let Ok(mut m) = rolling.lock() { + m.remove(&chat_id); + } + let rebound = send_markdown_card(client, "", &chat_id, &rendered).await; + if let (false, Some(new_id)) = (!terminal, rebound) { + if let Ok(mut m) = rolling.lock() { + m.insert(chat_id.clone(), (new_id, acc)); + } + } + feishu_log(&format!("push_to_chat 回退重绑 chat={chat_id} len={}", rendered.len())); + return; + } + let processed = clip_for_card(&md_to_lark_friendly(text), 8000); // 主动推送无 message_id 上下文:send_markdown_card 的 reply 分支跳过,直接 chat 卡片 - send_markdown_card(client, "", &chat_id, &processed).await; - feishu_log(&format!("push_to_chat chat={chat_id} len={}", processed.len())); + let sent_id = send_markdown_card(client, "", &chat_id, &processed).await; + // 首推开卡绑定句柄;终态不绑,下条另起新卡 + if let (false, Some(new_id)) = (!terminal, sent_id) { + if let Ok(mut m) = rolling.lock() { + m.insert(chat_id.clone(), (new_id, text.to_string())); + } + } + feishu_log(&format!("push_to_chat chat={chat_id} len={} terminal={terminal}", processed.len())); } /// 卡片正文长度防御:飞书卡片 JSON 约 30KB 上限,超长会让原地更新与新发双双失败, @@ -890,6 +1035,67 @@ fn is_table_row(line: &str) -> bool { line.trim_start().starts_with('|') } +/// 38a 出站 md 净化:md 转飞书友好格式。lark_md 不支持标题/表格等全谱语法,LLM 产出的 +/// #/##/| 原样出站即符号裸露(38 实证)。规则母本三条:标题行转粗体行、表格转列表行 +/// (复用 downgrade_md_tables 单一实现)、代码块围栏内原样保留,其余原样。 +/// 幂等:粗体行/列表行二次进入不再变形,send_markdown_card 与 push_to_chat 链重复 +/// 调用安全。df-im-gateway StreamText 出站侧做净化时须对齐同一规则(同规则互引, +/// 38 双层防线的出站兜底层,以本函数为正典)。 +pub fn md_to_lark_friendly(text: &str) -> String { + let mut out: Vec = Vec::new(); + let mut plain: Vec<&str> = Vec::new(); + let mut in_fence = false; + for line in text.lines() { + if line.trim_start().starts_with("```") { + if !in_fence { + out.extend(sanitize_plain_lines(&plain)); + plain.clear(); + } + in_fence = !in_fence; + out.push(line.to_string()); + continue; + } + if in_fence { + out.push(line.to_string()); + } else { + plain.push(line); + } + } + out.extend(sanitize_plain_lines(&plain)); + out.join("\n") +} + +/// 围栏外逐段净化:表格降级 + 标题行转粗体行 +fn sanitize_plain_lines(lines: &[&str]) -> Vec { + let joined = lines.join("\n"); + downgrade_md_tables(&joined) + .lines() + .map(|line| match heading_content(line) { + Some(content) if !content.is_empty() => format!("**{content}**"), + _ => line.to_string(), + }) + .collect() +} + +/// 标题行识别:1 到 6 个 # 后接空格(CommonMark 形态,防 #标签 之类误伤) +fn heading_content(line: &str) -> Option<&str> { + let t = line.trim_start(); + let hashes = t.chars().take_while(|&c| c == '#').count(); + if (1..=6).contains(&hashes) { + if let Some(rest) = t[hashes..].strip_prefix(' ') { + return Some(rest.trim_end()); + } + } + None +} + +/// 34 终态判据:文本以 ✅/❌ 开头(对齐看板符号语义规范:✅ 仅限完成且成功,❌ 失败须附原因) +/// 视为事务收尾,滚动汇总卡更新后清句柄,下条推送另起新卡 +fn is_terminal_push(text: &str) -> bool { + let t = text.trim_start(); + t.starts_with('✅') || t.starts_with('❌') +} + /// 分隔行形态:| 开头,且整行只由 | - : 与空白构成并含至少一段 --- fn is_table_separator(line: &str) -> bool { let t = line.trim(); @@ -935,14 +1141,30 @@ fn truncate(s: &str, n: usize) -> String { s.chars().take(n).collect() } -/// 回查该会话最后一条非空 assistant 文本(通知桥空内容时的事实源兜底) -async fn last_assistant_reply(db: &df_storage::db::Database, conv_id: &str) -> Option { +/// #22 时间戳过滤判据(纯函数,单测锁定):给定受理时刻时只认其后落库的消息, +/// 消息缺时间戳视为不在边界内(宁缺勿冒充旧回答);无受理时刻不过滤 +fn msg_after_acceptance(msg_ts: Option, accepted_ms: Option) -> bool { + accepted_ms.map_or(true, |ts| msg_ts.map_or(false, |t| t > ts)) +} + +/// 回查该会话最后一条非空 assistant 文本(通知桥空内容时的事实源兜底)。 +/// #22 任务边界:only_after_ms 传入受理时刻(Unix 毫秒)时只认该时刻之后的消息, +/// 防 35min 兜底回查把上一任务旧回答冒充本次结果;None 不过滤(旧语义)。 +async fn last_assistant_reply( + db: &df_storage::db::Database, + conv_id: &str, + only_after_ms: Option, +) -> Option { use df_storage::crud::AiMessageRepo; let repo = AiMessageRepo::new(db); let msgs = repo.list_by_conversation(conv_id).await.ok()?; msgs.into_iter() .rev() - .find(|m| m.role == "assistant" && !m.content.trim().is_empty()) + .find(|m| { + m.role == "assistant" + && !m.content.trim().is_empty() + && msg_after_acceptance(m.timestamp, only_after_ms) + }) .map(|m| m.content) } @@ -954,7 +1176,7 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap