修复: 飞书走查六项 + 停靠状态写侧根因解 + agentic/audit 加固
飞书集成走查六项(feishu_channel.rs +420):
1. 审批卡按钮化: on_card_action + handle_card_action(解析 {approval_id,approved}
→ ai_approve) + send_approval_button_card(Card JSON 1.0 双按钮, 失败降级文本卡)
2. 首帧流式提速: AiTextDelta 首个 delta 立即刷占位卡, 不等 3s ticker
3. 断线重连通知: 按 run_loop 时长判据(≥30s 推断开 / <30s 累计 3 次推不稳定),
新增 notify_monitor()
4. 占位卡带任务原文回显(truncate 60)
5. 白名单配置引导: 后端分步自助引导 + FeishuPanel hintFeishuWhitelist + i18n
6. 审批决议后追加执行结果: spawn_approval_result_watcher 订阅 AiToolCallCompleted
(匹配 tool_call_id, 60s 超时兜底); 订阅须在 ai_approve 前完成(无订阅者跳过序列化)
停靠(parked)写侧根因解(904):
emit_task_parked(ai/mod.rs) 升级为「写 parked + 发事件 + 通知」三件套——
新增 set_loop_state(conv_id, Some("parked")) fire-and-forget; 8 个业务调用点
(H2 三闸/熔断两路/挂起取消/max_rounds/midstream/fatal + 审批超时) 全经单点。
治「停靠卡继续必报 AI 未在暂停态」(全仓原无写 parked 路径)。
agentic/audit 加固: context_lifecycle/stream_relay/mod 调整 + audit record 完善 +
tool_failure_memory/tool_result_offload/tools/task/settings 同步。
验收: cargo check -p devflow EXIT 0。
This commit is contained in:
+399
-21
@@ -13,7 +13,9 @@ 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::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;
|
||||
@@ -68,11 +70,36 @@ struct FeishuConfig {
|
||||
app_id: String,
|
||||
#[serde(default)]
|
||||
app_secret: String,
|
||||
/// 空 = 放行所有(首启抓 open_id 回填);open_id 按应用隔离,严禁跨应用搬运
|
||||
#[serde(default)]
|
||||
/// 空 = 放行所有(首启抓 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())
|
||||
@@ -161,13 +188,38 @@ pub fn start_if_enabled(app: tauri::AppHandle) {
|
||||
// 与 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 {
|
||||
if let Err(e) = run_loop(ctx.clone(), leaked).await {
|
||||
tracing::error!("[feishu] 长连接退出,{}s 后重连: {e}", backoff);
|
||||
feishu_log(&format!("长连接断开,{}s 后重连: {e}", backoff));
|
||||
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 {
|
||||
tracing::info!("[feishu] 长连接正常退出,{}s 后重连", backoff);
|
||||
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);
|
||||
@@ -175,10 +227,24 @@ pub fn start_if_enabled(app: tauri::AppHandle) {
|
||||
});
|
||||
}
|
||||
|
||||
/// 运维通知:推给 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_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 再刷新)
|
||||
@@ -203,6 +269,15 @@ async fn run_loop(ctx: Arc<Ctx>, client: &'static LarkClient) -> anyhow::Result<
|
||||
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
|
||||
@@ -304,11 +379,12 @@ async fn handle_message(
|
||||
// D3:任何入站消息都是事件管道存活证据,先于一切过滤更新活性时间戳
|
||||
ctx.last_alive.store(now_secs(), Ordering::SeqCst);
|
||||
if msg.message_type != "text" {
|
||||
// #25 非文本消息静默丢修复:回提示卡(D7 媒体能力前的兜底),不再零反馈
|
||||
// #25 非文本消息静默丢修复:回提示卡(D7 媒体能力前的兜底),不再零反馈。
|
||||
// 补替代路径引导:用户发图/文件往往是想让 AI 处理内容,直接告知改用文字描述
|
||||
let _ = reply_markdown_card(
|
||||
&ctx.client,
|
||||
&msg.message_id,
|
||||
"📎 暂仅支持文本消息,图片/文件等媒体能力规划中",
|
||||
"📎 暂仅支持文本消息,图片/文件等媒体能力规划中。\n\n如需处理图片/文件内容,请用文字描述你的需求(如「把这段文字翻译成英文:…」)。",
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
@@ -340,10 +416,14 @@ async fn handle_message(
|
||||
|
||||
// 白名单(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!("⛔ 你不在白名单内。你的 open_id:`{open_id}`(发给管理员回填后重试)");
|
||||
reply_message(&ctx.client, &message_id, &chat_id, &hint).await;
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -452,14 +532,39 @@ async fn handle_message(
|
||||
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, approved).await
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -489,8 +594,12 @@ async fn handle_message(
|
||||
feishu_log(&format!("收到消息 chat={chat_id} open_id={open_id} text={}", truncate(&text, 80)));
|
||||
// 立即确认(交互丝滑):占位改用单行卡片并记下其 message_id,完成后原地更新为结果,
|
||||
// 占位与结果合一不刷屏;占位 reply 失败则置空,完成路径回退现状(新发卡片)
|
||||
// 2026-10-08 占位卡带任务原文回显:原固定「🤔 已接收,处理中…」用户不知在做什么,
|
||||
// 到受理卡之间(per-chat 锁排队 + spawn)有一段无内容可对证期;回显摘要让用户即时确认
|
||||
// 消息已被正确接收。text 此后会被 move 进 spawn_agentic_session,此处先取摘要
|
||||
let placeholder_text = format!("🤔 已接收,处理中…\n任务:{}", truncate(&text, 60));
|
||||
let placeholder_id =
|
||||
reply_markdown_card(&ctx.client, &message_id, "🤔 已接收,处理中…").await;
|
||||
reply_markdown_card(&ctx.client, &message_id, &placeholder_text).await;
|
||||
// per-chat 串行:同会话消息排队(防通知桥单槽被后条覆盖致前条串扰)
|
||||
let chat_lock = {
|
||||
let mut locks = ctx.chat_locks.lock().await;
|
||||
@@ -776,12 +885,22 @@ async fn run_approval_notifier(ctx: Arc<Ctx>) {
|
||||
.unwrap_or("")
|
||||
.replace('\n', " ");
|
||||
let brief = truncate(&reason, 80);
|
||||
let card = if brief.is_empty() {
|
||||
format!("⏸ 需审批: `{name}`。回复「批准」或「拒绝」继续")
|
||||
// 审批卡按钮化(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 {
|
||||
format!("⏸ 需审批: `{name}` {brief}。回复「批准」或「拒绝」继续")
|
||||
};
|
||||
send_markdown_card(&ctx.client, "", &chat_id, &card).await;
|
||||
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,
|
||||
@@ -790,6 +909,229 @@ async fn run_approval_notifier(ctx: Arc<Ctx>) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 卡片按钮回调(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(
|
||||
@@ -810,6 +1152,31 @@ async fn latest_pending_approval(
|
||||
.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> {
|
||||
@@ -900,6 +1267,18 @@ async fn run_stream_refresher(
|
||||
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——短响应(简单问答/翻译)原需等满一个节流周期
|
||||
// 才见首字,感知停顿明显。仅首帧走即时刷(acc_chars 首次>painted),
|
||||
// 后续仍按 3s 节流守飞书 QPS 配额
|
||||
if 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 刷卡(纯工具轮无文本也有"活着"信号)
|
||||
@@ -1664,7 +2043,6 @@ mod tests {
|
||||
assert!(out2.starts_with("💓 **仍在跑:第 4 轮 · 最近动作 `9 步工具调用` · 静默 95s**\n\n"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[test]
|
||||
fn preview_heartbeat_no_round_no_steps() {
|
||||
// 13a-2 静默且零步零轮(纯 LLM 长思考):轮次回退「启动」,动作回退「LLM 思考中」
|
||||
|
||||
Reference in New Issue
Block a user