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

优化: 飞书渠道文本链路与资源释放

Browse Source
This commit is contained in:
lxy
2026-10-05 23:41:20 +08:00
parent cb260aba4b
commit 0f3c3ce1aa
2 changed files with 607 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-im-gateway/src/feishu.rs
+17 -8
View File
@@ -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<tokio::sync::Mutex<HashMap<String, String>>>,
}
impl FeishuAdapter {
pub fn new(cfg: &FeishuConfig) -> anyhow::Result<Self> {
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<InboundEvent>) -> 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();
src-tauri/src/feishu_channel.rs
+588 -24
View File
@@ -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>) -> String {
struct Ctx {
app: tauri::AppHandle,
client: LarkClient,
whitelist: Vec<String>,
/// 白名单缓存(D2 热加载):每消息现读 KV 刷新,读取失败沿用本缓存,改配置免重启
whitelist: Arc<tokio::sync::RwLock<Vec<String>>>,
/// chat_id → conversation_id(启动时从 KV 恢复,变更回写)
chat_map: Arc<tokio::sync::Mutex<HashMap<String, String>>>,
/// per-chat 串行锁:同会话消息排队处理,防通知桥单槽跨消息串扰(P1-1)
chat_locks: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
/// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流
last_alive: Arc<AtomicI64>,
/// D3 在途会话数:spawn 到投递完成期间 >0,看门狗要求为 0 才告警,防有回复待投递时误报
inflight: Arc<AtomicUsize>,
}
/// 启动入口:lib.rs setup 调用。未启用/配置缺失/构造失败一律记日志返回,不阻塞应用启动。
@@ -58,22 +67,38 @@ pub fn start_if_enabled(app: tauri::AppHandle) {
let db = app.state::<crate::state::AppState>().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::<FeishuConfig>(&raw) else {
tracing::error!("[feishu] feishu-config 解析失败,渠道不启动");
let cfg = match serde_json::from_str::<FeishuConfig>(&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<Ctx>, client: LarkClient) -> anyhow::Result<()> {
async fn run_loop(ctx: Arc<Ctx>, 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 {
// 异步处理:回调内绝不等 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<Ctx>) {
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<AtomicUsize>);
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<Ctx>,
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::<crate::state::AppState>().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::<crate::state::AppState>();
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
state.register_agent_completion(conv_id.clone(), tx);
// 流式预览:订阅既有 ai_event_bus 拿本 conv 的 AiTextDelta 增量(emit_ai_chat_event
// 双写总线已接入,渠道侧纯消费,零改动接线)。spawn 前订阅防早期增量丢失;
// 终版文本经 deliver_tx 也交给本任务单点投递,占位卡片自 spawn 起只被这一个任务写,
// 中间态与终版写卡天然串行,无迟到中间态覆盖终版的竞态
let (deliver_tx, deliver_rx) = tokio::sync::oneshot::channel::<String>();
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::<crate::state::AppState>().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<String> {
let db = ctx.app.state::<crate::state::AppState>().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::<FeishuConfig>(&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<Ctx>,
conv_id: String,
placeholder_id: Option<String>,
message_id: String,
chat_id: String,
mut delta_rx: tokio::sync::broadcast::Receiver<serde_json::Value>,
mut deliver_rx: tokio::sync::oneshot::Receiver<String>,
) {
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<String> {
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<String>,
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<String> = 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<String> = 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<String> {
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<String, String>) {
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"), "");
}
}