fix(feishu,channel): 修正 reaction 语义并增容 bus 队列,避免丢最后一条消息
- feishu.rs: 仅终态消息(AssistantResponse | ErrorNotification)成功发送 后才移除 reaction,ToolCall 等中间过程不再触碰;发送失败/空内容 保留 reaction 作为异常信号,解决『reaction 被提前移除掩盖最终响应 丢失』的语义错位问题 - manager.rs: MessageBus 容量从 100 扩至 256,降低多 topic 并发场景下 bus 级 try_send 的瞬时积压丢弃概率 与前一提交的 outbound_dispatcher 双队列优先级重构配合:AssistantResponse / ErrorNotification 走独立 high 队列必达,终态消息送达后才移除 reaction, 彻底解决『飞书经常丢最后一条消息且 reaction 状态误导』的问题。
This commit is contained in:
parent
e037de88a2
commit
fe7037e4ad
@ -2504,6 +2504,15 @@ impl Channel for FeishuChannel {
|
|||||||
"open_id"
|
"open_id"
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// reaction 语义:仅"终态消息"成功发送后才移除 reaction。
|
||||||
|
// 终态 = AssistantResponse(最终响应)或 ErrorNotification(agent 异常终止的错误通知)。
|
||||||
|
// ToolCall 等中间过程事件不触碰 reaction——让 reaction 真实反映用户是否已收到
|
||||||
|
// agent 的最终产出(reaction 持续 = 尚未收到响应或错误;reaction 消失 = 已收到)。
|
||||||
|
// 发送失败时保留 reaction 作为异常信号;空内容不视为送达,保留 reaction。
|
||||||
|
let is_terminal = matches!(
|
||||||
|
msg.event_kind,
|
||||||
|
OutboundEventKind::AssistantResponse | OutboundEventKind::ErrorNotification
|
||||||
|
);
|
||||||
let remove_reaction = async {
|
let remove_reaction = async {
|
||||||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||||||
};
|
};
|
||||||
@ -2514,7 +2523,14 @@ impl Channel for FeishuChannel {
|
|||||||
|
|
||||||
// Empty content
|
// Empty content
|
||||||
if content.is_empty() {
|
if content.is_empty() {
|
||||||
remove_reaction.await;
|
// 空最终响应是异常:保留 reaction 让用户察觉,仅记录 warn。
|
||||||
|
// 非最终响应(如 ToolCall 空内容)本来就不触碰 reaction。
|
||||||
|
if is_terminal {
|
||||||
|
tracing::warn!(
|
||||||
|
chat_id = %msg.chat_id,
|
||||||
|
"Final response has empty content, keeping reaction as anomaly signal"
|
||||||
|
);
|
||||||
|
}
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -2526,7 +2542,10 @@ impl Channel for FeishuChannel {
|
|||||||
let result = self
|
let result = self
|
||||||
.dispatch_send(receive_id, receive_id_type, "text", content, reply_to)
|
.dispatch_send(receive_id, receive_id_type, "text", content, reply_to)
|
||||||
.await;
|
.await;
|
||||||
remove_reaction.await;
|
// 仅最终响应且发送成功才移除 reaction
|
||||||
|
if is_terminal && result.is_ok() {
|
||||||
|
remove_reaction.await;
|
||||||
|
}
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
MsgFormat::Post => {
|
MsgFormat::Post => {
|
||||||
@ -2535,7 +2554,9 @@ impl Channel for FeishuChannel {
|
|||||||
let result = self
|
let result = self
|
||||||
.dispatch_send(receive_id, receive_id_type, "post", &post_body, reply_to)
|
.dispatch_send(receive_id, receive_id_type, "post", &post_body, reply_to)
|
||||||
.await;
|
.await;
|
||||||
remove_reaction.await;
|
if is_terminal && result.is_ok() {
|
||||||
|
remove_reaction.await;
|
||||||
|
}
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
MsgFormat::Interactive => {
|
MsgFormat::Interactive => {
|
||||||
@ -2561,11 +2582,17 @@ impl Channel for FeishuChannel {
|
|||||||
reply_to,
|
reply_to,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
remove_reaction.await;
|
// 回退成功才移除 reaction;回退失败则保留作为异常信号
|
||||||
|
if is_terminal && result.is_ok() {
|
||||||
|
remove_reaction.await;
|
||||||
|
}
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
remove_reaction.await;
|
// 所有 chunk 成功:仅最终响应移除 reaction
|
||||||
|
if is_terminal {
|
||||||
|
remove_reaction.await;
|
||||||
|
}
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -2618,19 +2645,23 @@ impl Channel for FeishuChannel {
|
|||||||
Ok(()) => sent_media += 1,
|
Ok(()) => sent_media += 1,
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
tracing::warn!(error = %error, path = %path, media_type = %media_item.media_type, "Failed to send media message to Feishu");
|
tracing::warn!(error = %error, path = %path, media_type = %media_item.media_type, "Failed to send media message to Feishu");
|
||||||
|
// 媒体失败:保留 reaction 作为异常信号(不调用 remove_reaction)
|
||||||
return Err(error);
|
return Err(error);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if msg.content.trim().is_empty() && sent_media == 0 {
|
if msg.content.trim().is_empty() && sent_media == 0 {
|
||||||
remove_reaction.await;
|
// 无内容无媒体:保留 reaction 作为异常信号
|
||||||
return Err(ChannelError::Other(
|
return Err(ChannelError::Other(
|
||||||
"No supported media items were sent to Feishu".to_string(),
|
"No supported media items were sent to Feishu".to_string(),
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
remove_reaction.await;
|
// 全部成功:仅最终响应移除 reaction
|
||||||
|
if is_terminal {
|
||||||
|
remove_reaction.await;
|
||||||
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -26,7 +26,7 @@ impl ChannelManager {
|
|||||||
|
|
||||||
Self {
|
Self {
|
||||||
channels: Arc::new(RwLock::new(channels)),
|
channels: Arc::new(RwLock::new(channels)),
|
||||||
bus: MessageBus::new(100),
|
bus: MessageBus::new(256),
|
||||||
websocket_channel,
|
websocket_channel,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user