Files
DevFlow/src-tauri/src/feishu_channel.rs
T
lxy 795935bf6a 新增: 飞书文件消息入站与补充注入链完善
- 文件消息入站:20MB 上限落盘引用,模型经 read_file 按需读盘;下载带 3 次退避+120s 超时防线,复用媒体域 hash 去重/OSS 双写
- 群聊门控对齐 u-claw 模式:前移至类型分流前,@自己/@所有人响应,未艾特静默旁观(原门控后置致未支持类型误回提示卡)
- 修复补充注入丢失:pending_injects 未消化计数+收敛出口续轮+桌面注入 IPC+前端本地气泡
- df-ai-core 测试构造器补 source 字段(溯源字段遗漏,全量门禁暴露)
2026-10-15 14:27:22 +08:00

2862 lines
137 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 飞书直连渠道:桌面端出站 WS 长连接(免公网回调),消息进 ai-chat 会话链路
//!
//! 形态依据:docs/02-架构设计/专项设计/飞书直连接入-形态修正与问题记录-2026-10-05.md
//! (用户 10-05 推翻网关链路,定稿桌面本地直连;参照 u-claw feishu_channel.go 单进程闭环)
//! 内核自 crates/df-im-gateway/src/feishu.rs 提炼(用户令:并入现有,不新建 crate);
//! 会话复用 spawn_agentic_session 通用核心(pin_active=false,与编排节点同款调用方),
//! 完成通知走 AppState 通知桥(register/notify_agent_completion,编排节点先例)。
//!
//! 第一批:文本收发(整段回复);流式预览已接入(订阅 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::{
Action, ActionComponent, Button, ButtonType, Card, CardDocument, Div, Element, Header, Markdown as MdElement,
};
use larksuite_oapi_sdk_rs::channel::{Channel, ChannelPolicy, SendInput};
use larksuite_oapi_sdk_rs::{EventDispatcher, LarkClient, LarkError, RequestOption};
use serde::Deserialize;
use df_storage::crud::SettingsRepo;
use df_types::types::new_id;
use df_ai::provider::ContentPart;
/// app_settings 键:飞书渠道配置(与设置面板 FeishuPanel 同键契约)
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();
/// 渠道客户端只读访问(37 监控告警等外部推送方用):None=渠道未启动
pub(crate) fn active_client() -> Option<&'static LarkClient> {
ACTIVE_CLIENT.get().copied()
}
/// 34 滚动汇总卡句柄:chat_id 对 (在途事务卡 message_id, 已累计正文)。同 chat 多步
/// 进度推送原地追加为一卡,防逐条直推刷屏;内存句柄重启即失(重启后首推另起新卡)。
/// std Mutex 只做句柄与字符串存取,不跨 await 持有,短锁语义
static ROLLING_CARDS: std::sync::OnceLock<std::sync::Mutex<HashMap<String, (String, String)>>> =
std::sync::OnceLock::new();
/// 24 取消收口旁路:conv_id 对完成订阅任务的取消发送端。引擎停止路径(ai_stop_loop 置
/// stop_flag 后 loop 于轮边界收尾)只 emit AiCompleted,watcher 消费不到;/cancel 后须旁路
/// 唤醒 watcher 投「已停止」终版,防空等至双超时弃守期间反复重刷过期提示。oneshot 消费
/// 一次即摘除,watcher 任意路径退出亦摘除,陈旧条目最多导致一次空发,无泄漏
static CANCEL_WATCHERS: std::sync::OnceLock<
tokio::sync::Mutex<HashMap<String, tokio::sync::oneshot::Sender<()>>>,
> = std::sync::OnceLock::new();
fn cancel_watchers(
) -> &'static tokio::sync::Mutex<HashMap<String, tokio::sync::oneshot::Sender<()>>> {
CANCEL_WATCHERS.get_or_init(|| tokio::sync::Mutex::new(HashMap::new()))
}
#[derive(Debug, Clone, Deserialize)]
struct FeishuConfig {
#[serde(default)]
enabled: bool,
#[serde(default)]
app_id: String,
#[serde(default)]
app_secret: String,
/// 空 = 放行所有(首启抓 open_id 回填);open_id 按应用隔离,严禁跨应用搬运。
/// 兼容历史字符串形态(逗号分隔):前端旧版曾存字符串,解析失败会静默沿用旧缓存
/// 致白名单改动不生效,故此处 deserialize_with 双形态兼容(数组优先,字符串拆分)
#[serde(default, deserialize_with = "de_whitelist")]
whitelist: Vec<String>,
}
/// 白名单反序列化双形态兼容:数组(正典)或逗号分隔字符串(历史脏数据)。
/// 非法形态(数字/对象等)→ 空 Vec(等价放行所有,不 panic 不报错)
fn de_whitelist<'de, D>(deserializer: D) -> Result<Vec<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
use serde::Deserialize;
let v = Option::<serde_json::Value>::deserialize(deserializer)?;
Ok(match v {
Some(serde_json::Value::Array(arr)) => arr
.into_iter()
.filter_map(|x| x.as_str().map(|s| s.trim().to_string()))
.filter(|s| !s.is_empty())
.collect(),
Some(serde_json::Value::String(s)) => s
.split(',')
.map(|x| x.trim().to_string())
.filter(|x| !x.is_empty())
.collect(),
_ => Vec::new(),
})
}
/// settings 值归一:前端 JSON.stringify 落库带引号,去首尾引号(SettingsNorm 正典同款)
fn norm(raw: Option<String>) -> String {
raw.map(|s| s.trim().trim_matches('"').to_string())
.unwrap_or_default()
}
/// 渠道运行上下文(回调闭包共享)
struct Ctx {
app: tauri::AppHandle,
client: LarkClient,
/// 白名单缓存(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<()>>>>>,
/// 防抖聚合窗:chat_id → 窗内追加消息蓄积。首条登记,追加并入即返回,
/// 窗满首条取走统一受理;生命周期同 chat_locks(渠道重启随 Ctx 重建)
debounce: Arc<tokio::sync::Mutex<HashMap<String, DebounceEntry>>>,
/// D3 活性时间戳(unix 秒):入站事件与连接建立成功时更新,看门狗据此判事件断流
last_alive: Arc<AtomicI64>,
/// D3 在途会话数:消息受理(handle_message 白名单通过)到任务投递完成期间 >0,
/// 看门狗要求为 0 才告警,防有回复待投递时误报
inflight: Arc<AtomicUsize>,
}
/// 启动入口:lib.rs setup 调用。未启用/配置缺失/构造失败一律记日志返回,不阻塞应用启动。
pub fn start_if_enabled(app: tauri::AppHandle) {
tauri::async_runtime::spawn(async move {
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 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() {
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;
}
};
// 启动时从 KV 恢复 chat 映射
let restored: HashMap<String, String> = serde_json::from_str(
&norm(repo.get(KV_CHAT_MAP).await.ok().flatten()),
)
.unwrap_or_default();
let ctx = Arc::new(Ctx {
app,
client: client.clone(),
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())),
debounce: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
last_alive: Arc::new(AtomicI64::new(now_secs())),
inflight: Arc::new(AtomicUsize::new(0)),
});
// D1 僵尸资源修复:client 的 'static 化只做一次(原在 run_loop 内,每次重连都
// Box::leak 一份 LarkClient,频繁断线场景内存阶梯上涨);Channel::builder 仅借用
// 引用,重连循环内复用同一 leaked 引用,只重建 Channel/dispatcher
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 静默死
// 2026-10-08 重连用户通知:此前重连只落日志,渠道静默死数分钟用户无感知。
// 判据:run_loop 运行时长——>30s 视为「曾成功连接过」(连接建立后才有长驻),
// 其后的退出即真断线,推「重新连接」通知;连续失败≥3 推告警。通知目标取
// df-monitor-chat-id(运维通道,与监控告警同键;未配则仅落日志)
let mut backoff = 2u64;
let mut consecutive_failures = 0u32;
loop {
let started = tokio::time::Instant::now();
let run_result = run_loop(ctx.clone(), leaked).await;
let ran_secs = started.elapsed().as_secs();
let was_connected = ran_secs >= 30;
if was_connected {
// 曾成功连接 → 本次退出是真断线,重置失败计数并推重连通知
consecutive_failures = 0;
let reason = match &run_result {
Err(e) => format!("{e}"),
Ok(_) => "连接正常关闭".to_string(),
};
tracing::warn!("[feishu] 长连接断开(运行 {ran_secs}s),{}s 后重连: {reason}", backoff);
feishu_log(&format!("长连接断开(运行 {ran_secs}s),{}s 后重连: {reason}", backoff));
notify_monitor(&ctx, &format!("⚠️ 飞书渠道连接断开,{backoff}s 后自动重连中…")).await;
} else {
consecutive_failures += 1;
let reason = match &run_result {
Err(e) => format!("{e}"),
Ok(_) => "连接立即关闭".to_string(),
};
tracing::error!("[feishu] 长连接快速退出(运行 {ran_secs}s,连续失败 {consecutive_failures}),{}s 后重连: {reason}", backoff);
feishu_log(&format!("长连接快速退出(运行 {ran_secs}s,连续失败 {consecutive_failures}),{}s 后重连: {reason}", backoff));
if consecutive_failures == 3 {
notify_monitor(&ctx, "⚠️ 飞书渠道连接不稳定(连续 3 次快速断开),后台持续重连中,请检查网络或凭证配置。").await;
}
}
tokio::time::sleep(std::time::Duration::from_secs(backoff)).await;
backoff = (backoff * 2).min(300);
}
});
}
/// 运维通知:推给 df-monitor-chat-id 配置的飞书群(与监控告警同键)。未配/发送失败
/// 仅落日志,不阻塞重连循环。重连通知是锦上添花,失败静默
async fn notify_monitor(ctx: &Ctx, text: &str) {
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let repo = SettingsRepo::new(&db);
let chat_id = norm(repo.get("df-monitor-chat-id").await.ok().flatten());
if chat_id.is_empty() {
feishu_log(&format!("[notify] 未配置 df-monitor-chat-id,仅落日志: {text}"));
return;
}
let _ = send_markdown_card(&ctx.client, "", &chat_id, text).await;
}
async fn run_loop(ctx: Arc<Ctx>, client: &'static LarkClient) -> anyhow::Result<()> {
let dispatcher = EventDispatcher::new("", "");
let ctx_msg = ctx.clone();
let ctx_ready = ctx.clone();
let ctx_card = 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(())
}
})
.on_card_action(move |action| {
let ctx = ctx_card.clone();
async move {
tauri::async_runtime::spawn(async move {
handle_card_action(ctx, action).await;
});
Ok(())
}
})
.build();
tracing::info!("[feishu] 长连接启动(桌面端直连,免公网回调)");
let result = channel
.start()
.await
.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)
}
/// 命令解析前归一(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))
}
/// 图片消息喂模型的任务语境(#958):飞书 image 为独立消息类型无法带文字,
/// 用引导词让 vision 模型知道「这是用户发来的图,按图作答」
const IMAGE_INBOUND_PROMPT: &str = "(用户通过飞书发来一张图片,请查看图片并按图片内容处理)";
/// 图片入站大小上限(#958):base64 转换膨胀 4/3,vision 端点普遍接受 ≤20MB 内嵌图,
/// 10MB 原图留足余量;超限引导用户压缩,防超大图烧 token 与转码内存
const MAX_IMAGE_BYTES: usize = 10 * 1024 * 1024;
/// 文件入站大小上限:文件不经 base64 内嵌(模型经 read_file 按需读盘),上限只防
/// 恶意超大件烧下载流量与磁盘,20MB 覆盖常规文档/报表/压缩包
const MAX_FILE_BYTES: usize = 20 * 1024 * 1024;
/// 飞书消息资源下载带防线(图片/文件共用,对齐 u-claw fetchResource 经验):3 次指数
/// 退避(1s/4s)+单次 120s 超时——飞书资源接口偶发抖动,无重试时用户收到一次性失败;
/// 超时防接口挂起拖死受理链。空响应与错误同待遇进重试(可能为服务端瞬时异常),
/// 耗尽后报最后一次原因。resource_type=image/file,同一资源接口
async fn download_message_resource_with_retry(
client: &LarkClient,
message_id: &str,
file_key: &str,
resource_type: &str,
label: &str,
) -> Result<Vec<u8>, String> {
let mut last_err = String::from("下载失败: 未知错误");
for attempt in 0u32..3 {
if attempt > 0 {
tokio::time::sleep(std::time::Duration::from_secs(1 << (attempt - 1))).await;
}
let dl = tokio::time::timeout(
std::time::Duration::from_secs(120),
client.channel_messaging().download_message_resource(
message_id,
file_key,
resource_type,
&RequestOption::default(),
),
)
.await;
match dl {
Ok(Ok(r)) if !r.data.is_empty() => return Ok(r.data),
Ok(Ok(_)) => last_err = format!("{label}内容为空"),
Ok(Err(e)) => last_err = format!("下载失败: {e}"),
Err(_) => last_err = "下载超时(120s)".into(),
}
}
Err(last_err)
}
/// 下载飞书图片消息资源并落盘,返回本地引用 ContentPart(#958 引用化存储)。
/// 存储面只存 `file://` 引用(base64 不进 SQLite,防大图随历史全量重放膨胀),
/// 发送请求时由 df-ai::image_ref 单点按需读盘转 base64(模型使用按需转换)。
/// 走 channel_messaging 便捷方法(与 reply/send 同风格),租户 token 由 client 自管。
/// media_type 由字节魔数判定(下载响应无可靠 MIME;飞书图床为 png/jpeg/webp/gif),
/// 未知格式拒收——错误 media_type 会被 vision 端点 400,宁早拒不晚炸。
async fn download_feishu_image(
client: &LarkClient,
store: &crate::media_store::MediaStore,
message_id: &str,
file_key: &str,
) -> Result<ContentPart, String> {
let data = download_message_resource_with_retry(client, message_id, file_key, "image", "图片")
.await?;
if data.len() > MAX_IMAGE_BYTES {
return Err(format!(
"图片超过 10MB 上限(约 {}MB),请压缩后重发",
data.len() / 1024 / 1024
));
}
let media_type = sniff_image_mime(&data)
.ok_or_else(|| "不支持的图片格式(仅支持 png/jpeg/webp/gif)".to_string())?;
// 存储决策收口 media_store:hash 去重落盘,OSS 配置在则双写并引用 https(跨设备可解析),
// 失败降级本机 file://(不阻塞)。本函数只管飞书协议(下载+防线),不管存储。
// 飞书 image 消息无原始文件名(None=纯 hash 名)
let url = store.save(&data, media_type, None).await?;
Ok(ContentPart::Image {
url: Some(url),
base64: None,
media_type: Some(media_type.into()),
alt: None,
})
}
/// 文件消息喂模型的任务语境:告知文件名与本机落盘路径,引导用 read_file 按需读取;
/// 二进制/超大文件的读取局限如实交底,防模型谎称已完整处理。落盘前(无路径)给
/// 占位版,仅供空判与命令/审批判定,不喂模型不落卡
fn file_inbound_prompt(file_name: &str, path: Option<&str>) -> String {
match path {
Some(p) => format!(
"(用户通过飞书发来文件「{file_name}」,已保存到本地路径 {p};请按用户需求用 read_file 工具读取该文件处理。二进制/超大文件可能无法完整读取,如实告知用户)"
),
None => format!("(用户通过飞书发来文件「{file_name}」,保存中)"),
}
}
/// 文件超限判定:超 20MB 上限时给可读错误话术(提示压缩/拆分),宁拒不落盘
fn file_size_over_limit_error(len: usize) -> Option<String> {
(len > MAX_FILE_BYTES).then(|| {
format!(
"文件超过 20MB 上限(约 {}MB),请压缩或拆分后重发",
len / 1024 / 1024
)
})
}
/// 文件展示名裁决:SDK 归一化 file_name 优先(保真,清洗首尾空白);缺失回退
/// `file_<内容 hash 前 8>`(与落盘 hash 命名同源,可对账辨认)
fn file_display_name(raw: &str, bytes: &[u8]) -> String {
let raw = raw.trim();
if raw.is_empty() {
format!("file_{}", &crate::media_store::content_hash(bytes)[..8])
} else {
raw.to_string()
}
}
/// 下载飞书文件消息资源并落盘,返回 (展示名, 本机绝对路径)。
/// 与图片同一资源接口(resource_type=file)与下载防线;落盘走 media_store 通用入口
/// (hash 去重+OSS 双写+失败降级,原文件名缀后保目录可读)。返回本机路径喂模型
/// read_file(OSS https 引用对本机读取无意义,不作返回);超 20MB 拒收回可读错误
async fn download_feishu_file(
client: &LarkClient,
store: &crate::media_store::MediaStore,
message_id: &str,
res: &larksuite_oapi_sdk_rs::channel::ChannelResource,
) -> Result<(String, String), String> {
let data =
download_message_resource_with_retry(client, message_id, &res.file_key, "file", "文件")
.await?;
if let Some(err) = file_size_over_limit_error(data.len()) {
return Err(err);
}
let display_name = file_display_name(&res.file_name, &data);
let original = res.file_name.trim();
let (_, path) = store
.save_with_path(
&data,
"application/octet-stream",
(!original.is_empty()).then_some(original),
)
.await?;
Ok((display_name, path.to_string_lossy().into_owned()))
}
/// 读取媒体 OSS 配置(KV `media-oss-config`;未配置/解析失败 → None=纯本地模式)
async fn read_media_oss_config(
db: &df_storage::db::Database,
) -> Option<crate::media_store::OssConfig> {
crate::media_store::read_oss_config(db).await
}
/// 图片格式魔数判定(#958):PNG/JPEG/GIF/WEBP 四态,与飞书图床常见格式对齐
fn sniff_image_mime(b: &[u8]) -> Option<&'static str> {
if b.starts_with(&[0x89, b'P', b'N', b'G']) {
Some("image/png")
} else if b.starts_with(&[0xFF, 0xD8, 0xFF]) {
Some("image/jpeg")
} else if b.starts_with(b"GIF8") {
Some("image/gif")
} else if b.len() > 12 && b.starts_with(b"RIFF") && &b[8..12] == b"WEBP" {
Some("image/webp")
} else {
None
}
}
/// 防抖窗口(拍板 0.5s):同 chat 连发消息的合并等待时长
const DEBOUNCE_WINDOW: std::time::Duration = std::time::Duration::from_millis(500);
/// 防抖窗内一条 chat 的追加消息蓄积。首条原文/图片不入此(留在本地,窗满时合并),
/// 结构上保证窗满取走必得 entry 且最坏退化为单条,消息不丢
#[derive(Default)]
struct DebounceEntry {
/// 追加消息合并文本(空行拼接)
text: String,
/// 追加消息图片片(到达序合并)
parts: Option<Vec<ContentPart>>,
/// 已追加条数(不含首条)
appended: usize,
}
impl DebounceEntry {
/// 追加一条进窗:文本空行拼接、图片片到达序合并、计数累加
fn append(&mut self, text: &str, parts: Option<Vec<ContentPart>>) {
self.text = merge_debounce_text(&self.text, text);
self.parts = merge_debounce_parts(self.parts.take(), parts);
self.appended += 1;
}
}
/// 防抖合并纯函数:文本空行拼接(任一侧为空直取另一侧,不产首尾空行)
fn merge_debounce_text(base: &str, next: &str) -> String {
if base.is_empty() {
next.to_string()
} else if next.is_empty() {
base.to_string()
} else {
format!("{base}\n\n{next}")
}
}
/// 防抖合并纯函数:图片片按到达序拼接(单侧 None 直取另一侧)
fn merge_debounce_parts(
base: Option<Vec<ContentPart>>,
next: Option<Vec<ContentPart>>,
) -> Option<Vec<ContentPart>> {
match (base, next) {
(Some(mut a), Some(b)) => {
a.extend(b);
Some(a)
}
(a, b) => a.or(b),
}
}
/// 窗满合并结果(首条视角):合并后任务载荷+占位卡句柄,交回 handle_message 续原流程
struct MergedIncoming {
text: String,
parts: Option<Vec<ContentPart>>,
placeholder_id: Option<String>,
}
/// 连发消息防抖聚合(u-claw 同款)。防抖点在注入旁路之后:命令/审批/注入即时路径不进窗,
/// 白名单与图片下载已在前逐条完成。追加消息并入 entry 即返回 None(调用方直接收尾,
/// 不回卡防刷屏——首条占位卡承载合并提示,不取 per-chat 锁不 spawn,受理计数随 RAII
/// 回落);首条登记空 entry 后回占位卡(文案不变)等窗,窗满取走 entry 返回合并结果,
/// 有追加时占位卡原地 PATCH 合并提示。sleep 全程不持任何锁;取走与追加的互斥由
/// DebounceMap 锁保证,窗满后到达的消息查 None 自成新窗,自洽
async fn debounce_collect(
ctx: &Ctx,
chat_id: &str,
message_id: &str,
text: String,
parts: Option<Vec<ContentPart>>,
) -> Option<MergedIncoming> {
{
let mut map = ctx.debounce.lock().await;
if let Some(entry) = map.get_mut(chat_id) {
entry.append(&text, parts);
return None;
}
map.insert(chat_id.to_string(), DebounceEntry::default());
}
// 立即确认(交互丝滑):占位单行卡片原地更新为结果,占位与结果合一不刷屏;
// reply 失败置空,完成路径回退新发卡片。带任务原文回显,让用户在到受理卡的
// 无内容可对证期即时确认消息被正确接收;摘要先取(text 随后参与合并 move)
let placeholder_text = format!("🤔 已接收,处理中…\n任务:{}", truncate(&text, 60));
let placeholder_id =
reply_markdown_card(&ctx.client, message_id, &placeholder_text).await;
tokio::time::sleep(DEBOUNCE_WINDOW).await;
// 取走窗内蓄积。entry 仅本首条插入、追加只改不摘,None 理论不可达,兜底退化为单条
let entry = ctx
.debounce
.lock()
.await
.remove(chat_id)
.unwrap_or_default();
if entry.appended > 0 {
deliver_result(
&ctx.client,
&placeholder_id,
message_id,
chat_id,
&format!("📥 已合并 {} 条消息,开始处理", entry.appended + 1),
)
.await;
}
Some(MergedIncoming {
text: merge_debounce_text(&text, &entry.text),
parts: merge_debounce_parts(parts, entry.parts),
placeholder_id,
})
}
/// 入站文本:群聊需 @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);
// 群聊响应门控对齐 u-claw 模式(feishu_events.go mentionedBot 四条):①@自己响应
// ②@所有人响应 ③@别的bot不响应 ④无mention静默旁观;私聊恒响应。SDK 的
// mentioned_bot/mention_all 已承载①②③的精确判定。门控必须在类型分流之前——
// 分流段对语音等未支持类型会回提示卡,门控放后面时未艾特的群消息也会得到回复
// (2026-10-15 用户实测「群里没艾特也处理」根因)。
if msg.chat_type == "group" && !msg.mentioned_bot && !msg.mention_all {
return;
}
// 类型分流(#958 图片入站,文件入站同源扩展):text 原样;image 走下载装配(vision
// ContentPart);file 走下载落盘(read_file 本机路径);语音等其余类型维持提示卡。
// image/file 消息 SDK 归一化已把 file_key/file_name 放入 msg.resources(ChannelResource),
// 无需自行解析 content JSON。
let (image_key, file_res) = if msg.message_type == "image" {
match msg.resources.iter().find(|r| r.resource_type == "image") {
Some(r) => (Some(r.file_key.clone()), None),
None => {
let _ = reply_markdown_card(
&ctx.client,
&msg.message_id,
"⚠️ 图片消息缺少 image_key,无法处理。",
)
.await;
return;
}
}
} else if msg.message_type == "file" {
match msg.resources.iter().find(|r| r.resource_type == "file") {
Some(r) => (None, Some(r.clone())),
None => {
let _ = reply_markdown_card(
&ctx.client,
&msg.message_id,
"⚠️ 文件消息缺少 file_key,无法处理。",
)
.await;
return;
}
}
} else if msg.message_type != "text" {
// #25 非文本消息静默丢修复:回提示卡,不再零反馈。文本/图片/文件已支持,
// 余下语音等类型如实告知规划中
let _ = reply_markdown_card(
&ctx.client,
&msg.message_id,
"📎 暂支持文本、图片与文件消息,语音等其他类型规划中。",
)
.await;
return;
} else {
(None, None)
};
// 附件消息不带文字(飞书 image/file 均为独立消息类型),msg.text 的图链/文件标签
// 对模型无意义,不喂:图片喂固定引导词交代任务语境;文件语境需落盘路径,先预填
// 文件名占位版(命令/审批判定恒不命中),下载点覆写为含路径终版
let mut text = if image_key.is_some() {
IMAGE_INBOUND_PROMPT.to_string()
} else if let Some(res) = &file_res {
file_inbound_prompt(res.file_name.trim(), None)
} else {
let text = msg.text.clone().unwrap_or_default().trim().to_string();
// 飞书 @ 占位清洗(如 @_user_1):@ 语义已由 mentioned_bot 承载,占位串只污染文本
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
.as_ref()
.and_then(|u| u.open_id.clone())
.unwrap_or_default();
// 18 探活口径修正:inflight 计数从此(白名单通过即受理)覆盖到投递完成——原计数点
// 在 spawn 前,占位卡 HTTP+per-chat 锁排队(可等上一任务全程)不在覆盖内,该窗口
// 连接假活会误报「无在途会话」。全部早退路径(占位卡失败/注入旁路/常规路径后续
// return)经 RAII 回落,纯过滤段(@/空文本/白名单拒)在计数点之前不受影响。
ctx.inflight.fetch_add(1, Ordering::SeqCst);
let _inflight_guard = InflightGuard(ctx.inflight.clone());
// 白名单(D2 热加载):空=放行;非空且不在内 → 提示并附自己的 open_id(首启自助回填,
// 免翻日志)。每消息现读 KV,设置面板改白名单即生效免重启
// 2026-10-08 配置引导优化:提示改为分步自助引导(原仅附 open_id,用户不知填哪)
let whitelist = read_whitelist(&ctx).await;
if !whitelist.is_empty() && !whitelist.iter().any(|w| w == &open_id) {
let hint = format!(
"⛔ 你不在白名单内,无法使用。\n\n**你的 open_id**:\n`{open_id}`\n\n**开通步骤**:\n1. 复制上方 open_id\n2. 桌面端「设置 → 飞书集成 → 白名单」粘贴(多个用逗号分隔)\n3. 保存后重发消息即可(免重启)"
);
// 用 markdown 卡片(非 reply_message 纯文本),让 **粗体**/`code`/列表 正确渲染
let _ = reply_markdown_card(&ctx.client, &message_id, &hint).await;
return;
}
// 图片入站(#958):下载消息资源→落盘→引用片。置于白名单后(未授权不下载,
// 不替白名单外用户烧流量)、占位发票前(失败早退,无占位卡需原地更新)
let parts = match &image_key {
Some(key) => {
let state = ctx.app.state::<crate::state::AppState>();
let oss_cfg = read_media_oss_config(&state.db).await;
let store = crate::media_store::MediaStore::new(state.data_dir.clone(), oss_cfg);
match download_feishu_image(&ctx.client, &store, &message_id, key).await {
Ok(part) => Some(vec![part]),
Err(e) => {
feishu_log(&format!("图片下载失败 chat={chat_id}: {e}"));
let _ = reply_markdown_card(
&ctx.client,
&message_id,
&format!("⚠️ 图片下载失败:{e}\n\n请重发,或直接用文字描述你的需求。"),
)
.await;
return;
}
}
}
None => None,
};
// 文件入站:下载消息资源→落盘拿本机路径→覆写任务语境终版(含路径,模型 read_file
// 按需读取)。与图片同段白名单后下载(未授权不烧流量)、占位发票前早退(失败回
// 可读错误卡,超 20MB 话术在下载错误内)
if let Some(res) = &file_res {
let state = ctx.app.state::<crate::state::AppState>();
let oss_cfg = read_media_oss_config(&state.db).await;
let store = crate::media_store::MediaStore::new(state.data_dir.clone(), oss_cfg);
match download_feishu_file(&ctx.client, &store, &message_id, res).await {
Ok((name, path)) => {
text = file_inbound_prompt(&name, Some(&path));
}
Err(e) => {
feishu_log(&format!("文件下载失败 chat={chat_id}: {e}"));
let _ = reply_markdown_card(
&ctx.client,
&message_id,
&format!("⚠️ 文件下载失败:{e}\n\n请重发,或直接用文字描述你的需求。"),
)
.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::<crate::state::AppState>().db.clone();
let snapshot = ctx.chat_map.lock().await.clone();
save_chat_map(&db, &snapshot).await;
let state = ctx.app.state::<crate::state::AppState>();
let busy_note = match prev {
Some(cid) if state.conv_states.is_active(&cid) => {
"\n⚠️ 原会话任务仍在后台执行,本群不再接收其推送;如需停止请发 /cancel。"
}
_ => "",
};
reply_message(
&ctx.client,
&message_id,
&chat_id,
&format!("✅ 已重置会话{busy_note}"),
)
.await;
return;
}
// #21 /status 查询当前会话任务状态(话术曾引用不存在的命令,用户发 /status 被当任务
// 烧一轮 LLM)。只读不 spawn:查 conv 绑定与活跃态给进度摘要。
"status" => {
let state = ctx.app.state::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let reply = match conv_id {
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::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let reply = match conv_id {
Some(cid) if state.conv_states.is_active(&cid) => {
feishu_log(&format!("/cancel 停止 conv={cid} chat={chat_id}"));
match crate::commands::ai::ai_stop_loop(ctx.app.clone(), state, cid.clone())
.await
{
Ok(_) => {
if let Some(tx) = cancel_watchers().lock().await.remove(&cid) {
let _ = tx.send(());
}
"🛑 已发送停止指令,任务收尾中。".to_string()
}
Err(e) => format!("⚠️ 停止指令未生效:{}", truncate(&e, 120)),
}
}
_ => "当前无在途任务。".to_string(),
};
reply_message(&ctx.client, &message_id, &chat_id, &reply).await;
return;
}
_ => {}
}
}
// 24 审批决议通路:该会话有挂起审批时,「批准/拒绝」直调审批原语不入对话链路
// (决议文本若落常规路径会被注入运行中任务,审批永远无人拍板)。决议取该会话最新
// 一条挂起;无挂起不拦截,文本照常落常规路径,零行为变更
if text == "批准" || text == "拒绝" {
let state = ctx.app.state::<crate::state::AppState>();
let conv_id = ctx.chat_map.lock().await.get(&chat_id).cloned();
let pending = match conv_id {
Some(ref cid) => latest_pending_approval(&state, cid).await,
None => None,
};
if let Some(tc_id) = pending {
let approved = text == "批准";
feishu_log(&format!("审批决议 chat={chat_id} tc={tc_id} approved={approved}"));
// 2026-10-08 决议后追加执行结果:先订阅(工具执行前,防 publish_event 无订阅者跳过),
// 再调 ai_approve,后 spawn watcher 消费 AiToolCallCompleted 拿结果摘要
let result_rx = state.ai_event_bus.subscribe();
let reply =
match crate::commands::ai::ai_approve(ctx.app.clone(), state, tc_id.clone(), approved).await
{
// 幂等已处理(重复决议/陈旧挂起):ai_approve 返 Ok("已处理(...)"),不误报
Ok(msg) if msg.starts_with("已处理") => {
format!("ℹ️ 该审批已处理过({msg}),无需重复操作。")
}
Ok(_) if approved => "✅ 已批准,任务继续执行。".to_string(),
Ok(_) => "🚫 已拒绝,该工具调用终止,任务按拒绝结果继续。".to_string(),
Err(e) => format!("⚠️ 审批处理失败:{}", truncate(&e, 160)),
};
// 多条挂起提示:本会话仍有其他待审批时追加提醒,防用户以为已全批
// (state 已被 ai_approve move,重新取句柄)
let remaining = match conv_id {
Some(ref cid) => {
let st = ctx.app.state::<crate::state::AppState>();
count_pending_approvals(&st, cid).await
}
None => 0,
};
let reply = if remaining > 0 {
format!("{reply}\n\n⚠️ 本会话还有 {remaining} 条待审批,请继续回复「批准」或「拒绝」。")
} else {
reply
};
reply_message(&ctx.client, &message_id, &chat_id, &reply).await;
// 仅批准且决议成功时订阅执行结果(拒绝不执行工具,无结果可言;已处理幂等跳过)
if approved && !reply.starts_with('⚠') && !reply.starts_with('ℹ') {
spawn_approval_result_watcher(ctx.clone(), chat_id.clone(), tc_id, result_rx);
}
return;
}
}
// ── 纠偏注入旁路(todo.md 16,2026-10-06):会话运行中 → 直接注入运行中 conv,不排队 ──
// 排队语义的痛点:per-chat 锁把补充指令排到 agent 完成后才处理,长任务期间纠偏无效。
// 注入语义:conv 活跃(Generating/Compressed)时直接写 per_conv.messages,loop 下一轮
// build_for_request 自然带上;引擎层能力(chat.rs inject_runtime_message),飞书/桌面同源。
// 非活跃(Idle/Error)→ Err → 落回下方常规路径,零行为变更。
{
let state = ctx.app.state::<crate::state::AppState>();
let conv_id_now = ctx.chat_map.lock().await.get(&chat_id).cloned();
if let Some(cid) = conv_id_now {
// 附件消息(图片/文件)不注入,走与图片相同的后续链:inject_runtime_message
// 只收文本,图片注入则图被丢、模型只见引导词空跑;文件作为独立新任务语义
// 同样应过受理/防抖/发票链,conv 占用时受理守卫如实回告,不并入运行中任务
if image_key.is_none() && file_res.is_none() && state.conv_states.is_active(&cid) {
match crate::commands::ai::commands::chat::inject_runtime_message(&state, &cid, &text).await {
Ok(true) => {
let _ = reply_markdown_card(&ctx.client, &message_id, "✅ 补充指令已注入运行中任务,下一轮生效").await;
return;
}
_ => { /* 注入失败/竞态退出:落回常规路径(spawn 会因 can_accept_request 被拦,收敛安全) */ }
}
}
}
}
// chat_id → conv_id 映射(内存命中优先,未命中新建)
feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80)));
// ── 连续消息防抖聚合(u-claw 同款):同 chat 0.5s 窗内连发合并为一次任务 ──
// 追加消息在此收尾(None),窗满首条携合并载荷继续下方原流程(per-chat 锁→conv
// 映射→受理守卫→spawn/959)
let Some(merged) = debounce_collect(&ctx, &chat_id, &message_id, text, parts).await else {
return;
};
let MergedIncoming { text, parts, placeholder_id } = merged;
// 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 在途计数已前移至 handle_message 白名单通过处(探活口径修正,18),此处不再重复计数
let state = ctx.app.state::<crate::state::AppState>();
// #20 受理守卫:发票(受理卡)前先过引擎读侧闸。不可启动(前任务占用/压缩暂存/停止中)
// 不发票、不 spawn、不注册桥与流式任务,堵「受理卡已发后续零通知」的伪受理断层。
// 959 必达改造(对齐 u-claw 收件箱续投):拒绝路径改落入待发箱(queue 尽快项),
// 前序任务完成调度器自动派发+结果回推本 chat——重试责任不在用户,不再让用户重发。
// 入队失败(写库异常等)降级回原占用话术,消息不丢语义不倒退
if !state.conv_states.can_accept_request(&conv_id) {
feishu_log(&format!("受理守卫拒绝 chat={chat_id} conv={conv_id}"));
match crate::commands::ai::outbox::enqueue_internal(
&ctx.app,
&state,
text.clone(),
None, // 尽快队列项:前序任务空闲即发
None,
None,
conv_id.clone(),
)
.await
{
Ok(_) => {
feishu_log(&format!("已入待发箱 chat={chat_id} conv={conv_id}"));
deliver_result(
&ctx.client,
&placeholder_id,
&message_id,
&chat_id,
"📥 已登记待发:前序任务完成后将自动发出,无需重发(期间可继续追加,均按补充登记)。",
)
.await;
}
Err(e) => {
feishu_log(&format!("待发箱入队失败 chat={chat_id}: {e}"));
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 触发
// tx.send 失败静默,无泄漏(bridge 条目 send 后即移除)。
let (tx, _rx) = tokio::sync::oneshot::channel::<String>();
state.register_agent_completion(conv_id.clone(), tx);
// 过程可见性(13a):预览任务订阅既有 ai_event_bus 拿本 conv 的增量(delta/轮次/工具/
// 进度)节流刷新占位卡;受理卡与终版由完成订阅任务经帧通道投递。本任务为占位卡唯一
// 写者——中间态与终版同任务串行,无迟到中间态覆盖终版的竞态。spawn 前订阅防早期增量丢失
let (frame_tx, frame_rx) = tokio::sync::mpsc::channel::<CardFrame>(8);
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,
frame_rx,
));
let task_brief = truncate(&text, 60); // move 前先取展示摘要(spawn 会 move text)
match crate::commands::ai::commands::chat::spawn_agentic_session(
&ctx.app,
&state,
text,
None,
None,
None,
Some(conv_id.clone()),
parts, // #958 图片入站:图片消息带 vision 片,文本消息 None;vision 能力判定/降级在 spawn 内
None,
false, // 不劫持用户活跃面板(编排节点同款语义)
// #958 溯源:飞书原始消息定位随 user 消息落库,可跳原消息/区分多端来源
Some(df_ai::provider::MessageSource {
channel: "feishu".into(),
message_id: message_id.clone(),
chat_id: Some(chat_id.clone()),
sender: if open_id.is_empty() { None } else { Some(open_id.clone()) },
}),
)
.await
{
Ok(_) => {
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let snapshot = ctx.chat_map.lock().await.clone();
save_chat_map(&db, &snapshot).await;
}
Err(e) => {
// 启动失败:通知桥条目清掉防死条目(编排节点同款)
state.notify_agent_completion(&conv_id, String::new());
// 关帧通道让预览任务退出(此路无任何帧产出,卡片未被写过),
// 待其退出后再走失败投递,保卡片写入单点次序
drop(frame_tx);
let _ = stream_task.await;
feishu_log(&format!("spawn 失败 conv={conv_id}: {e}"));
// 18 渠道错误话术透传:并发护栏拒绝(生成中)≠真失败——固定「启动失败」话术
// 吞掉真实原因,用户误判 bot 坏了重试白撞。按引擎层错误文案分类:
// 生成中→明示任务进行中不会丢;其余→附截断原始原因(不吞因)。
let user_msg = if e.contains("正在生成") || e.contains("生成中") {
// #26 占用期话术改实况:委托制无真排队,禁承诺「任务完成后处理」
QUEUE_BUSY_MSG.to_string()
} else {
format!("❌ 会话启动失败:{},请稍后重试", truncate(&e, 120))
};
deliver_result(
&ctx.client,
&placeholder_id,
&message_id,
&chat_id,
&user_msg,
)
.await;
return;
}
}
// ── #14 任务式委托(2026-10-06):受理即返回,完成走事件驱动 ──
// spawn Ok = 任务已受理并在后台运行(loop 独立 task,30min 超时底座既有,断点续跑兜底)。
// 原同步等待段(timeout 1800s oneshot)整体移除:发「走查代码」级重任务不再挂住本
// 消息处理 30 分钟,后续消息(含注入旁路)即时响应。终版投递改由 completion_watcher
// 消费 ai_event_bus 的 AiAgentFinished 广播(引擎层 deliver_completion_report_if_waited
// 统一发出,单一事实源),原地 edit_card 占位卡为结果。
// 卡片所有权(13a 修订):占位卡唯一写者是预览任务(run_stream_refresher),受理卡与
// 终版都由本完成订阅任务经帧通道交给它按序落卡;预览任务内部串行,消除「受理卡与终版
// 两个 HTTP 请求网络乱序,终版被受理卡覆盖丢失」的竞态窗口。13a 前独立预览任务在受理
// 即被关停(drop 帧通道),心跳/阶段进展形同虚设;此处交回帧通道驱动使其真正落到卡片。
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());
// 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();
let watched_chat = chat_id.clone();
tauri::async_runtime::spawn(async move {
run_completion_watcher(
ctx,
conv_id,
watched_chat,
ack,
inflight_guard,
cancel_rx,
frame_tx,
stream_task,
)
.await;
// watcher 任意路径退出都摘除取消句柄,防陈旧条目驻留(后续 /cancel 空发无害)
cancel_watchers().lock().await.remove(&conv_watched);
});
}
/// #27 快任务宽限窗:受理卡发出后的事件驱动判定窗(原入场 elapsed>=3s 恒假死分支)。
/// 窗内到达的完成事件由广播 rx 缓冲零丢失;窗结束时以「conv 非活跃 + 边界内有产出」
/// 判定秒级任务已完成。8s 覆盖秒级任务;长任务仅推迟主循环进入,不影响终版到达时刻
const FAST_TASK_GRACE: std::time::Duration = std::time::Duration::from_secs(8);
/// 卡片帧:占位卡唯一写者(预览任务)与裁决者(完成订阅任务)之间的投递契约。
/// terminal=true 的帧(终版/取消/失败)写入后预览任务收工退出,保证卡片最后写入即终版。
struct CardFrame {
text: String,
terminal: bool,
}
// ── 出站文件回传(#958,u-claw replyLocalFiles 同模式) ──
// 约定:AI 报告中写 `[交付] <绝对路径>` 行,渠道自动把文件上传飞书回发——
// 图片走图片消息、其余走文件消息(飞书原生附件体验,免用户去桌面端翻)。
// 正文中提及路径但无标记行=叙述,不发送(防止口述路径误上传)。
/// 提取报告中全部 `[交付] <路径>` 行的路径(纯函数)。容错:全半角冒号后空格可选,
/// 路径含空格整行取(行内唯一路径假设,与 AI 约定一致)
fn extract_delivery_paths(text: &str) -> Vec<String> {
text.lines()
.filter_map(|line| {
let line = line.trim();
let rest = line.strip_prefix("[交付]")?;
let path = rest.trim().trim_start_matches(':').trim_start_matches(':').trim();
(!path.is_empty()).then(|| path.to_string())
})
.collect()
}
/// 敏感文件拦截(防御纵深,对齐 u-claw replyBlocked):名字命中凭据/密钥特征即拒,
/// AI 被诱导交付敏感文件时兜底。返回 Some(原因)=拦截
fn delivery_blocked(path: &str) -> Option<&'static str> {
let lower = path.to_lowercase();
const SENSITIVE: [&str; 8] = [
".env", "id_rsa", "credential", "password", "secret", "token", "api_key", "apikey",
];
SENSITIVE
.iter()
.find(|k| lower.contains(**k))
.map(|_| "敏感文件(凭据/密钥特征)")
}
/// 扩展名 → 飞书 file_type(上传接口必填;对齐飞书开放平台枚举)
fn feishu_file_type(ext: &str) -> &'static str {
match ext {
"pdf" => "pdf",
"xls" | "xlsx" | "csv" => "xls",
"doc" | "docx" | "rtf" => "doc",
"ppt" | "pptx" => "ppt",
"mp4" | "mov" | "avi" | "mkv" => "mp4",
"mp3" | "wav" | "m4a" | "aac" | "flac" => "mp3",
_ => "stream",
}
}
/// 上传回传全部交付文件。单文件失败只提示不中断(部分成功优于整体失败);
/// 文件不存在静默跳过并记日志(过滤残余误匹配,"写了路径文件不在"经日志可查)
async fn reply_deliveries(client: &LarkClient, chat_id: &str, text: &str) {
for path in extract_delivery_paths(text) {
if let Some(why) = delivery_blocked(&path) {
feishu_log(&format!("回传跳过({why}): {path}"));
send_text(client, chat_id, &format!("⚠️ 已拦截交付请求:{path}({why})")).await;
continue;
}
let p = std::path::Path::new(&path);
match tokio::fs::metadata(p).await {
Ok(m) if m.is_file() => {}
_ => {
feishu_log(&format!("回传跳过(文件不存在或为目录): {path}"));
continue;
}
}
let bytes = match tokio::fs::read(p).await {
Ok(b) => b,
Err(e) => {
feishu_log(&format!("回传读取失败({path}): {e}"));
continue;
}
};
let opt = &RequestOption::default();
let ext = p
.extension()
.map(|e| e.to_string_lossy().to_lowercase())
.unwrap_or_default();
let is_image = matches!(ext.as_str(), "png" | "jpg" | "jpeg" | "gif" | "webp");
let sent: Result<(), String> = if is_image {
// 图片:上传 image → image 消息直发(直发不做 reply,占位卡已是回复链,
// 附件直发即可,避免嵌套回复噪声)
match client.im().image.create("message", bytes, opt).await {
Ok(resp) => match resp.data.as_ref().and_then(|d| d.image_key.clone()) {
Some(key) => {
let input = SendInput {
msg_type: Some("image".into()),
image_key: Some(key),
chat_id: Some(chat_id.to_string()),
..Default::default()
};
client
.channel_messaging()
.send(&input, opt)
.await
.map(|_| ())
.map_err(|e| format!("发送失败: {e}"))
}
None => Err("响应缺 image_key".into()),
},
Err(e) => Err(format!("图片上传失败: {e}")),
}
} else {
// 文件:上传 file → file 消息直发(飞书好友互发文件同款体验)
let name = p
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| "file".into());
match client
.im()
.file
.create(feishu_file_type(&ext), &name, None, bytes, opt)
.await
{
Ok(resp) => match resp.data.as_ref().and_then(|d| d.file_key.clone()) {
Some(key) => {
let input = SendInput {
msg_type: Some("file".into()),
file_key: Some(key),
chat_id: Some(chat_id.to_string()),
..Default::default()
};
client
.channel_messaging()
.send(&input, opt)
.await
.map(|_| ())
.map_err(|e| format!("发送失败: {e}"))
}
None => Err("响应缺 file_key".into()),
},
Err(e) => Err(format!("文件上传失败: {e}")),
}
};
if let Err(e) = sent {
feishu_log(&format!("回传失败({path}): {e}"));
let base = path.rsplit(['/', '\\']).next().unwrap_or(path.as_str());
send_text(client, chat_id, &format!("⚠️ 文件回传失败:{base}({e})")).await;
} else {
feishu_log(&format!("回传完成: {path}"));
}
}
}
/// 完成事件订阅任务(#14 两段式):只做裁决——消费 ai_event_bus 的 AiAgentFinished(本 conv)
/// 决定终版文本,经帧通道交给预览任务原地落卡(占位卡唯一写者是预览任务,受理卡/终版/阶段
/// 提示全部经帧通道按序落卡)。引擎层 deliver_completion_report_if_waited 是广播单一发出点。
///
/// 超时语义:35min(> loop 30min 上限)首超时回查边界内产出,无果投降级提示但不退出
/// (任务后台不丢,晚到完成事件仍送达);二次超时真弃守(#23:两轮均无事件且无边界内
/// 产出,认任务失联,停止重刷)。
async fn run_completion_watcher(
ctx: Arc<Ctx>,
conv_id: String,
chat_id: String,
ack: String,
// #20:计数生命周期跟任务走,guard 持有至 watcher 退出
_inflight: InflightGuard,
// 24 取消收口:/cancel 旁路唤醒(引擎停止路径只 emit AiCompleted,本任务消费不到)
mut cancel_rx: tokio::sync::oneshot::Receiver<()>,
// 13a:占位卡写权限交预览任务,本任务所有落卡经此投递(terminal 标记终版)
frame_tx: tokio::sync::mpsc::Sender<CardFrame>,
// 预览任务句柄:投终版后 await 其退出,保证落卡完成后才释放 inflight
stream_task: tauri::async_runtime::JoinHandle<()>,
) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
// #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);
// 落受理卡(经预览任务原地更新占位卡,未收到终版前预览任务持续刷新过程进展)
let _ = frame_tx.send(CardFrame { text: ack, terminal: false }).await;
// #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()));
let _ = frame_tx.send(CardFrame { text: t, terminal: true }).await;
let _ = stream_task.await;
return;
}
}
}
// 事件等待循环:首超时投降级提示不退出;二次超时真弃守(#23)
let timeout = std::time::Duration::from_secs(2100);
let mut timeouts = 0usize;
loop {
let frame = tokio::select! {
// 24 取消收口:/cancel 旁路唤醒,投「已停止」终版后退出,不再等引擎事件
_ = &mut cancel_rx => {
feishu_log(&format!("watcher 收到取消指令 conv={conv_id}"));
CardFrame { text: "🛑 任务已停止。可发 /status 查看会话状态,或直接发新任务。".to_string(), terminal: true }
}
item = rx.recv() => match item {
Ok(v) => {
if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) {
continue;
}
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}"));
CardFrame {
text: format!("❌ 任务执行失败:{},可稍后重试,或发 /status 查看会话状态。", truncate(err, 300)),
terminal: true,
}
}
Some("AiAgentFinished") => {
// 广播语义:agent_id==conv_id 时 summary 即本会话最终报告(引擎层构造)
CardFrame {
text: v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string(),
terminal: true,
}
}
_ => 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 {
feishu_log(&format!("watcher 二次超时弃守 conv={conv_id}"));
CardFrame {
text: "⏰ 任务超 70 分钟无任何结果(已停止等待;引擎仍在后台运行则结果稍后可在桌面端查看;可发 /status 查询)。".to_string(),
terminal: true,
}
} else {
// 首超时:DB 回查兜底(任务可能刚完成事件已发完);回查有果照投终版,
// 无果投降级提示并继续等待(晚到事件仍送达)
match last_assistant_reply(&state.db, &conv_id, Some(accepted_ms)).await {
Some(t) if !t.trim().is_empty() => CardFrame { text: t, terminal: true },
_ => {
let _ = frame_tx.send(CardFrame {
text: "⏰ 任务执行超 35 分钟仍未返回(后台仍在运行,完成后本卡片自动更新;或本会话 /status 查询)。".to_string(),
terminal: false,
}).await;
continue;
}
}
}
}
};
feishu_log(&format!("watcher 终版 conv={conv_id} len={}", frame.text.len()));
// #958 出站回传(u-claw replyLocalFiles 同模式):报告内 [交付] 标记行的本地文件
// 自动上传飞书(图片→图片消息/其余→文件消息),用户在飞书直接收原生附件。
// 终版文本中的标记行本身保留(AI 写路径也是叙述的一部分,卡内可见)
reply_deliveries(&ctx.client, &chat_id, &frame.text).await;
let _ = frame_tx.send(frame).await;
let _ = stream_task.await; // 等预览任务落卡终版并退出,再释放 inflight
return;
}
}
/// 24 审批可见性:渠道生命周期订阅任务。消费 ai_event_bus 的 AiApprovalRequired,反查
/// conv 绑定的飞书 chat 推卡提示(回复「批准」或「拒绝」);无绑定(纯桌面会话)不推,
/// 桌面 ApprovalOverlay 既有通路处理。审批挂起时流式卡静止无解释,本推送补齐可见性,
/// 决议闭环由 handle_message 的「批准/拒绝」拦截承接
async fn run_approval_notifier(ctx: Arc<Ctx>) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
loop {
match rx.recv().await {
Ok(v) => {
if v.get("type").and_then(|t| t.as_str()) != Some("AiApprovalRequired") {
continue;
}
let Some(cid) = v.get("conversation_id").and_then(|c| c.as_str()) else {
continue;
};
let chat_id = ctx
.chat_map
.lock()
.await
.iter()
.find(|(_, c)| c.as_str() == cid)
.map(|(k, _)| k.clone());
let Some(chat_id) = chat_id else {
continue;
};
let name = v.get("name").and_then(|n| n.as_str()).unwrap_or("工具");
let reason = v
.get("reason")
.and_then(|r| r.as_str())
.unwrap_or("")
.replace('\n', " ");
let brief = truncate(&reason, 80);
// 审批卡按钮化(2026-10-08):改用带「同意/拒绝」按钮的交互卡片,
// 按钮 value 携带 {approval_id, approved},回调经 on_card_action →
// handle_card_action → ai_approve 原语(与文本决议通路同原语,双通道并存)。
// 文本「批准/拒绝」通路保留作降级(旧卡片/按钮回调异常时仍可用)。
let tc_id = v.get("id").and_then(|i| i.as_str()).unwrap_or("");
if tc_id.is_empty() {
// 无 tool_call_id 无法构造按钮载荷,降级文本卡(回复「批准/拒绝」)
let card = if brief.is_empty() {
format!("⏸ 需审批: `{name}`。回复「批准」或「拒绝」继续")
} else {
format!("⏸ 需审批: `{name}` {brief}。回复「批准」或「拒绝」继续")
};
send_markdown_card(&ctx.client, "", &chat_id, &card).await;
} else {
send_approval_button_card(&ctx.client, &chat_id, tc_id, name, &brief).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,
}
}
}
/// 卡片按钮回调(2026-10-08 审批卡按钮化):解析按钮 value 的 {approval_id, approved},
/// 经 ai_approve 原语决议。与文本「批准/拒绝」通路同原语,双通道并存(按钮优先,
/// 文本作降级)。回调在独立 spawn 内执行,不阻塞事件泵。
async fn handle_card_action(
ctx: Arc<Ctx>,
action: larksuite_oapi_sdk_rs::channel::NormalizedCardAction,
) {
// 按钮约定的 value 形态:{"approval_id": "<tool_call_id>", "approved": true|false}
let approval_id = action
.action_value
.get("approval_id")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
if approval_id.is_empty() {
tracing::warn!("[feishu] 卡片回调缺 approval_id,忽略");
return;
}
let approved = action
.action_value
.get("approved")
.and_then(|v| v.as_bool())
.unwrap_or(false);
// chat_id 兜底:SDK 归一化未带 open_chat_id 时,从 approval_id 反查 conv → chat_map
// 反查绑定 chat,防 send_markdown_card(client, "", "", ...) 发到空 chat 静默失败
let mut chat_id = action.open_chat_id.clone().unwrap_or_default();
if chat_id.is_empty() {
let state = ctx.app.state::<crate::state::AppState>();
if let Some(cid) = approval_conversation_id(&state, &approval_id).await {
chat_id = ctx
.chat_map
.lock()
.await
.iter()
.find(|(_, c)| c.as_str() == cid)
.map(|(k, _)| k.clone())
.unwrap_or_default();
if !chat_id.is_empty() {
feishu_log(&format!("卡片回调 chat_id 缺失,反查补齐 chat={chat_id}"));
}
}
}
if chat_id.is_empty() {
// 仍拿不到落点:决议照常执行(不能因投递无门而丢决议),但记日志便于排查
tracing::warn!("[feishu] 卡片回调 chat_id 缺失且反查失败,回执无法投递 tc={approval_id}");
feishu_log(&format!("卡片回调 chat_id 缺失且反查失败 tc={approval_id}"));
}
// 白名单校验(与 handle_message 同口径):卡片按钮回调此前无校验,非白名单用户
// 若能看到审批卡(被转发/群内他人),点按钮可直接批准工具执行,白名单形同虚设。
// 取回调发起者 open_id 校验;取不到(空)时保守拒绝(宁拒勿放)
let operator_open_id = action.operator_open_id.clone().unwrap_or_default();
let whitelist = read_whitelist(&ctx).await;
if !whitelist.is_empty() && !whitelist.iter().any(|w| w == &operator_open_id) {
feishu_log(&format!(
"卡片回调白名单拒绝 chat={chat_id} operator={operator_open_id}"
));
if !chat_id.is_empty() {
let _ = send_markdown_card(
&ctx.client,
"",
&chat_id,
"⛔ 你不在白名单内,无权审批。请在桌面端「设置 → 飞书集成 → 白名单」添加你的 open_id。",
)
.await;
}
return;
}
feishu_log(&format!(
"审批按钮回调 chat={chat_id} tc={approval_id} approved={approved}"
));
let state = ctx.app.state::<crate::state::AppState>();
// 2026-10-08 决议后追加执行结果:先订阅(工具执行前,防 publish_event 无订阅者跳过)
let result_rx = state.ai_event_bus.subscribe();
// 结果回执发新消息(不依赖 NormalizedCardAction 的 open_message_id 字段,
// 该字段在长连接归一化结构中的存在性未经 SDK 源码证实;open_chat_id 已由网关实现验证)
let reply = match crate::commands::ai::ai_approve(
ctx.app.clone(),
state,
approval_id.clone(),
approved,
)
.await
{
// 幂等已处理(重复点击/陈旧卡片):ai_approve 返 Ok("已处理(...)"),不能误报「已批准」
Ok(msg) if msg.starts_with("已处理") => {
format!("ℹ️ 该审批已处理过({msg}),无需重复操作。")
}
Ok(_) if approved => "✅ 已批准,任务继续执行。".to_string(),
Ok(_) => "🚫 已拒绝,该工具调用终止,任务按拒绝结果继续。".to_string(),
Err(e) => format!("⚠️ 审批处理失败:{}", truncate(&e, 160)),
};
let _ = send_markdown_card(&ctx.client, "", &chat_id, &reply).await;
// 仅批准且决议成功时订阅执行结果(拒绝不执行工具,无结果可言);
// 已处理幂等路径也跳过(工具早已执行完,结果无从取)
if approved && !reply.starts_with('⚠') && !reply.starts_with('ℹ') {
spawn_approval_result_watcher(ctx.clone(), chat_id.clone(), approval_id, result_rx);
}
}
/// 审批决议后执行结果 watcher(2026-10-08):订阅 ai_event_bus 的 AiToolCallCompleted
/// (匹配 tool_call_id),拿 result 摘要回推 chat。原决议后只回「已批准」,用户不知工具
/// 执行结果(成功/失败/输出)。设 60s 超时兜底(工具执行通常秒级,长工具超时静默放弃)。
///
/// 订阅须在 ai_approve 调用前完成——publish_event 在无订阅者时跳过序列化,晚订阅必丢事件。
fn spawn_approval_result_watcher(
ctx: Arc<Ctx>,
chat_id: String,
tool_call_id: String,
mut rx: tokio::sync::broadcast::Receiver<serde_json::Value>,
) {
tauri::async_runtime::spawn(async move {
let timeout = std::time::Duration::from_secs(60);
let deadline = tokio::time::Instant::now() + timeout;
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
feishu_log(&format!("[approval-result] 等待工具结果超时 tc={tool_call_id}"));
return;
}
match tokio::time::timeout(remaining, rx.recv()).await {
Ok(Ok(v)) => {
if v.get("type").and_then(|t| t.as_str()) != Some("AiToolCallCompleted") {
continue;
}
if v.get("id").and_then(|i| i.as_str()) != Some(tool_call_id.as_str()) {
continue;
}
// result 是 serde_json::Value,提取文本摘要(字符串直用,对象转 JSON 截断)
let summary = match v.get("result") {
Some(serde_json::Value::String(s)) => truncate(s, 500),
Some(other) => truncate(&other.to_string(), 500),
None => "(无输出)".to_string(),
};
let text = format!("📋 工具执行结果:\n{summary}");
let _ = send_markdown_card(&ctx.client, "", &chat_id, &text).await;
feishu_log(&format!("[approval-result] 已回推结果 tc={tool_call_id}"));
return;
}
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => return,
Err(_) => {
feishu_log(&format!("[approval-result] 等待工具结果超时 tc={tool_call_id}"));
return;
}
}
}
});
}
/// 审批按钮卡片(2026-10-08):标题 + 详情 + 「同意/拒绝」按钮。
/// 按钮 value 携带 {approval_id, approved},点击触发 card.action.trigger 事件,
/// 经 on_card_action → handle_card_action → ai_approve。发送失败降级文本卡(回复「批准/拒绝」)。
async fn send_approval_button_card(
client: &LarkClient,
chat_id: &str,
approval_id: &str,
tool_name: &str,
brief: &str,
) {
let detail = if brief.is_empty() {
format!("⏸ 需审批: `{tool_name}`")
} else {
format!("⏸ 需审批: `{tool_name}`\n\n{brief}")
};
// 按钮 value 需为 SDK 的 JsonValue(其实现了 From<serde_json::Value>,直接 into 转换)
let to_json_value = |v: serde_json::Value| -> larksuite_oapi_sdk_rs::JsonValue { v.into() };
let actions = Action::new()
.action(ActionComponent::Button(
Button::new(larksuite_oapi_sdk_rs::card::v1::Text::plain("✅ 同意"))
.button_type(ButtonType::Primary)
.value(to_json_value(serde_json::json!({
"approval_id": approval_id,
"approved": true
}))),
))
.action(ActionComponent::Button(
Button::new(larksuite_oapi_sdk_rs::card::v1::Text::plain("❌ 拒绝"))
.button_type(ButtonType::Danger)
.value(to_json_value(serde_json::json!({
"approval_id": approval_id,
"approved": false
}))),
));
let card = Card::new()
.header(Header::new("🔐 工具审批请求"))
.element(Element::Div(Div::new(
larksuite_oapi_sdk_rs::card::v1::Text::lark_md(md_to_lark_friendly(&detail)),
)))
.element(Element::Action(actions));
let document = match CardDocument::new(card) {
Ok(d) => d,
Err(e) => {
tracing::warn!("[feishu] 审批按钮卡校验失败,降级文本卡: {e}");
let fallback = format!("⏸ 需审批: `{tool_name}` {brief}。回复「批准」或「拒绝」继续");
send_markdown_card(client, "", chat_id, &fallback).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()
};
match client
.channel_messaging()
.send(&input, &RequestOption::default())
.await
{
Ok(sent) if !sent.message_id.is_empty() => {
feishu_log(&format!("审批按钮卡已发 chat={chat_id} card={}", sent.message_id));
}
Ok(_) => {
tracing::warn!("[feishu] 审批按钮卡发送未返回 message_id");
}
Err(e) => {
tracing::error!("[feishu] 审批按钮卡发送失败,降级文本卡: {e}");
let fallback = format!("⏸ 需审批: `{tool_name}` {brief}。回复「批准」或「拒绝」继续");
send_markdown_card(client, "", chat_id, &fallback).await;
}
}
}
/// 该 conv 最新一条挂起审批的 tool_call_id(决议原语的消费目标);无挂起返回 None。
/// created_at 缺失视为最旧(宁取有时序的),同刻并列取遍历后者(HashMap 无序,可接受)
async fn latest_pending_approval(
state: &crate::state::AppState,
conv_id: &str,
) -> Option<String> {
let session = state.ai_session.lock().await;
session
.pending_approvals
.iter()
.filter(|(_, a)| a.conversation_id.as_deref() == Some(conv_id))
.max_by_key(|(_, a)| {
a.created_at
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0)
})
.map(|(k, _)| k.clone())
}
/// 按 tool_call_id 反查其所属 conversation_id(卡片回调 chat_id 缺失时反查补齐用)
async fn approval_conversation_id(
state: &crate::state::AppState,
tool_call_id: &str,
) -> Option<String> {
let session = state.ai_session.lock().await;
session
.pending_approvals
.get(tool_call_id)
.and_then(|a| a.conversation_id.clone())
}
/// 该 conv 挂起审批计数(多条挂起提示用):决议一条后仍有 N 条时提醒用户继续
async fn count_pending_approvals(
state: &crate::state::AppState,
conv_id: &str,
) -> usize {
let session = state.ai_session.lock().await;
session
.pending_approvals
.iter()
.filter(|(_, a)| a.conversation_id.as_deref() == Some(conv_id))
.count()
}
/// 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;
/// 13a-2 静默期心跳阈值:超此秒数无本 conv 内容进展(delta/轮次/工具/进度),tick 强制刷心跳行。
/// 60s 取「长工具常见执行时长」与「卡片冻结感阈值」之间——再短会与常规刷新混叠,
/// 再长用户已开始疑心卡死。心跳行此后每静默分钟刷一次(桶去重),内容恢复即自然停。
/// 注意:AiHeartbeat 等引擎保活事件不计入活动钟(非内容进展),否则心跳节拍会把静默永远压在阈值内
const HEARTBEAT_SILENT_THRESHOLD: u64 = 60;
/// 13a-2 心跳刷判定(纯函数,单测锁定):内容静默达阈值且进入新的静默分钟桶时返回该桶号
/// (应刷),否则 None(不刷)。桶去重保证每个静默分钟只刷一次,内容恢复(静默回落)即停。
fn heartbeat_due(silent_secs: u64, painted_bucket: Option<u64>, threshold: u64) -> Option<u64> {
if silent_secs < threshold {
return None;
}
let bucket = silent_secs / 60;
if Some(bucket) == painted_bucket {
None
} else {
Some(bucket)
}
}
/// 流式预览任务:订阅 ai_event_bus 的增量(delta/轮次/工具/进度)节流把累计内容 edit_card
/// 进占位卡;受理卡/阶段提示/终版经帧通道(frame_rx)传入,本任务是占位卡唯一写者。
///
/// 竞态设计:占位卡在本任务存活期只有本任务写,中间态 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 frame_rx: tokio::sync::mpsc::Receiver<CardFrame>,
) {
let mut acc = String::new();
// 累计字符数 O(1) 维护,与上次已刷数比对判"有无新增量",避免无变化空刷烧配额
let mut acc_chars = 0usize;
let mut painted_chars = 0usize;
// ── 过程流增量(todo.md 13):轮次/工具轨迹提炼,卡片头部一行摘要 ──
// 长任务(206 轮级)纯工具轮无文本增量,原版卡片静止疑似卡死;轨迹行给出「在干活」证据。
// 提炼原则:决策价值过滤——只留「第 N 轮 + 当前工具名」,历史工具折叠计数,不刷屏不顶正文。
let mut round: Option<u32> = None;
let mut current_tool: Option<String> = None;
// 13a-2 最近动作:工具完成后 current_tool 清空,但「最近在干什么」仍有展示价值(心跳行数据源)
let mut last_tool: Option<String> = None;
let mut tool_steps: usize = 0usize;
let mut painted_sig = String::new();
// 13a-3 阶段进度:最近一次 AiTaskProgress 快照(round/max/tokens),心跳行与头行数据源
let mut progress_round: Option<u32> = None;
let mut progress_max: Option<u32> = None;
let mut progress_tokens: Option<u64> = None;
// 13a-1 刷新可观测:流式刷失败连败计数(成功归零;==阈值告警一次,继续累计防重复告警)
let mut refresh_failures: u32 = 0;
// 13a-2 静默期心跳:最后内容进展时刻(delta/轮次/工具/进度;不含 AiHeartbeat);静默超阈值
// 强制刷心跳行,防长工具/LLM 长往返期间卡片冻结在旧帧,用户误判卡死(todo.md 13a-2)
let mut last_activity = tokio::time::Instant::now();
let mut painted_hb_bucket: Option<u64> = None;
// 受理卡(首帧)到达前不写卡:防心跳/预览帧先于受理卡落卡,次序错乱
let mut started = false;
let mut ticker = tokio::time::interval(STREAM_FLUSH_INTERVAL);
// 刷新落后只保最新帧不补帧:预览语义是最新输出才有价值
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
// 帧通道:受理卡/阶段提示/终版。走 deliver_result(净化+裁剪+新发兜底),与预览同任务串行
frame = frame_rx.recv() => {
let Some(f) = frame else {
// 发送端丢弃(watcher 退出/受理失败):无更多帧,收工
return;
};
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &f.text).await;
started = true;
if f.terminal {
return;
}
}
item = delta_rx.recv() => match item {
Ok(v) => {
// conv 路由过滤(全事件共用);type tag 即变体名(AiChatEvent 无 serde rename)
if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) {
continue;
}
match v.get("type").and_then(|t| t.as_str()) {
Some("AiTextDelta") => {
if let Some(delta) = v.get("delta").and_then(|d| d.as_str()) {
acc.push_str(delta);
acc_chars += delta.chars().count();
// 首帧提速(2026-10-08):首个 delta 到达且受理卡已落地即立即刷占位卡,
// 不等 3s ticker——短响应(简单问答/翻译)原需等满一个节流周期才见首字,
// 感知停顿明显。仅首帧走即时刷,后续仍按 3s 节流守飞书 QPS 配额
if started && painted_chars == 0 && acc_chars > 0 {
if let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) {
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, None, progress_snapshot)).await {
painted_chars = acc_chars;
}
}
}
}
}
// 轮次推进:立即标脏强制下一 tick 刷卡(纯工具轮无文本也有"活着"信号)
Some("AiAgentRound") => {
if let Some(r) = v.get("round").and_then(|r| r.as_u64()) {
round = Some(r as u32);
painted_chars = painted_chars.saturating_sub(1); // 标脏:强制重刷
}
}
// 工具开始:更新当前工具名;完成(含失败)即清空当前工具
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); // 标脏
}
}
Some("AiToolCallCompleted") => {
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); // 标脏
}
}
// AiHeartbeat 等引擎保活事件不重置内容静默钟:否则 20s 心跳节拍会把静默
// 永远压在阈值内,13a-2 心跳行永不触发(静默指内容进展停顿,非事件停顿)
_ => continue,
}
// 13a-2 活动时钟:本 conv 内容事件(delta/轮次/工具/进度)算活着,重置静默计时
last_activity = tokio::time::Instant::now();
}
// 预览允许丢帧:终版事实源在帧通道,消费落后只降级预览不断链路
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::warn!("[feishu] 流式预览消费落后,丢弃 {n} 条增量(预览降级,终版不受影响)");
}
// 总线随 AppState 常驻理论不关;防御性收口:总线关闭即收工(进程收尾)
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
},
_ = ticker.tick() => {
// 受理卡未落地前不刷:防预览/心跳帧抢先覆盖受理卡
if !started {
continue;
}
// 占位缺失(首张卡片 reply 失败)时无从原地刷,保持静默等终版走新发回退
let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) else {
continue;
};
// ── 13a-2 静默期心跳:内容静默达阈值且进入新静默分钟桶 → 刷心跳行(纯函数判定) ──
// 长工具/LLM 长往返期间无 delta 无轮次推进,卡片静止;心跳行给出「引擎活着+
// 进行到哪」证据。桶去重保证每静默分钟只刷一次,内容恢复即自然停
let silent_secs = last_activity.elapsed().as_secs();
let hb_bucket = heartbeat_due(silent_secs, painted_hb_bucket, HEARTBEAT_SILENT_THRESHOLD);
let hb_needed = hb_bucket.is_some();
// 常规信号标脏:过程事件已回拨 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, hb_bucket.unwrap_or(u64::MAX));
if sig == painted_sig {
continue;
}
// 13a-1 刷新可观测:刷失败落盘+连败计数,≥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 let Some(b) = hb_bucket {
painted_hb_bucket = Some(b); // 心跳桶已刷,本静默分钟不再刷
}
} 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} 次刷新失败,卡片可能冻结——流式预览降级,终版仍走帧通道(不受影响)"
));
}
}
}
}
}
}
/// 流式中间态卡片:头部一行过程摘要(轮次/工具轨迹提炼)+ 正文流式增量(保尾截断)。
///
/// 提炼原则(todo.md 13「决策价值过滤」):历史步骤折叠为计数,只透出「当前在干什么」——
/// 读者要的是"活着没卡死+进行到哪",不是 20 个工具名的流水墙;正文仍以流式文本为主,
/// 纯工具轮(无文本)时头部行是唯一进度证据。13a-2:静默超阈值时追加心跳行
/// 「💓 仍在跑:第 X 轮 · 最近动作 · 静默 Ys」——长工具/LLM 长往返期间卡片冻结疑似
/// 卡死,心跳行是「引擎活着」的用户可见证据。
fn stream_preview(
acc: &str,
round: Option<u32>,
current_tool: Option<&str>,
last_tool: Option<&str>,
tool_steps: usize,
silent_secs: Option<u64>,
progress: Option<(u32, Option<u32>, Option<u64>)>,
) -> String {
// 头部摘要行:轮次 + 当前工具(执行中)/已执行步数;全部缺省(首轮文本前)不显头行。
// 13a-3:round 缺省时回退 AiTaskProgress 的 round(里程碑轮快照),并附累计 token
let mut head = String::new();
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}{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();
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!("{head}{body}\n▌")
}
/// 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
pub(crate) async fn send_markdown_card(
client: &LarkClient,
message_id: &str,
chat_id: &str,
text: &str,
) -> Option<String> {
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 None;
}
};
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 mut replied: Option<String> = 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
}
}
}
/// 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
}
}
}
/// 结果投递统一入口:先做出站净化(38a,update_card 原地路径不经 send_markdown_card,
/// 须在此净化;回退新发路径二次进入幂等)与裁剪,优先原地更新占位卡片(占位与结果合一);
/// 占位缺失或更新失败回退新发卡片(内含纯文本兜底)
async fn deliver_result(
client: &LarkClient,
placeholder_id: &Option<String>,
message_id: &str,
chat_id: &str,
text: &str,
) {
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("占位卡片更新失败,回退新发卡片");
}
let _ = send_markdown_card(client, message_id, chat_id, &text).await;
}
/// 19 outbound 统一出口:主动推送(无占位卡上下文的外部推送方——挂机进度汇总/巡检通知
/// 等将来形态)。降级/裁剪/卡片链与 deliver_result 同源(禁再开独立发送路径绕过降级链,
/// 11:23 实证:绕过链的回复带完整 md 表格未降级,踩回表格渲染不稳坑)。
/// 34 滚动汇总卡:同 chat 在途事务卡存在时原地追加(句柄见 ROLLING_CARDS),✅/❌ 终态
/// 更新后清句柄下条另起新卡;原地更新失败清句柄回退新发并重绑(比照 deliver_result 兜底)。
/// 29 推送路由:conv_hint 按绑定关系精确反查 chat;缺省/未命中时单绑定直用(单绑定行为
/// 不变),多绑定取最近活跃,不再取 KV 首条盲投。渠道未启动(无 client)或无映射时静默跳过。
/// 19 正典出口预留给将来推送方(挂机进度汇总/巡检通知),当前无调用方
#[allow(dead_code)]
pub(crate) async fn push_to_chat(app: &tauri::AppHandle, text: &str, conv_hint: Option<&str>) {
let Some(client) = ACTIVE_CLIENT.get() else {
return; // 渠道未启动:推送无落点,静默跳过(调用方自行决定是否记日志)
};
let db = app.state::<crate::state::AppState>().db.clone();
let repo = SettingsRepo::new(&db);
let raw = repo.get(KV_CHAT_MAP).await.ok().flatten();
let Some(map) = raw.and_then(|r| serde_json::from_str::<HashMap<String, String>>(&norm(Some(r))).ok())
else {
return;
};
let Some(chat_id) = resolve_push_chat(&db, &map, conv_hint).await else {
return;
};
// 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 卡片
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()));
}
/// 29 推送路由裁决:conv_hint 有绑定命中即用(多命中取最近活跃);缺省/未命中时单绑定
/// 直用(行为与旧首条取法一致),多绑定取最近活跃。无绑定返回 None 静默跳过
async fn resolve_push_chat(
db: &df_storage::db::Database,
map: &HashMap<String, String>,
conv_hint: Option<&str>,
) -> Option<String> {
if let Some(hint) = conv_hint {
let hits: Vec<&String> = map
.iter()
.filter(|(_, c)| c.as_str() == hint)
.map(|(k, _)| k)
.collect();
match hits.len() {
1 => return Some(hits[0].clone()),
n if n > 1 => return most_recent_chat(db, &hits).await,
_ => {} // hint 未命中:绑定可能已重建,落多绑定裁决
}
}
match map.len() {
0 => None,
1 => map.keys().next().cloned(),
_ => {
let all: Vec<&String> = map.keys().collect();
most_recent_chat(db, &all).await
}
}
}
/// 多绑定裁决:逐 conv 取最后一条消息时间戳,最新者胜;全部无消息取遍历首个(稳定退化)
async fn most_recent_chat(db: &df_storage::db::Database, chats: &[&String]) -> Option<String> {
use df_storage::crud::AiMessageRepo;
let repo = AiMessageRepo::new(db);
let mut best: Option<(&String, i64)> = None;
for cid in chats {
let ts = repo
.list_by_conversation(cid)
.await
.ok()
.and_then(|msgs| msgs.last().and_then(|m| m.timestamp))
.unwrap_or(i64::MIN);
if best.map_or(true, |(_, bts)| ts > bts) {
best = Some((cid, ts));
}
}
best.map(|(k, _)| (*k).clone())
}
/// 卡片正文长度防御:飞书卡片 JSON 约 30KB 上限,超长会让原地更新与新发双双失败,
/// 一路穿透到纯文本兜底仍不可读;入口统一裁剪,尾注指回桌面端(u-claw archiveIfClipped 同口径)
fn clip_for_card(text: &str, max_chars: usize) -> String {
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) {
let input = SendInput {
receive_id: Some(chat_id.to_string()),
text: Some(text.to_string()),
..Default::default()
};
if let Err(e) = client.channel_messaging().send(&input, &RequestOption::default()).await {
tracing::error!("[feishu] 发送失败: {e}");
}
}
/// 剥离飞书 @ 占位串(如 @_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('|')
}
/// 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<String> = 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<String> {
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();
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(":"))
}
/// 13a-1 刷新可观测:流式刷连败告警阈值(== 触发一次,继续累计不重复告警)
const REFRESH_FAIL_ALERT_THRESHOLD: u32 = 3;
/// 关键事件日志落盘(零依赖 fs append;Roaming 下 feishu-channel.log,排查空回复等复现凭据)
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");
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()
}
/// #22 时间戳过滤判据(纯函数,单测锁定):给定受理时刻时只认其后落库的消息,
/// 消息缺时间戳视为不在边界内(宁缺勿冒充旧回答);无受理时刻不过滤
fn msg_after_acceptance(msg_ts: Option<i64>, accepted_ms: Option<i64>) -> 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<i64>,
) -> 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()
&& msg_after_acceptance(m.timestamp, only_after_ms)
})
.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::{
ContentPart, DebounceEntry, delivery_blocked, extract_delivery_paths, feishu_file_type,
file_display_name, file_inbound_prompt, file_size_over_limit_error, heartbeat_due,
is_terminal_push, md_to_lark_friendly, merge_debounce_parts, merge_debounce_text,
msg_after_acceptance, parse_command, sniff_image_mime, strip_mention_placeholders,
stream_preview,
};
#[test]
fn delivery_paths_extract_only_marked_lines() {
// #958 出站回传:只认 [交付] 标记行,叙述路径不发送;全半角冒号容错
let text = "方案如下。\n已生成图表:\n[交付] E:\\out\\chart.png\n另见 E:\\ref\\说明.docx\n[交付]:/home/u/报表 最终.xlsx\n[交付]";
assert_eq!(
extract_delivery_paths(text),
vec!["E:\\out\\chart.png".to_string(), "/home/u/报表 最终.xlsx".to_string()]
);
assert!(extract_delivery_paths("没有标记").is_empty());
}
#[test]
fn delivery_blocked_catches_credential_names() {
assert!(delivery_blocked("E:\\proj\\.env").is_some());
assert!(delivery_blocked("/home/u/id_rsa").is_some());
assert!(delivery_blocked("E:\\app\\API_KEY.txt").is_some());
assert_eq!(delivery_blocked("E:\\out\\chart.png"), None);
assert_eq!(delivery_blocked("E:\\out\\report.pdf"), None);
}
#[test]
fn feishu_file_type_maps_common_exts() {
assert_eq!(feishu_file_type("pdf"), "pdf");
assert_eq!(feishu_file_type("xlsx"), "xls");
assert_eq!(feishu_file_type("docx"), "doc");
assert_eq!(feishu_file_type("mp4"), "mp4");
assert_eq!(feishu_file_type("zip"), "stream");
assert_eq!(feishu_file_type(""), "stream");
}
#[test]
fn image_mime_sniff_four_formats() {
// #958:四格式魔数命中 + 未知格式拒收(宁早拒不晚炸 vision 端点 400)
assert_eq!(sniff_image_mime(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A]), Some("image/png"));
assert_eq!(sniff_image_mime(&[0xFF, 0xD8, 0xFF, 0xE0]), Some("image/jpeg"));
assert_eq!(sniff_image_mime(b"GIF89a"), Some("image/gif"));
assert_eq!(
sniff_image_mime(b"RIFF\x24\x08\x00\x00WEBPVP8 "),
Some("image/webp")
);
assert_eq!(sniff_image_mime(b"<html>"), None);
assert_eq!(sniff_image_mime(&[]), None);
// RIFF 但非 WEBP(avi/wav 等)不误判
assert_eq!(sniff_image_mime(b"RIFF\x24\x08\x00\x00AVI "), None);
}
// ── 文件消息入站:任务语境构造/大小限流/展示名回退 ──
#[test]
fn file_prompt_carries_name_path_and_read_guidance() {
// 落盘后终版:文件名+本机路径+read_file 引导+读取局限交底,四要素齐备
let full = file_inbound_prompt("报表.xlsx", Some("E:\\data\\media\\ab12_报表.xlsx"));
assert!(full.contains("「报表.xlsx」"), "缺文件名: {full}");
assert!(full.contains("本地路径 E:\\data\\media\\ab12_报表.xlsx"), "缺路径: {full}");
assert!(full.contains("read_file"), "缺读取引导: {full}");
assert!(full.contains("如实告知用户"), "缺读取局限交底: {full}");
// 落盘前占位版:不含路径(不喂模型不落卡,仅供空判与命令/审批判定)
let pending = file_inbound_prompt("报表.xlsx", None);
assert!(!pending.contains("本地路径"), "占位版不该有路径: {pending}");
assert!(pending.contains("「报表.xlsx」"));
}
#[test]
fn file_size_limit_rejects_over_20mb_only() {
// 恰 20MB 放行(> 判定),超限给可读错误话术(含实际大小,引导压缩/拆分)
assert_eq!(file_size_over_limit_error(20 * 1024 * 1024), None);
assert_eq!(file_size_over_limit_error(1024), None);
let err = file_size_over_limit_error(21 * 1024 * 1024).unwrap();
assert!(err.contains("20MB") && err.contains("21MB"), "话术缺限值/实际值: {err}");
}
#[test]
fn file_display_name_prefers_original_and_hash_fallback() {
// 原始名保真(清洗首尾空白);缺失回退 file_<hash 前 8>,不同内容回退名不同
assert_eq!(file_display_name(" 报表.xlsx ", b"abc"), "报表.xlsx");
let fallback = file_display_name("", b"abc");
assert!(fallback.starts_with("file_"), "回退名形态: {fallback}");
assert_eq!(fallback.len(), "file_".len() + 8, "回退名取 hash 前 8: {fallback}");
assert_ne!(
fallback,
file_display_name(" ", b"abcd"),
"不同内容的回退名应可区分"
);
}
#[test]
fn command_case_and_tail_tolerance() {
// 30:/new 大小写与附言容错,附言不烧 LLM
assert_eq!(parse_command("/new"), Some(("new".to_string(), String::new())));
assert_eq!(
parse_command("/New 帮我开新会话"),
Some(("new".to_string(), "帮我开新会话".to_string()))
);
assert_eq!(parse_command("/STATUS"), Some(("status".to_string(), String::new())));
}
#[test]
fn command_fullwidth_normalized() {
// 30:半全角混排,全角斜杠/字母/空格归一后识别
assert_eq!(parse_command("/new"), Some(("new".to_string(), String::new())));
assert_eq!(
parse_command("/STATUS 附加"),
Some(("status".to_string(), "附加".to_string()))
);
}
#[test]
fn command_none_and_unknown_fallthrough() {
// 纯任务文本与裸斜杠不是命令
assert!(parse_command("帮我走查代码").is_none());
assert!(parse_command("/").is_none());
// 未知命令词解析得出,由调用方落回常规任务路径(零行为变更);无空格路径整体为命令词
assert_eq!(
parse_command("/usr/bin 定位").map(|(c, _)| c),
Some("usr/bin".to_string())
);
}
#[test]
fn mention_cn_no_space_not_swallowed() {
// 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"), "");
}
// ── 过程流增量摘要提炼(todo.md 13):头部行格式契约 ──
#[test]
fn preview_first_text_no_head() {
// 首轮纯文本(无轮次无工具):不显头行,原文+光标
let out = stream_preview("你好", None, None, None, 0, None, None);
assert_eq!(out, "你好\n▌");
}
#[test]
fn preview_tool_running_head() {
// 工具执行中:头行 = 轮次 + ⚙ 工具名 执行中
let out = stream_preview("分析中", Some(3), Some("read_file"), None, 5, None, None);
assert!(out.starts_with("**第 3 轮 · ⚙ `read_file` 执行中**\n\n"));
assert!(out.contains("分析中"));
assert!(out.ends_with("▌"));
}
#[test]
fn preview_tool_done_head() {
// 工具轮间空档(当前无工具):头行转「已完成 N 步」——纯工具轮的"活着"证据
let out = stream_preview("", Some(4), None, None, 7, None, None);
assert!(out.starts_with("**第 4 轮 · 已完成 7 步工具调用**\n\n"));
}
#[test]
fn preview_tail_keep_on_overflow() {
// 超上限保尾截断 + 前缀标注;头行不受截断影响
let long = "x".repeat(9000);
let out = stream_preview(&long, Some(1), None, None, 1, None, None);
assert!(out.starts_with("**第 1 轮 · 已完成 1 步工具调用**"));
assert!(out.contains("…(前文已略)"));
assert!(out.chars().count() < 8500); // 头行 + 8000 尾 + 标注
}
#[test]
fn preview_heartbeat_line_on_silent() {
// 13a-2 静默超阈值:心跳行置顶(在常规头行之前),含轮次/动作/静默秒数
let out = stream_preview("分析中", Some(3), Some("run_command"), None, 5, Some(67), None);
assert!(out.starts_with("💓 **仍在跑:第 3 轮 · `run_command` 执行中 · 静默 67s**\n\n"));
assert!(out.contains("**第 3 轮 · ⚙ `run_command` 执行中**")); // 常规头行保留
}
#[test]
fn preview_heartbeat_tool_done_fallback() {
// 13a-2 静默且工具间空档:动作优先用最近工具名,无则回退步数表述
let out = stream_preview("", Some(4), None, Some("grep"), 9, Some(95), None);
assert!(out.starts_with("💓 **仍在跑:第 4 轮 · `grep` 执行中 · 静默 95s**\n\n"));
// 无最近工具(纯 LLM 长思考)回退
let out2 = stream_preview("", Some(4), None, None, 9, Some(95), None);
assert!(out2.starts_with("💓 **仍在跑:第 4 轮 · 最近动作 `9 步工具调用` · 静默 95s**\n\n"));
}
#[test]
fn preview_heartbeat_no_round_no_steps() {
// 13a-2 静默且零步零轮(纯 LLM 长思考):轮次回退「启动」,动作回退「LLM 思考中」
let out = stream_preview("开头文本", None, None, None, 0, Some(80), None);
assert!(out.starts_with("💓 **仍在跑:启动 · 最近动作 `LLM 思考中` · 静默 80s**\n\n"));
assert!(out.contains("开头文本\n▌"));
}
#[test]
fn preview_progress_fallback_round_when_no_agent_round() {
// 13a-3:AiAgentRound 缺席(纯进度事件场景)时,头行回退 AiTaskProgress 快照
// 并附 max 轮与累计 token;max=0(不限轮)不显上限
let out = stream_preview("干活中", None, None, None, 2, None, Some((10, Some(50), Some(123456))));
assert!(out.starts_with("**第 10 轮/50 轮 · 累计 123456 tokens · 已完成 2 步工具调用**"));
// max_rounds=0 = 不限轮次,只显 token
let out2 = stream_preview("干活中", None, None, None, 2, None, Some((10, Some(0), Some(999))));
assert!(out2.starts_with("**第 10 轮 · 累计 999 tokens · 已完成 2 步工具调用**"));
}
#[test]
fn preview_heartbeat_uses_progress_round() {
// 13a-3×13a-2 组合:静默心跳的轮次也吃 AiTaskProgress 快照
let out = stream_preview("", None, None, None, 3, Some(72), Some((15, None, Some(42))));
assert!(out.starts_with("💓 **仍在跑:第 15 轮 · 最近动作 `3 步工具调用` · 静默 72s**\n\n"));
}
#[test]
fn heartbeat_due_threshold_and_bucket_dedup() {
// 13a-2 判定纯函数:未达阈值不刷;达阈值且新桶刷;同静默分钟桶去重;进新桶再刷
assert_eq!(heartbeat_due(59, None, 60), None);
assert_eq!(heartbeat_due(60, None, 60), Some(1));
assert_eq!(heartbeat_due(60, Some(1), 60), None);
assert_eq!(heartbeat_due(119, Some(1), 60), None); // 119/60=1 同桶
assert_eq!(heartbeat_due(120, Some(1), 60), Some(2));
}
// ── 38a 出站 md 净化 ──
#[test]
fn md_heading_to_bold_line() {
assert_eq!(md_to_lark_friendly("## 分析报告"), "**分析报告**");
assert_eq!(md_to_lark_friendly("# 总览\n正文"), "**总览**\n正文");
// 非 CommonMark 形态(# 后无空格)不误伤
assert_eq!(md_to_lark_friendly("#hashtag 保留"), "#hashtag 保留");
}
#[test]
fn md_table_to_list_rows() {
let md = "| 名称 | 数量 |\n|---|---|\n| 甲 | 1 |";
let out = md_to_lark_friendly(md);
assert_eq!(out, "- **名称**:**数量**\n- 甲:1");
}
#[test]
fn md_fence_kept_verbatim() {
// 围栏内 # 与表格语法原样保留(转换反伤代码可读性),围栏外正常净化
let md = "## 说明\n```rust\n# comment\n| a | b |\n```\n| x | y |\n|---|---|\n| 1 | 2 |";
let out = md_to_lark_friendly(md);
assert!(out.starts_with("**说明**"));
assert!(out.contains("```rust\n# comment\n| a | b |\n```"));
assert!(out.contains("- **x**:**y**"));
}
#[test]
fn md_plain_text_untouched_and_idempotent() {
let md = "普通段落 **粗体** 保留\n- 列表项";
let once = md_to_lark_friendly(md);
assert_eq!(once, md);
assert_eq!(md_to_lark_friendly(&once), once);
}
// ── 22 受理时间戳过滤 ──
#[test]
fn acceptance_window_filters_old_messages() {
// 有受理时刻:只认其后落库;边界时刻本身不算;缺时间戳的老消息视为不在边界内
assert!(msg_after_acceptance(Some(2000), Some(1000)));
assert!(!msg_after_acceptance(Some(1000), Some(1000)));
assert!(!msg_after_acceptance(Some(500), Some(1000)));
assert!(!msg_after_acceptance(None, Some(1000)));
// 无受理时刻不过滤(旧语义,调用方自担边界)
assert!(msg_after_acceptance(Some(1), None));
assert!(msg_after_acceptance(None, None));
}
// ── 34 推送终态判据 ──
#[test]
fn terminal_push_prefix_only() {
assert!(is_terminal_push("✅ 完成"));
assert!(is_terminal_push("❌ 失败:原因"));
assert!(is_terminal_push(" ✅ 缩进也算"));
assert!(!is_terminal_push("⏳ 进行中"));
assert!(!is_terminal_push("普通进度行"));
}
// ── 防抖合并 ──
#[test]
fn debounce_text_merges_with_blank_line() {
// 多文本空行拼接;空侧直取另一侧不产首尾空行;追加路径逐条 append 的链式形态
assert_eq!(merge_debounce_text("第一段", "第二段"), "第一段\n\n第二段");
assert_eq!(merge_debounce_text("", "第二段"), "第二段");
assert_eq!(merge_debounce_text("第一段", ""), "第一段");
let mut merged = merge_debounce_text("a", "b");
merged = merge_debounce_text(&merged, "c");
assert_eq!(merged, "a\n\nb\n\nc");
}
#[test]
fn debounce_parts_merge_in_arrival_order() {
// 图片片按消息到达序合并;单侧 None 直取另一侧;双 None 保持 None
let img1 = ContentPart::image_url("https://x/1.png");
let img2 = ContentPart::image_url("https://x/2.png");
let merged = merge_debounce_parts(
Some(vec![ContentPart::text("图1说明"), img1.clone()]),
Some(vec![img2.clone()]),
);
assert_eq!(merged, Some(vec![ContentPart::text("图1说明"), img1, img2.clone()]));
assert_eq!(
merge_debounce_parts(None, Some(vec![img2.clone()])),
Some(vec![img2.clone()])
);
assert_eq!(merge_debounce_parts(Some(vec![]), None), Some(vec![]));
assert_eq!(merge_debounce_parts(None, None), None);
}
#[test]
fn debounce_entry_counts_appends() {
// N 计数只含追加(首条不入 entry):3 条追加=合并 4 条受理;文本与图片同步蓄积
let mut entry = DebounceEntry::default();
assert_eq!(entry.appended, 0);
entry.append("第二条", None);
entry.append("第三条", Some(vec![ContentPart::image_url("https://x/2.png")]));
entry.append("第四条", None);
assert_eq!(entry.appended, 3);
assert_eq!(entry.text, "第二条\n\n第三条\n\n第四条");
assert_eq!(
entry.parts,
Some(vec![ContentPart::image_url("https://x/2.png")])
);
}
}