diff --git a/crates/df-im-gateway/src/feishu.rs b/crates/df-im-gateway/src/feishu.rs index d797ea5..5cca301 100644 --- a/crates/df-im-gateway/src/feishu.rs +++ b/crates/df-im-gateway/src/feishu.rs @@ -28,14 +28,20 @@ use crate::config::FeishuConfig; /// 飞书适配器:channel 归一化消息 + 卡片审批,单飞书应用单实例 #[derive(Clone)] pub struct FeishuAdapter { - client: LarkClient, + /// 常驻 client(&'static,构造期一次性 Box::leak)。D1(u-claw 教训:热重建不关旧 + /// 连接→僵尸资源):leak 只在 new()/offline() 各发生一次;run() 重连重入只重建 + /// Channel 不再重复 leak,防频繁断线场景 LarkClient 内存阶梯上涨。出站 dispatch + /// 与 Channel 构建共用同一份引用。 + client: &'static LarkClient, /// 流式会话注册表:chat_id → 已发出的流式消息 message_id(用于 edit_text 覆写) streams: Arc>>, } impl FeishuAdapter { pub fn new(cfg: &FeishuConfig) -> anyhow::Result { - let client = LarkClient::builder(&cfg.app_id, &cfg.app_secret).build()?; + let client: &'static LarkClient = Box::leak(Box::new( + LarkClient::builder(&cfg.app_id, &cfg.app_secret).build()?, + )); Ok(Self { client, streams: Arc::new(tokio::sync::Mutex::new(HashMap::new())), @@ -44,8 +50,11 @@ impl FeishuAdapter { /// 构造失败时的离线哨兵:client 用占位凭证(不启动 run,dispatch 记日志即弃) pub fn offline() -> Self { - let client = LarkClient::builder("offline", "offline").build() - .expect("占位凭证构造 LarkClient 不发网络请求,必成功"); + let client: &'static LarkClient = Box::leak(Box::new( + LarkClient::builder("offline", "offline") + .build() + .expect("占位凭证构造 LarkClient 不发网络请求,必成功"), + )); Self { client, streams: Arc::new(tokio::sync::Mutex::new(HashMap::new())), @@ -68,16 +77,16 @@ impl PlatformAdapter for FeishuAdapter { async fn run(&self, events: Sender) -> anyhow::Result<()> { // 长连接模式:空 verification_token/encrypt_key(免公网回调) let dispatcher = EventDispatcher::new("", ""); - let client = self.client.clone(); + let client = self.client; // 入站事件通道复制两份,分别给消息/卡片回调闭包 let events_for_msg = events.clone(); let events_for_card = events; let channel: Channel<'static> = { - // Channel<'a> 借用 client;用 Box::leak 提升为 'static(适配器生命周期即进程生命周期) - let leaked: &'static LarkClient = Box::leak(Box::new(client)); - Channel::builder(leaked, dispatcher) + // client 已是 &'static(构造期一次性 leak);run() 重连重入只重建 Channel, + // 不再重复 Box::leak(D1:防每次重入泄漏一份 LarkClient) + Channel::builder(client, dispatcher) .policy(ChannelPolicy::default().require_mention(false)) .on_message(move |msg| { let events = events_for_msg.clone(); diff --git a/src-tauri/src/feishu_channel.rs b/src-tauri/src/feishu_channel.rs index d533bad..1f65392 100644 --- a/src-tauri/src/feishu_channel.rs +++ b/src-tauri/src/feishu_channel.rs @@ -6,12 +6,14 @@ //! 会话复用 spawn_agentic_session 通用核心(pin_active=false,与编排节点同款调用方), //! 完成通知走 AppState 通知桥(register/notify_agent_completion,编排节点先例)。 //! -//! 第一批:文本收发(整段回复);流式 edit_text/审批卡片放第二批(内核留存于网关 feishu.rs)。 +//! 第一批:文本收发(整段回复);流式预览已接入(订阅 ai_event_bus 的 AiTextDelta 节流刷占位卡片)。 use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; use tauri::Manager; +use larksuite_oapi_sdk_rs::card::v1::{Card, CardDocument, Element, Markdown as MdElement}; use larksuite_oapi_sdk_rs::channel::{Channel, ChannelPolicy, SendInput}; use larksuite_oapi_sdk_rs::{EventDispatcher, LarkClient, LarkError, RequestOption}; use serde::Deserialize; @@ -47,9 +49,16 @@ fn norm(raw: Option) -> String { struct Ctx { app: tauri::AppHandle, client: LarkClient, - whitelist: Vec, + /// 白名单缓存(D2 热加载):每消息现读 KV 刷新,读取失败沿用本缓存,改配置免重启 + whitelist: Arc>>, /// chat_id → conversation_id(启动时从 KV 恢复,变更回写) chat_map: Arc>>, + /// per-chat 串行锁:同会话消息排队处理,防通知桥单槽跨消息串扰(P1-1) + chat_locks: Arc>>>>, + /// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流 + last_alive: Arc, + /// D3 在途会话数:spawn 到投递完成期间 >0,看门狗要求为 0 才告警,防有回复待投递时误报 + inflight: Arc, } /// 启动入口:lib.rs setup 调用。未启用/配置缺失/构造失败一律记日志返回,不阻塞应用启动。 @@ -58,22 +67,38 @@ pub fn start_if_enabled(app: tauri::AppHandle) { let db = app.state::().db.clone(); let repo = SettingsRepo::new(&db); let raw = norm(repo.get(KV_FEISHU_CONFIG).await.ok().flatten()); + // 早退点全部补 feishu_log:tracing 只进控制台,用户排障只看得到 feishu-channel.log, + // 四个早退原因各有不同,不落盘则"渠道未连接"完全盲查(2026-10-05 排查实证) if raw.is_empty() { tracing::info!("[feishu] feishu-config 未配置,渠道不启动"); + feishu_log("启动早退: feishu-config 未配置"); return; } - let Ok(cfg) = serde_json::from_str::(&raw) else { - tracing::error!("[feishu] feishu-config 解析失败,渠道不启动"); - return; + let cfg = match serde_json::from_str::(&raw) { + Ok(c) => c, + Err(e) => { + tracing::error!("[feishu] feishu-config 解析失败,渠道不启动: {e}"); + feishu_log(&format!("启动早退: feishu-config 解析失败: {e}")); + return; + } }; if !cfg.enabled || cfg.app_id.is_empty() || cfg.app_secret.is_empty() { - tracing::info!("[feishu] enabled=false 或凭证缺失,渠道不启动"); + let reason = if !cfg.enabled { + "enabled=false" + } else if cfg.app_id.is_empty() { + "app_id 缺失" + } else { + "app_secret 缺失" + }; + tracing::info!("[feishu] {reason},渠道不启动"); + feishu_log(&format!("启动早退: {reason}")); return; } let client = match LarkClient::builder(&cfg.app_id, &cfg.app_secret).build() { Ok(c) => c, Err(e) => { tracing::error!("[feishu] LarkClient 构造失败: {e}"); + feishu_log(&format!("启动早退: LarkClient 构造失败: {e}")); return; } }; @@ -85,43 +110,129 @@ pub fn start_if_enabled(app: tauri::AppHandle) { let ctx = Arc::new(Ctx { app, client: client.clone(), - whitelist: cfg.whitelist, + whitelist: Arc::new(tokio::sync::RwLock::new(cfg.whitelist)), chat_map: Arc::new(tokio::sync::Mutex::new(restored)), + chat_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + last_alive: Arc::new(AtomicI64::new(now_secs())), + inflight: Arc::new(AtomicUsize::new(0)), }); - if let Err(e) = run_loop(ctx, client).await { - tracing::error!("[feishu] 长连接退出: {e}"); + // D1 僵尸资源修复:client 的 'static 化只做一次(原在 run_loop 内,每次重连都 + // Box::leak 一份 LarkClient,频繁断线场景内存阶梯上涨);Channel::builder 仅借用 + // 引用,重连循环内复用同一 leaked 引用,只重建 Channel/dispatcher + let leaked: &'static LarkClient = Box::leak(Box::new(client)); + // 断线自动重连:网络波动/服务端断开导致 start 退出时 backoff 重试,防 bot 静默死 + let mut backoff = 2u64; + loop { + if let Err(e) = run_loop(ctx.clone(), leaked).await { + tracing::error!("[feishu] 长连接退出,{}s 后重连: {e}", backoff); + feishu_log(&format!("长连接断开,{}s 后重连: {e}", backoff)); + } else { + tracing::info!("[feishu] 长连接正常退出,{}s 后重连", backoff); + } + tokio::time::sleep(std::time::Duration::from_secs(backoff)).await; + backoff = (backoff * 2).min(300); } }); } -async fn run_loop(ctx: Arc, client: LarkClient) -> anyhow::Result<()> { +async fn run_loop(ctx: Arc, client: &'static LarkClient) -> anyhow::Result<()> { let dispatcher = EventDispatcher::new("", ""); let ctx_msg = ctx.clone(); - // Channel<'a> 借用 client;Box::leak 提升为 'static(渠道生命周期即进程生命周期) - let leaked: &'static LarkClient = Box::leak(Box::new(client)); - let channel: Channel<'static> = Channel::builder(leaked, dispatcher) + let ctx_ready = ctx.clone(); + // D3 看门狗与单次连接生命周期绑定:run_loop 退出时 abort,不逐重连累积 + let watchdog = tauri::async_runtime::spawn(run_liveness_watchdog(ctx.clone())); + // 新连接建立即视为存活起点(内部断线重连由 on_ready 再刷新) + ctx.last_alive.store(now_secs(), Ordering::SeqCst); + // Channel<'a> 借用 client;'static 由 start_if_enabled 在重连循环外一次性 Box::leak + // 提供(D1:原在此处 leak,每次重连泄漏一份 LarkClient) + let channel: Channel<'static> = Channel::builder(client, dispatcher) .policy(ChannelPolicy::default().require_mention(false)) + // D3:SDK 内部每次断线重连成功都会回调,刷新活性防断流误报(SDK 读空闲超时 + // 自愈真死连接后,断流计时从这里归零) + .on_ready(move || { + ctx_ready.last_alive.store(now_secs(), Ordering::SeqCst); + }) .on_message(move |msg| { let ctx = ctx_msg.clone(); async move { - handle_message(ctx, msg).await; + // 异步处理:回调内绝不等 agent 跑完(单消息长任务会阻塞事件泵, + // 2026-10-05 16:18 实证:rx 等待卡死后全部后续消息黑洞) + tauri::async_runtime::spawn(async move { + handle_message(ctx, msg).await; + }); Ok(()) } }) .build(); tracing::info!("[feishu] 长连接启动(桌面端直连,免公网回调)"); - channel + let result = channel .start() .await - .map_err(|e: LarkError| anyhow::anyhow!("飞书长连接退出: {e}"))?; + .map_err(|e: LarkError| anyhow::anyhow!("飞书长连接退出: {e}")); + watchdog.abort(); + result?; Ok(()) } +/// D3 探活常量:事件断流判定阈值。聊天渠道正常空闲也可能长时间无消息,阈值取长防把 +/// 安静误判为假活 +const LIVENESS_STALE_SECS: i64 = 600; +/// D3 看门狗巡检间隔 +const LIVENESS_TICK: std::time::Duration = std::time::Duration::from_secs(60); + +/// D3 探活看门狗:巡检"事件断流"(距最后事件/连接建立超阈值且无在途会话),命中落盘告警。 +/// +/// 只告警不强杀:Channel/ChannelBuilder 未暴露连接句柄、探活或关闭 API(SDK 0.3.12 勘察), +/// 若强杀只能靠取消 channel.start()——SDK 的 ping 循环是独立 spawn 的任务且持有连接 +/// 写半部,取消使其脱离管理,连接健康时永续 ping 形成僵尸连接、占飞书服务端连接配额, +/// 比假活更危险。真死连接(TCP 断/对端静默)由 SDK 读空闲超时(2×ping 周期+5s)加内部 +/// 自动重连自愈;本看门狗覆盖的是残余盲区:协议层活着但事件管道异常断流,告警给出 +/// 排查起点(此前 bot 静默死无任何日志可查)。 +async fn run_liveness_watchdog(ctx: Arc) { + let mut warned = false; + loop { + tokio::time::sleep(LIVENESS_TICK).await; + let elapsed = now_secs() - ctx.last_alive.load(Ordering::SeqCst); + if elapsed <= LIVENESS_STALE_SECS { + warned = false; + continue; + } + // 有会话在途不告警(回复投递依赖当前连接);同一段断流只告警一次防刷屏 + if ctx.inflight.load(Ordering::SeqCst) > 0 || warned { + continue; + } + warned = true; + tracing::warn!("[feishu] 探活告警: {elapsed}s 无任何事件且无在途会话,疑似连接假活"); + feishu_log(&format!( + "探活告警: {elapsed}s 无任何事件且无在途会话(疑似假活,持续出现请重启应用重建连接)" + )); + } +} + +/// D3 在途会话计数 RAII:handle_message 从注册等待桥到投递完成的任意退出路径 +/// (spawn 失败早退/投递完成)都保证计数回落 +struct InflightGuard(Arc); + +impl Drop for InflightGuard { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::SeqCst); + } +} + +fn now_secs() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + /// 入站文本:群聊需 @bot;白名单校验;/new 重置;其余走 ai-chat 会话链路 async fn handle_message( ctx: Arc, msg: larksuite_oapi_sdk_rs::channel::NormalizedMessage, ) { + // D3:任何入站消息都是事件管道存活证据,先于一切过滤更新活性时间戳 + ctx.last_alive.store(now_secs(), Ordering::SeqCst); if msg.message_type != "text" { return; } @@ -129,10 +240,13 @@ async fn handle_message( return; } let text = msg.text.clone().unwrap_or_default().trim().to_string(); + // 飞书 @ 占位清洗(如 @_user_1):@ 语义已由 mentioned_bot 承载,占位串只污染文本 + let text = strip_mention_placeholders(&text); if text.is_empty() { return; } let chat_id = msg.chat_id.clone(); + let message_id = msg.message_id.clone(); let open_id = msg .sender .user_id @@ -140,10 +254,12 @@ async fn handle_message( .and_then(|u| u.open_id.clone()) .unwrap_or_default(); - // 白名单:空=放行;非空且不在内 → 提示并附自己的 open_id(首启自助回填,免翻日志) - if !ctx.whitelist.is_empty() && !ctx.whitelist.iter().any(|w| w == &open_id) { + // 白名单(D2 热加载):空=放行;非空且不在内 → 提示并附自己的 open_id(首启自助回填, + // 免翻日志)。每消息现读 KV,设置面板改白名单即生效免重启 + let whitelist = read_whitelist(&ctx).await; + if !whitelist.is_empty() && !whitelist.iter().any(|w| w == &open_id) { let hint = format!("⛔ 你不在白名单内。你的 open_id:`{open_id}`(发给管理员回填后重试)"); - send_text(&ctx.client, &chat_id, &hint).await; + reply_message(&ctx.client, &message_id, &chat_id, &hint).await; return; } @@ -153,20 +269,51 @@ async fn handle_message( let db = ctx.app.state::().db.clone(); let snapshot = ctx.chat_map.lock().await.clone(); save_chat_map(&db, &snapshot).await; - send_text(&ctx.client, &chat_id, "✅ 已重置会话").await; + reply_message(&ctx.client, &message_id, &chat_id, "✅ 已重置会话").await; return; } // chat_id → conv_id 映射(内存命中优先,未命中新建) + feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80))); + // 立即确认(交互丝滑):占位改用单行卡片并记下其 message_id,完成后原地更新为结果, + // 占位与结果合一不刷屏;占位 reply 失败则置空,完成路径回退现状(新发卡片) + let placeholder_id = + reply_markdown_card(&ctx.client, &message_id, "🤔 已接收,处理中…").await; + // per-chat 串行:同会话消息排队(防通知桥单槽被后条覆盖致前条串扰) + let chat_lock = { + let mut locks = ctx.chat_locks.lock().await; + locks.entry(chat_id.clone()).or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))).clone() + }; + let _chat_guard = chat_lock.lock().await; let conv_id = { let mut map = ctx.chat_map.lock().await; map.entry(chat_id.clone()).or_insert_with(new_id).clone() }; + feishu_log(&format!("spawn 会话 conv={conv_id}")); + // D3 在途计数:从此到投递完成期间连接假活判定须让位,RAII 保任意退出路径回落 + ctx.inflight.fetch_add(1, Ordering::SeqCst); + let _inflight_guard = InflightGuard(ctx.inflight.clone()); let state = ctx.app.state::(); let (tx, rx) = tokio::sync::oneshot::channel::(); 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::(); + let delta_rx = state.ai_event_bus.subscribe(); + let stream_task = tauri::async_runtime::spawn(run_stream_refresher( + ctx.clone(), + conv_id.clone(), + placeholder_id.clone(), + message_id.clone(), + chat_id.clone(), + delta_rx, + deliver_rx, + )); + match crate::commands::ai::commands::chat::spawn_agentic_session( &ctx.app, &state, @@ -189,18 +336,294 @@ async fn handle_message( Err(e) => { // 启动失败:通知桥条目清掉防死条目(编排节点同款) state.notify_agent_completion(&conv_id, String::new()); - send_text(&ctx.client, &chat_id, &format!("❌ 会话启动失败: {e}")).await; + // 关终版通道让流式任务退出(此路无任何增量产出,卡片未被流式写过), + // 待其退出后再走失败投递,保卡片写入单点次序 + drop(deliver_tx); + let _ = stream_task.await; + feishu_log(&format!("spawn 失败 conv={conv_id}: {e}")); + deliver_result( + &ctx.client, + &placeholder_id, + &message_id, + &chat_id, + "❌ 会话启动失败,请稍后重试", + ) + .await; return; } } - // 等最终回复(通知桥 oneshot;超时兜底防悬挂) - let reply = match tokio::time::timeout(std::time::Duration::from_secs(600), rx).await { + // 等最终回复(通知桥 oneshot;超时 30 分钟,任务仍在后台跑不丢失) + let reply = match tokio::time::timeout(std::time::Duration::from_secs(1800), rx).await { Ok(Ok(text)) if !text.trim().is_empty() => text, - Ok(_) => "（会话已结束,无文本回复）".to_string(), - Err(_) => "⏱ 回复超时,请重试或 /new 重置".to_string(), + // A 韧性兜底:通知桥带回空内容 → 回查 DB 该会话最后一条 assistant(事实源),有则照发 + _ => { + let db = ctx.app.state::().db.clone(); + let fallback = last_assistant_reply(&db, &conv_id).await; + feishu_log(&format!("conv={conv_id} 通知桥空/超时,DB回查={}", truncate(fallback.as_deref().unwrap_or("无"), 60))); + match fallback { + Some(text) => text, + None => "⚠️ 本轮没有生成文本回复(已记录排查)。任务若仍在进行可稍后 /new 重置或重发。".to_string(), + } + } }; - send_text(&ctx.client, &chat_id, &reply).await; + feishu_log(&format!("回复 chat={chat_id} len={}", reply.len())); + // 终版投递:交流式任务单点执行(内部复用 deliver_result 原回退链); + // 任务已死(通道已关或异常退出)则兜底直投,此时无并发写卡方,回退链行为不变; + // 若任务在投递后异常退出,重复一次终版 edit_card 内容相同,无副作用 + let delivered = match deliver_tx.send(reply.clone()) { + Ok(()) => matches!(stream_task.await, Ok(())), + Err(_) => false, + }; + if !delivered { + deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &reply).await; + } +} + +/// D2 白名单热加载:每消息现读 KV,设置面板改白名单即生效免重启;读取/解析失败沿用 +/// 上次缓存——拒收口径随缓存,不因 DB 抖动放大为误放行;成功时回写缓存供降级使用 +async fn read_whitelist(ctx: &Ctx) -> Vec { + let db = ctx.app.state::().db.clone(); + let repo = SettingsRepo::new(&db); + let raw = repo.get(KV_FEISHU_CONFIG).await.ok().flatten(); + match raw.and_then(|r| serde_json::from_str::(&norm(Some(r))).ok()) { + Some(cfg) => { + *ctx.whitelist.write().await = cfg.whitelist.clone(); + cfg.whitelist + } + None => { + tracing::warn!("[feishu] 白名单 KV 读取/解析失败,沿用上次缓存"); + ctx.whitelist.read().await.clone() + } + } +} + +/// 流式预览节流间隔:飞书卡片编辑接口有 QPS 约束,3s 在可读性与配额间取衡(约束下限 2s) +const STREAM_FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3); +/// 流式预览上限字符数:对齐终版 clip_for_card 口径;超限从头截断保尾部 +const STREAM_PREVIEW_MAX_CHARS: usize = 8000; + +/// 流式预览任务:消费 ai_event_bus 的 AiTextDelta,节流把累计文本 edit_card 进占位卡片; +/// 终版文本由 deliver_tx 单点传入,复用 deliver_result 投递后退出。 +/// +/// 竞态设计:占位卡片在本任务存活期只有本任务写,中间态 tick 与终版投递同任务串行, +/// 上一次 edit_card 的 HTTP 响应返回后才可能发下一次,飞书服务端所见次序即代码次序, +/// 终版不会被迟到的中间态覆盖,故无需 AbortHandle。中间态刷失败不新发卡片回退 +/// (不值得为预览刷屏),新发回退链仍归终版 deliver_result 独有,行为不受本任务影响。 +async fn run_stream_refresher( + ctx: Arc, + conv_id: String, + placeholder_id: Option, + message_id: String, + chat_id: String, + mut delta_rx: tokio::sync::broadcast::Receiver, + mut deliver_rx: tokio::sync::oneshot::Receiver, +) { + let mut acc = String::new(); + // 累计字符数 O(1) 维护,与上次已刷数比对判"有无新增量",避免无变化空刷烧配额 + let mut acc_chars = 0usize; + let mut painted_chars = 0usize; + 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; + } + // 发送端被丢弃(spawn 失败路径)时不投递,由调用方兜底 + return; + } + item = delta_rx.recv() => match item { + Ok(v) => { + // 双过滤:只要本 conv 的流式增量;type tag 即变体名(AiChatEvent 无 serde rename) + if v.get("type").and_then(|t| t.as_str()) != Some("AiTextDelta") { + continue; + } + if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) { + continue; + } + if let Some(delta) = v.get("delta").and_then(|d| d.as_str()) { + acc.push_str(delta); + acc_chars += delta.chars().count(); + } + } + // 预览允许丢帧:终版事实源在通知桥 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; + } + }, + _ = ticker.tick() => { + // 占位缺失(首张卡片 reply 失败)时无从原地刷,保持静默等终版走新发回退 + let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) else { + continue; + }; + if acc_chars == painted_chars { + continue; + } + if update_card(&ctx.client, pid, &stream_preview(&acc)).await { + painted_chars = acc_chars; + } + // 刷失败静默跳过,下个 tick 自然重试 + } + } + } +} + +/// 流式中间态文本:尾部加光标符示意仍在生成;超上限从头截断保尾部,前缀标注省略事实 +/// 防读者误把中间态当全文 +fn stream_preview(acc: &str) -> String { + let total = acc.chars().count(); + let body = if total > STREAM_PREVIEW_MAX_CHARS { + let tail: String = acc.chars().skip(total - STREAM_PREVIEW_MAX_CHARS).collect(); + format!("…(前文已略)\n{tail}") + } else { + acc.to_string() + }; + format!("{body}\n▌") +} + +/// markdown 回复:interactive 卡片承载(lark_md 原生渲染标题/列表/粗体,text 类型会符号裸露) +async fn send_markdown_card(client: &LarkClient, message_id: &str, chat_id: &str, text: &str) { + 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; + } + }; + let content = serde_json::to_string(document.card()).unwrap_or_default(); + let input = SendInput { + receive_id: Some(chat_id.to_string()), + card: Some(content), + ..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 { + tracing::error!("[feishu] 卡片发送失败: {e}"); + // 卡片彻底失败仍回退纯文本,保可达 + reply_message(client, message_id, chat_id, text).await; + } + } +} + +/// reply 一张 markdown 卡片,成功返回卡片消息 id(完成后原地更新用) +/// (SDK strict reply 内部已校验业务码,Ok 即业务成功;id 为空视为失败) +async fn reply_markdown_card(client: &LarkClient, message_id: &str, text: &str) -> Option { + 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}"); + return None; + } + }; + let content = serde_json::to_string(document.card()).unwrap_or_default(); + // strict reply 只认 message_id 参数,携带 receive_id 等接收方字段会被 SDK 拒绝 + let input = SendInput { + card: Some(content), + ..Default::default() + }; + match client + .channel_messaging() + .reply(message_id, &input, &RequestOption::default()) + .await + { + Ok(sent) if !sent.message_id.is_empty() => Some(sent.message_id), + Ok(_) => None, + Err(e) => { + tracing::warn!("[feishu] 占位卡片回复失败: {e}"); + None + } + } +} + +/// 原地更新卡片(占位→结果,不刷屏)。edit_card 返回 EmptyResp:HTTP 200 但业务 code≠0 +/// (卡片被撤回/JSON 超限等)时 err 为 nil 伪成功,必须再校验 success +async fn update_card(client: &LarkClient, message_id: &str, text: &str) -> bool { + let card = match CardDocument::new(Card::new().element(Element::Markdown(MdElement::new( + text.to_string(), + )))) { + Ok(d) => d, + Err(e) => { + tracing::warn!("[feishu] 结果卡片构建失败: {e}"); + return false; + } + }; + match client + .channel_messaging() + .edit_card(message_id, &card, &RequestOption::default()) + .await + { + Ok(resp) if resp.success() => true, + Ok(resp) => { + tracing::warn!("[feishu] 卡片更新业务失败 code={:?}", resp.code_error); + false + } + Err(e) => { + tracing::warn!("[feishu] 卡片更新失败: {e}"); + false + } + } +} + +/// 结果投递统一入口:先做表格降级(md 表格在卡片 markdown 组件渲染不稳), +/// 优先原地更新占位卡片(占位与结果合一);占位缺失或更新失败回退新发卡片(内含纯文本兜底) +async fn deliver_result( + client: &LarkClient, + placeholder_id: &Option, + message_id: &str, + chat_id: &str, + text: &str, +) { + let text = clip_for_card(&downgrade_md_tables(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; +} + +/// 卡片正文长度防御:飞书卡片 JSON 约 30KB 上限,超长会让原地更新与新发双双失败, +/// 一路穿透到纯文本兜底仍不可读;入口统一裁剪,尾注指回桌面端(u-claw archiveIfClipped 同口径) +fn clip_for_card(text: &str, max_chars: usize) -> String { + if text.chars().count() <= max_chars { + return text.to_string(); + } + let clipped: String = text.chars().take(max_chars).collect(); + format!("{clipped}\n\n---\n（内容过长已截断，完整内容见桌面端会话）") +} + +/// 回复到原消息(reply 落回消息所在话题/上下文,修复回复跑到群主界面的问题);失败回退 chat 直发 +async fn reply_message(client: &LarkClient, message_id: &str, chat_id: &str, text: &str) { + let opt = &RequestOption::default(); + if !message_id.is_empty() { + if let Err(e) = client.channel_messaging().reply_text(message_id, text, false, opt).await { + tracing::warn!("[feishu] reply 失败,回退 chat 直发: {e}"); + } else { + return; + } + } + send_text(client, chat_id, text).await; } async fn send_text(client: &LarkClient, chat_id: &str, text: &str) { @@ -214,8 +637,149 @@ async fn send_text(client: &LarkClient, chat_id: &str, text: &str) { } } +/// 剥离飞书 @ 占位串(如 @_user_1),@ 语义已由 mentioned_bot 承载。 +/// 占位符固定形态 `@_user_` + 连续数字:只剥此形态(顺带吞其后单个空格防双空格); +/// 前缀后非数字(如 @_user_x)按字面保留。不能「找空白跳」——中文无空格 +/// (@_user_1帮我查)会整段吞掉致消息被静默丢弃(review 2026-10-05 ①)。 +fn strip_mention_placeholders(text: &str) -> String { + const MARK: &str = "@_user_"; + let mut out = String::with_capacity(text.len()); + let mut rest = text; + while let Some(i) = rest.find(MARK) { + let after = &rest[i + MARK.len()..]; + let digit_len: usize = after + .chars() + .take_while(|c| c.is_ascii_digit()) + .map(char::len_utf8) + .sum(); + if digit_len == 0 { + // 非占位形态:MARK 字面保留,rest 从 MARK 之后继续扫描(严格前进防死循环) + out.push_str(&rest[..i + MARK.len()]); + } else { + out.push_str(&rest[..i]); + rest = &after[digit_len..]; + if let Some(stripped) = rest.strip_prefix(' ') { + rest = stripped; + } + } + rest = if digit_len == 0 { after } else { rest }; + } + out.push_str(rest); + out.trim().to_string() +} + +/// md 表格降级:表格块(连续 | 开头行且第二行为 |---| 分隔形态)转「- 列1：列2」列表行, +/// 单元格内既有行内格式(粗体等)原样保留,表格前后非表格文本不动。 +/// 表头也转列表且单元格加粗:列名信息不丢,且避免 | 行残留导致竖线裸露 +fn downgrade_md_tables(text: &str) -> String { + let lines: Vec<&str> = text.lines().collect(); + let mut out: Vec = Vec::with_capacity(lines.len()); + let mut i = 0; + while i < lines.len() { + if is_table_row(lines[i]) && i + 1 < lines.len() && is_table_separator(lines[i + 1]) { + out.push(table_line_to_list(lines[i], true)); + i += 2; // 表头已转,分隔行只是语法标记,丢弃 + while i < lines.len() && is_table_row(lines[i]) { + out.push(table_line_to_list(lines[i], false)); + i += 1; + } + continue; + } + out.push(lines[i].to_string()); + i += 1; + } + out.join("\n") +} + +fn is_table_row(line: &str) -> bool { + line.trim_start().starts_with('|') +} + +/// 分隔行形态:| 开头,且整行只由 | - : 与空白构成并含至少一段 --- +fn is_table_separator(line: &str) -> bool { + let t = line.trim(); + t.starts_with('|') && t.contains("---") && t.chars().all(|c| matches!(c, '|' | '-' | ':' | ' ')) +} + +/// 单行表格转列表行;bold=true 用于表头行,加粗区隔列名 +fn table_line_to_list(line: &str, bold: bool) -> String { + // 单元格内 \| 是竖线字面量,先占位再还原,防被当列分隔切开 + const ESC: char = '\u{0}'; + let cells: Vec = line + .trim() + .trim_matches('|') + .replace("\\|", &ESC.to_string()) + .split('|') + .map(|c| c.trim().replace(ESC, "\\|")) + .filter(|c| !c.is_empty()) + .map(|c| { + if bold { + format!("**{c}**") + } else { + c + } + }) + .collect(); + format!("- {}", cells.join("：")) +} + +/// 关键事件日志落盘(零依赖 fs append;Roaming 下 feishu-channel.log,排查空回复等复现凭据) +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"); + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(p) { + let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()).unwrap_or(0); + let _ = writeln!(f, "[{ts}] {line}"); + } + } +} + +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 { + 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()) + .map(|m| m.content) +} + async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap) { let repo = SettingsRepo::new(db); let json = serde_json::to_string(map).unwrap_or_default(); let _ = repo.set(KV_CHAT_MAP, &json).await; } + +#[cfg(test)] +mod tests { + use super::strip_mention_placeholders; + + #[test] + fn mention_cn_no_space_not_swallowed() { + // review ①:中文紧邻占位(无空白)不能整段吞——此前 skip 找空白,skip=len 吞全文 + assert_eq!(strip_mention_placeholders("@_user_1帮我查任务"), "帮我查任务"); + } + + #[test] + fn mention_prefix_and_digits_only_stripped() { + // 只剥 @_user_ + 数字,数字后内容保留(含标点/英文) + assert_eq!(strip_mention_placeholders("@_user_12 hi"), "hi"); + assert_eq!(strip_mention_placeholders("@_user_1,今天天气"), ",今天天气"); + // 非数字紧跟(非官方占位形态)不误剥前缀 + assert_eq!(strip_mention_placeholders("@_user_x 保留"), "@_user_x 保留"); + } + + #[test] + fn mention_multiple_and_plain_text_untouched() { + assert_eq!(strip_mention_placeholders("a @_user_1 b @_user_2 c"), "a b c"); + assert_eq!(strip_mention_placeholders("普通消息不动"), "普通消息不动"); + assert_eq!(strip_mention_placeholders("@_user_1"), ""); + } +}