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

文档: 无人值守专项设计档+todo 销账(#10/#11/#12), Web 规范合规批(App/视图/组件样式)

Browse Source
This commit is contained in:
lxy
2026-10-06 12:13:39 +08:00
parent 04db8c0d3b
commit 0dfb50be9b
29 changed files with 419 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/feishu_channel.rs
+221 -37
View File
@@ -273,6 +273,27 @@ async fn handle_message(
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 {
if 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)));
// 立即确认(交互丝滑):占位改用单行卡片并记下其 message_id,完成后原地更新为结果,
@@ -295,7 +316,11 @@ async fn handle_message(
ctx.inflight.fetch_add(1, Ordering::SeqCst);
let _inflight_guard = InflightGuard(ctx.inflight.clone());
let state = ctx.app.state::<crate::state::AppState>();
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
// 通知桥注册(#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);
// 流式预览:订阅既有 ai_event_bus 拿本 conv 的 AiTextDelta 增量(emit_ai_chat_event
@@ -314,6 +339,7 @@ async fn handle_message(
deliver_rx,
));
let task_brief = truncate(&text, 60); // move 前先取展示摘要(spawn 会 move text)
match crate::commands::ai::commands::chat::spawn_agentic_session(
&ctx.app,
&state,
@@ -353,30 +379,99 @@ async fn handle_message(
}
}
// 等最终回复(通知桥 oneshot;超时 30 分钟,任务仍在后台跑不丢失)
let reply = match tokio::time::timeout(std::time::Duration::from_secs(1800), rx).await {
Ok(Ok(text)) if !text.trim().is_empty() => text,
// A 韧性兜底:通知桥带回空内容 → 回查 DB 该会话最后一条 assistant(事实源),有则照发
_ => {
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let fallback = last_assistant_reply(&db, &conv_id).await;
feishu_log(&format!("conv={conv_id} 通知桥空/超时,DB回查={}", truncate(fallback.as_deref().unwrap_or("无"), 60)));
match fallback {
Some(text) => text,
None => "⚠️ 本轮没有生成文本回复(已记录排查)。任务若仍在进行可稍后 /new 重置或重发。".to_string(),
// ── #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 占位卡为结果。
// 卡片所有权移交(走查🔴#1 修正):占位卡自此只被 watcher 任务写——受理卡与终版
// 都在 watcher 单任务内按序落卡(受理卡先写,快任务兜底命中时跳过受理卡直投终版),
// 彻底消除「受理卡与终版两个 HTTP 请求网络乱序,终版被受理卡覆盖丢失」的竞态窗口。
drop(deliver_tx);
let _ = stream_task.await; // 等流式任务退出(快,只走通道关闭分支)
let ack = format!("✅ 已受理,后台执行中\n任务:{task_brief}");
feishu_log(&format!("受理返回 chat={chat_id} conv={conv_id}"));
// 完成订阅任务:独立于消息处理生命周期,直到 AiAgentFinished 广播到达投递终版。
// (spawn Err 路径在上方已 return,永不到达此处——watcher 依赖引擎层正常完成广播)
tauri::async_runtime::spawn(async move {
run_completion_watcher(ctx, conv_id, placeholder_id, message_id, chat_id, ack).await;
});
}
/// 完成事件订阅任务(#14 两段式):占位卡唯一写者。入场先投受理卡,再消费
/// ai_event_bus 的 AiAgentFinished(本 conv)原地 edit_card 为终版。引擎层
/// deliver_completion_report_if_waited 是广播单一发出点,watcher 纯消费无状态。
///
/// 快任务竞态兜底(走查🟡#1 加门槛):广播可能先于订阅(秒级任务)——立即回查
/// 一次,但须同时满足 conv 非活跃 + 回查时间晚于受理(防 spawn 返回但 loop 尚未
/// 置 active 的时序窗口误判) + 文本非空,三者齐才直投终版跳过受理卡(卡片写入
/// 单任务串行,无网络乱序覆盖)。任一不满足则先落受理卡再进事件等待。
///
/// 超时语义(走查🟡#2 修正):35min(> loop 30min 上限)超时投降级提示后**不退出**,
/// 继续挂事件等待——任务后台不丢,晚到的完成事件仍会送到卡片;watcher 与任务
/// 同生命周期,彻底无果(引擎侧异常未广播)才在二次超时后弃守。
async fn run_completion_watcher(
ctx: Arc<Ctx>,
conv_id: String,
placeholder_id: Option<String>,
message_id: String,
chat_id: String,
ack: String,
) {
let state = ctx.app.state::<crate::state::AppState>();
let mut rx = state.ai_event_bus.subscribe();
let accepted_at = tokio::time::Instant::now();
// 快任务兜底:立即回查一次(事件可能已错过);三分量门槛防时序窗口误判
let settled = !state.conv_states.is_active(&conv_id)
&& accepted_at.elapsed() >= std::time::Duration::from_secs(3)
&& matches!(last_assistant_reply(&state.db, &conv_id).await,
Some(t) if !t.trim().is_empty());
if settled {
let text = last_assistant_reply(&state.db, &conv_id)
.await
.unwrap_or_default();
feishu_log(&format!("watcher 快任务兜底 conv={conv_id} len={}", text.len()));
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await;
return;
}
// 落受理卡(占位缺失回退新发,deliver_result 内部链路)
deliver_result(&ctx.client, &placeholder_id.clone(), &message_id.clone(), &chat_id.clone(), &ack).await;
// 事件等待循环:首超时 35 分钟投降级提示不退出,继续等晚到事件;二次超时弃守。
let timeout = std::time::Duration::from_secs(2100);
loop {
let text = tokio::select! {
item = rx.recv() => match item {
Ok(v) => {
if v.get("type").and_then(|t| t.as_str()) != Some("AiAgentFinished") {
continue;
}
if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) {
continue;
}
// 广播语义:agent_id==conv_id 时 summary 即本会话最终报告(引擎层构造)
v.get("summary").and_then(|s| s.as_str()).unwrap_or("").to_string()
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
},
_ = tokio::time::sleep(timeout) => {
// 首超时:DB 回查兜底(任务可能刚完成事件已发完);回查有果照投终版,
// 无果投降级提示并**继续等待**(走查🟡#2:超时不弃投递,晚到事件仍送达)
match last_assistant_reply(&state.db, &conv_id).await {
Some(t) if !t.trim().is_empty() => t,
_ => {
let notice = "⏰ 任务执行超 35 分钟仍未返回(后台仍在运行,完成后本卡片自动更新;或本会话 /status 查询)。".to_string();
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &notice).await;
continue;
}
}
}
}
};
feishu_log(&format!("回复 chat={chat_id} len={}", reply.len()));
// 终版投递:交流式任务单点执行(内部复用 deliver_result 原回退链);
// 任务已死(通道已关或异常退出)则兜底直投,此时无并发写卡方,回退链行为不变;
// 若任务在投递后异常退出,重复一次终版 edit_card 内容相同,无副作用
let delivered = match deliver_tx.send(reply.clone()) {
Ok(()) => matches!(stream_task.await, Ok(())),
Err(_) => false,
};
if !delivered {
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &reply).await;
};
feishu_log(&format!("watcher 终版 conv={conv_id} len={}", text.len()));
deliver_result(&ctx.client, &placeholder_id, &message_id, &chat_id, &text).await;
return;
}
}
@@ -423,6 +518,13 @@ async fn run_stream_refresher(
// 累计字符数 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;
let mut tool_steps: usize = 0usize;
let mut painted_sig = String::new();
let mut ticker = tokio::time::interval(STREAM_FLUSH_INTERVAL);
// 刷新落后只保最新帧不补帧:预览语义是最新输出才有价值
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -438,16 +540,37 @@ async fn run_stream_refresher(
}
item = delta_rx.recv() => match item {
Ok(v) => {
// 双过滤:只要本 conv 的流式增量;type tag 即变体名(AiChatEvent 无 serde rename)
if v.get("type").and_then(|t| t.as_str()) != Some("AiTextDelta") {
continue;
}
// conv 路由过滤(全事件共用);type tag 即变体名(AiChatEvent 无 serde rename)
if v.get("conversation_id").and_then(|c| c.as_str()) != Some(conv_id.as_str()) {
continue;
}
if let Some(delta) = v.get("delta").and_then(|d| d.as_str()) {
acc.push_str(delta);
acc_chars += delta.chars().count();
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();
}
}
// 轮次推进:立即标脏强制下一 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());
tool_steps += 1;
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
}
Some("AiToolCallCompleted") => {
current_tool = None;
painted_chars = painted_chars.saturating_sub(1); // 标脏
}
_ => {}
}
}
// 预览允许丢帧:终版事实源在通知桥 oneshot,消费落后只降级预览不断链路
@@ -467,11 +590,18 @@ async fn run_stream_refresher(
let Some(pid) = placeholder_id.as_deref().filter(|id| !id.is_empty()) else {
continue;
};
// 信号标脏:过程事件(轮次/工具)已把 painted_chars 回拨 1,此处字符数不等即刷
if acc_chars == painted_chars {
continue;
}
if update_card(&ctx.client, pid, &stream_preview(&acc)).await {
// 签名去重:正文长度与轨迹行均未变则跳过(防同 tick 内事件风暴空刷)
let sig = format!("{}|{}|{}|{}", acc_chars, round.unwrap_or(0), current_tool.as_deref().unwrap_or(""), tool_steps);
if sig == painted_sig {
continue;
}
if update_card(&ctx.client, pid, &stream_preview(&acc, round, current_tool.as_deref(), tool_steps)).await {
painted_chars = acc_chars;
painted_sig = sig;
}
// 刷失败静默跳过,下个 tick 自然重试
}
@@ -479,9 +609,28 @@ async fn run_stream_refresher(
}
}
/// 流式中间态文本:尾部加光标符示意仍在生成;超上限从头截断保尾部,前缀标注省略事实
/// 防读者误把中间态当全文
fn stream_preview(acc: &str) -> String {
/// 流式中间态卡片:头部一行过程摘要(轮次/工具轨迹提炼)+ 正文流式增量(保尾截断)。
///
/// 提炼原则(todo.md 13「决策价值过滤」):历史步骤折叠为计数,只透出「当前在干什么」——
/// 读者要的是"活着没卡死+进行到哪",不是 20 个工具名的流水墙;正文仍以流式文本为主,
/// 纯工具轮(无文本)时头部行是唯一进度证据。
fn stream_preview(
acc: &str,
round: Option<u32>,
current_tool: Option<&str>,
tool_steps: usize,
) -> String {
// 头部摘要行:轮次 + 当前工具(执行中)/已执行步数;全部缺省(首轮文本前)不显头行
let mut head = String::new();
if round.is_some() || tool_steps > 0 {
let r = round.map(|n| format!("第 {n} 轮")).unwrap_or_default();
let t = match current_tool {
Some(name) => format!("⚙ `{name}` 执行中"),
None => format!("已完成 {tool_steps} 步工具调用"),
};
head = format!("**{r} · {t}**\n\n");
}
// 正文:超上限保尾截断(对齐原口径)
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();
@@ -489,7 +638,7 @@ fn stream_preview(acc: &str) -> String {
} else {
acc.to_string()
};
format!("{body}\n▌")
format!("{head}{body}\n▌")
}
/// markdown 回复:interactive 卡片承载(lark_md 原生渲染标题/列表/粗体,text 类型会符号裸露)
@@ -759,7 +908,7 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap<String, Stri
#[cfg(test)]
mod tests {
use super::strip_mention_placeholders;
use super::{strip_mention_placeholders, stream_preview};
#[test]
fn mention_cn_no_space_not_swallowed() {
@@ -782,4 +931,39 @@ mod tests {
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, 0);
assert_eq!(out, "你好\n▌");
}
#[test]
fn preview_tool_running_head() {
// 工具执行中:头行 = 轮次 + ⚙ 工具名 执行中
let out = stream_preview("分析中", Some(3), Some("read_file"), 5);
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, 7);
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, 1);
assert!(out.starts_with("**第 1 轮 · 已完成 1 步工具调用**"));
assert!(out.contains("…(前文已略)"));
assert!(out.chars().count() < 8500); // 头行 + 8000 尾 + 标注
}
}