xiaoski/PicoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(feishu): preserve reply and thread semantics

Browse Source
This commit is contained in:
xiaoxixi 2026-07-19 15:52:53 +08:00
parent 0512d91729
commit 7e53fa658b
3 changed files with 254 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -243,7 +243,7 @@ WebUI/TUI 上传文件默认保存到 `~/.picobot/media/cli_chat`，单文件上
| `cli_chat` | Ratatui 终端客户端，通过 WebSocket 连接 Gateway |
| `feishu` | 飞书/Lark 消息、反应、文件上传下载和媒体引用 |
飞书默认只接受 `allow_from` 中的用户，且群聊消息必须明确 @ 机器人（可通过 `channels.feishu.require_mention=false` 关闭）。默认只发送终态结果；设置 `channels.feishu.live_updates=true` 后会创建一张卡片并持续编辑，`live_update_interval_ms` 默认 500ms，运行时限制在 250–5000ms。外部渠道始终不会收到模型 reasoning。
飞书默认只接受 `allow_from` 中的用户，且群聊消息必须明确 @ 机器人（可通过 `channels.feishu.require_mention=false` 关闭）。回复会使用飞书原生引用/话题语义保持在原消息位置。默认只发送终态结果；设置 `channels.feishu.live_updates=true` 后会创建一张卡片并持续编辑，`live_update_interval_ms` 默认 500ms，运行时限制在 250–5000ms。外部渠道始终不会收到模型 reasoning。
### 会话

4
docs/ARCHITECTURE.md
View File

@ -125,7 +125,7 @@ sequenceDiagram
- 每个 session 有一条容量为 32 的队列，同一 session 串行处理，不同 session 的 worker 可并发执行。
- 队列满时明确拒绝新消息，不允许无界积压。
- Slash command 不进入 Agent 队列，由 `SessionManager` 直接执行，因此 `/stop` 等控制操作不会排在长模型调用之后。
- `InboundMessage` 只保存规范化输入：`sender_id`、`received_at`、媒体和一个 `ChannelContext`。核心只解释其中的 `reply_to`；reaction/message ID 等平台字段作为 `private` 不透明传到对应 Turn/普通回复，不能散落为核心层 magic key。持久化的用户消息保留真实接收时间和 `UserInput` 来源，客户端历史投影不暴露来源中的平台用户 ID。
- `InboundMessage` 只保存规范化输入：`sender_id`、`received_at`、媒体和一个 `ChannelContext`。核心只解释其中的 `reply_to`，其语义是本轮出站应回复的当前入站消息；被用户引用的父消息只用于补充模型上下文。reaction/message ID、话题 root/thread 等平台字段作为 `private` 不透明传到对应 Turn/普通回复，不能散落为核心层 magic key。持久化的用户消息保留真实接收时间和 `UserInput` 来源，客户端历史投影不暴露来源中的平台用户 ID。
- Session 为每个 Agent 请求创建一个 `TurnController`。Provider 向 AgentLoop 发 delta，AgentLoop 发结构化 TurnEvent，只有 TurnController 能把事件归约为有序 block 和单调 revision 的完整快照。
- Turn 快照经 Tokio `watch` 发布，语义为 latest-wins；慢客户端或慢渠道跳过中间状态，不反压 Provider。终态明确编码在快照中，不依赖 sender 关闭。
- Agent 本轮消息原子持久化成功后才发布 `Completed`。取消或失败若已有可见正文，则保存为 `cancelled`/`interrupted` partial；只有 reasoning 时不创建 assistant 历史。
@ -140,7 +140,7 @@ sequenceDiagram
- `TurnDeliveryService` 返回可等待的终态句柄；sink 生命周期启动不等于终态已送达。Session 在终态重试最终失败时通过普通出站路径兜底一次。
- `cli_chat` 将同一 `turn_updated` 快照发给 TUI 和 WebUI。客户端只保留当前 session 中 revision 更新的 `active_turn`，终态随后由持久化历史校准。
- 飞书对每个 DATA 帧先在 2 秒硬期限内 ACK，再进行有界分片重组，并把完整事件交给容量 32 的连接内处理队列；媒体下载和引用查询不占用正常的 WebSocket 读循环。队列饱和时当前事件在连接任务中同步处理而不丢弃。连接异常采用有上限的指数退避持续重连，不因累计故障永久停止。
- 飞书在协议解析阶段按 `allow_from` 拒绝未授权用户；群聊默认必须明确 @ 运行时解析出的机器人身份，身份解析失败时安全地忽略群消息。飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。
- 飞书在协议解析阶段按 `allow_from` 拒绝未授权用户；群聊默认必须明确 @ 运行时解析出的机器人身份，身份解析失败时安全地忽略群消息。飞书把当前消息 ID 作为 `reply_to`，并在私有 metadata 中携带 root/thread 信息；Sink 使用原生 reply API 及 `reply_in_thread` 保持客户端引用和话题位置。飞书默认 `FinalOnly`；开启 `live_updates` 后，第一个可见快照创建卡片，后续编辑同一卡片，终态编辑失败则发送完整结果兜底。reaction 清理在 finish、abort 和 Gateway shutdown 中幂等执行。
- DeliveryCoordinator 与 OutboundDispatcher 共享 `(channel, chat_id)` 写锁，避免活动 Turn 终态与独立消息并发写入同一目标。
### 出站投递

399
src/channels/feishu.rs
View File

@ -159,6 +159,10 @@ struct LarkMessage {
#[serde(default)]
parent_id: Option<String>,
#[serde(default)]
root_id: Option<String>,
#[serde(default)]
thread_id: Option<String>,
#[serde(default)]
mentions: Vec<serde_json::Value>,
}
@ -197,6 +201,77 @@ struct ParsedMessage {
/// ID of the message this message is replying to (if any).
/// Used to fetch quoted message content for display.
parent_id: Option<String>,
root_id: Option<String>,
thread_id: Option<String>,
chat_type: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct FeishuSendTarget {
receive_id: String,
receive_id_type: &'static str,
reply_to: Option<String>,
reply_in_thread: bool,
}
impl FeishuSendTarget {
fn from_message(
chat_id: String,
reply_to: Option<String>,
metadata: &HashMap<String, String>,
) -> Self {
let thread_id = metadata
.get("feishu.thread_id")
.filter(|value| !value.is_empty());
if reply_to.is_none()
&& let Some(thread_id) = thread_id
{
return Self {
receive_id: thread_id.clone(),
receive_id_type: "thread_id",
reply_to: None,
reply_in_thread: false,
};
}
let receive_id_type = if chat_id.starts_with("oc_") {
"chat_id"
} else {
"open_id"
};
Self {
receive_id: chat_id,
receive_id_type,
reply_to,
reply_in_thread: thread_id.is_some(),
}
}
fn request(&self, msg_type: &str, content: &str) -> (String, serde_json::Value) {
let uuid = uuid::Uuid::new_v4().to_string();
if let Some(message_id) = &self.reply_to {
return (
format!("{}/im/v1/messages/{message_id}/reply", FEISHU_API_BASE),
serde_json::json!({
"msg_type": msg_type,
"content": content,
"reply_in_thread": self.reply_in_thread,
"uuid": uuid
}),
);
}
(
format!(
"{}/im/v1/messages?receive_id_type={}",
FEISHU_API_BASE, self.receive_id_type
),
serde_json::json!({
"receive_id": self.receive_id,
"msg_type": msg_type,
"content": content,
"uuid": uuid
}),
)
}
}
impl FeishuChannel {
@ -733,7 +808,10 @@ impl FeishuChannel {
/// Add a reaction emoji to a message and store the reaction_id for later removal.
/// Returns the reaction_id if successful, None otherwise.
async fn add_reaction(&self, message_id: &str) -> Result<Option<String>, ChannelError> {
let emoji = self.config.reaction_emoji.as_str();
let emoji = self.config.reaction_emoji.trim();
if emoji.is_empty() {
return Ok(None);
}
let token = self.get_tenant_access_token().await?;
let resp = self
@ -923,7 +1001,10 @@ impl FeishuChannel {
}
let text = if text.len() > Self::REPLY_CONTEXT_MAX_LEN {
format!("{}...", &text[..Self::REPLY_CONTEXT_MAX_LEN])
format!(
"{}...",
&text[..text.floor_char_boundary(Self::REPLY_CONTEXT_MAX_LEN)]
)
} else {
text
};
@ -931,30 +1012,22 @@ impl FeishuChannel {
Some(format!("[Reply to: {}]", text))
}
/// Send a message to Feishu chat with specified message type and content.
/// Content is passed as-is (already a JSON string for file/media, or plain text for fallback).
async fn send_message_to_feishu(
/// Deliver one native Feishu message using create or reply semantics.
async fn send_raw_message(
&self,
receive_id: &str,
receive_id_type: &str,
target: &FeishuSendTarget,
msg_type: &str,
content: &str,
) -> Result<(), ChannelError> {
) -> Result<String, ChannelError> {
let token = self.get_tenant_access_token().await?;
let (url, body) = target.request(msg_type, content);
let resp = self
.http_client
.post(format!(
"{}/im/v1/messages?receive_id_type={}",
FEISHU_API_BASE, receive_id_type
))
.post(url)
.header("Content-Type", "application/json")
.header("Authorization", format!("Bearer {}", token))
.json(&serde_json::json!({
"receive_id": receive_id,
"msg_type": msg_type,
"content": content
}))
.json(&body)
.send()
.await
.map_err(|e| {
@ -965,6 +1038,11 @@ impl FeishuChannel {
struct SendResp {
code: i32,
msg: String,
data: Option<SendData>,
}
#[derive(Deserialize)]
struct SendData {
message_id: String,
}
let send_resp: SendResp = resp
@ -972,6 +1050,42 @@ impl FeishuChannel {
.await
.map_err(|e| ChannelError::Other(format!("Parse send response error: {}", e)))?;
if matches!(send_resp.code, 230011 | 231003)
&& target.reply_to.is_some()
&& !target.reply_in_thread
{
tracing::warn!(
code = send_resp.code,
"Feishu reply target is unavailable; falling back to a new chat message"
);
let mut create_target = target.clone();
create_target.reply_to = None;
let (url, body) = create_target.request(msg_type, content);
let response = self
.http_client
.post(url)
.bearer_auth(self.get_tenant_access_token().await?)
.json(&body)
.send()
.await
.map_err(|error| {
ChannelError::ConnectionError(format!(
"Fallback send message HTTP error: {error}"
))
})?;
let fallback: SendResp = response.json().await.map_err(|error| {
ChannelError::Other(format!("Parse fallback send response error: {error}"))
})?;
if fallback.code != 0 {
return Err(ChannelError::Other(format!(
"Fallback send failed: code={} msg={}",
fallback.code, fallback.msg
)));
}
return Ok(fallback
.data
.map_or_else(String::new, |data| data.message_id));
}
if send_resp.code != 0 {
return Err(ChannelError::Other(format!(
"Send message failed: code={} msg={}",
@ -979,7 +1093,9 @@ impl FeishuChannel {
)));
}
Ok(())
Ok(send_resp
.data
.map_or_else(String::new, |data| data.message_id))
}
/// Extract service_id from WebSocket URL query params
@ -1069,6 +1185,9 @@ impl FeishuChannel {
let msg_type = msg.message_type.as_str();
let raw_content = msg.content.clone();
let parent_id = msg.parent_id.clone();
let root_id = msg.root_id.clone();
let thread_id = msg.thread_id.clone().or_else(|| root_id.clone());
let chat_type = msg.chat_type.clone();
#[cfg(debug_assertions)]
tracing::debug!(msg_type = %msg_type, chat_id = %chat_id, open_id = %open_id, "Parsing message content");
@ -1101,6 +1220,9 @@ impl FeishuChannel {
content,
media,
parent_id,
root_id,
thread_id,
chat_type,
}))
}
@ -1244,7 +1366,17 @@ impl FeishuChannel {
};
let mut private_context = HashMap::new();
private_context.insert("feishu.message_id".to_string(), message_id);
private_context.insert("feishu.message_id".to_string(), message_id.clone());
private_context.insert("feishu.chat_type".to_string(), parsed.chat_type);
if let Some(parent_id) = parsed.parent_id.as_ref() {
private_context.insert("feishu.parent_id".to_string(), parent_id.clone());
}
if let Some(root_id) = parsed.root_id {
private_context.insert("feishu.root_id".to_string(), root_id);
}
if let Some(thread_id) = parsed.thread_id {
private_context.insert("feishu.thread_id".to_string(), thread_id);
}
if let Some(reaction_id) = reaction_id {
private_context.insert("feishu.reaction_id".to_string(), reaction_id);
}
@ -1257,7 +1389,7 @@ impl FeishuChannel {
received_at: crate::bus::message::current_timestamp(),
media: parsed.media,
channel_context: crate::bus::ChannelContext {
reply_to: parsed.parent_id,
reply_to: Some(message_id),
private: private_context,
},
};
@ -2144,59 +2276,19 @@ impl FeishuChannel {
/// Send an interactive card message to Feishu.
async fn send_interactive_card(
&self,
receive_id: &str,
receive_id_type: &str,
target: &FeishuSendTarget,
card_content: &str,
) -> Result<String, ChannelError> {
let token = self.get_tenant_access_token().await?;
let resp = self
.http_client
.post(format!(
"{}/im/v1/messages?receive_id_type={}",
FEISHU_API_BASE, receive_id_type
let message_id = self
.send_raw_message(target, "interactive", card_content)
.await?;
if message_id.is_empty() {
Err(ChannelError::Other(
"Feishu send response has no message_id".into(),
))
.header("Content-Type", "application/json")
.header("Authorization", format!("Bearer {}", token))
.json(&serde_json::json!({
"receive_id": receive_id,
"msg_type": "interactive",
"content": card_content
}))
.send()
.await
.map_err(|e| {
ChannelError::ConnectionError(format!("Send interactive card HTTP error: {}", e))
})?;
#[derive(Deserialize)]
struct SendResp {
code: i32,
msg: String,
data: Option<SendData>,
} else {
Ok(message_id)
}
#[derive(Deserialize)]
struct SendData {
message_id: String,
}
let send_resp: SendResp = resp.json().await.map_err(|e| {
ChannelError::Other(format!("Parse send interactive card response error: {}", e))
})?;
if send_resp.code != 0 {
return Err(ChannelError::Other(format!(
"Send interactive card failed: code={} msg={}",
send_resp.code, send_resp.msg
)));
}
send_resp
.data
.map(|data| data.message_id)
.filter(|message_id| !message_id.is_empty())
.ok_or_else(|| ChannelError::Other("Feishu send response has no message_id".into()))
}
async fn update_interactive_card(
@ -2246,8 +2338,7 @@ trait FeishuTurnApi: Send {
struct FeishuTurnBackend {
channel: FeishuChannel,
receive_id: String,
receive_id_type: &'static str,
target: FeishuSendTarget,
metadata: HashMap<String, String>,
}
@ -2256,7 +2347,7 @@ impl FeishuTurnApi for FeishuTurnBackend {
async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError> {
let card = FeishuChannel::build_card_content(markdown);
self.channel
.send_interactive_card(&self.receive_id, self.receive_id_type, &card)
.send_interactive_card(&self.target, &card)
.await
}
@ -2587,30 +2678,21 @@ impl Channel for FeishuChannel {
}
async fn open_turn(&self, target: TurnTarget) -> Result<Box<dyn TurnSink>, ChannelError> {
let (receive_id, receive_id_type) = if target.chat_id.starts_with("oc_") {
(target.chat_id, "chat_id")
} else {
(target.reply_to.unwrap_or(target.chat_id), "open_id")
};
let send_target =
FeishuSendTarget::from_message(target.chat_id, target.reply_to, &target.metadata);
Ok(Box::new(FeishuTurnSink::new(Box::new(FeishuTurnBackend {
channel: self.clone(),
receive_id,
receive_id_type,
target: send_target,
metadata: target.metadata,
}))))
}
async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> {
let receive_id = if msg.chat_id.starts_with("oc_") {
&msg.chat_id
} else {
msg.reply_to.as_ref().unwrap_or(&msg.chat_id)
};
let receive_id_type = if msg.chat_id.starts_with("oc_") {
"chat_id"
} else {
"open_id"
};
let target = FeishuSendTarget::from_message(
msg.chat_id.clone(),
msg.reply_to.clone(),
&msg.metadata,
);
// If no media, send as interactive card with raw markdown
if msg.media.is_empty() {
@ -2625,15 +2707,13 @@ impl Channel for FeishuChannel {
let chunks = Self::split_markdown_chunks(content);
for chunk in &chunks {
let card = Self::build_card_content(chunk);
if let Err(e) = self
.send_interactive_card(receive_id, receive_id_type, &card)
.await
{
if let Err(e) = self.send_interactive_card(&target, &card).await {
tracing::warn!(error = %e, "Failed to send interactive card, falling back to text");
let text_content = serde_json::json!({ "text": chunk }).to_string();
let result = self
.send_message_to_feishu(receive_id, receive_id_type, "text", &text_content)
.await;
.send_raw_message(&target, "text", &text_content)
.await
.map(|_| ());
self.remove_reaction_from_metadata(&msg.metadata).await;
return result;
}
@ -2643,8 +2723,6 @@ impl Channel for FeishuChannel {
}
// Handle multimodal message - send with media
let token = self.get_tenant_access_token().await?;
// Separate images (can embed in post) from files (sent as separate messages)
let mut image_items = Vec::new();
let mut file_items = Vec::new();
@ -2669,12 +2747,7 @@ impl Channel for FeishuChannel {
};
let file_content = serde_json::json!({"file_key": file_key}).to_string();
if let Err(e) = self
.send_message_to_feishu(
receive_id,
receive_id_type,
file_msg_type,
&file_content,
)
.send_raw_message(&target, file_msg_type, &file_content)
.await
{
tracing::warn!(error = %e, msg_type = file_msg_type, "Failed to send file message");
@ -2749,51 +2822,7 @@ impl Channel for FeishuChannel {
.to_string()
};
let resp = self
.http_client
.post(format!(
"{}/im/v1/messages?receive_id_type={}",
FEISHU_API_BASE, receive_id_type
))
.header("Content-Type", "application/json")
.header("Authorization", format!("Bearer {}", token))
.json(&serde_json::json!({
"receive_id": receive_id,
"msg_type": msg_type,
"content": content
}))
.send()
.await
.map_err(|e| {
ChannelError::ConnectionError(format!("Send multimodal message HTTP error: {}", e))
})?;
let send_status = resp.status();
let send_body = resp
.text()
.await
.map_err(|e| ChannelError::Other(format!("Failed to read send response: {}", e)))?;
tracing::debug!(status = %send_status, body = %send_body, msg_type = %msg_type, "Feishu send message");
#[derive(Deserialize)]
struct SendResp {
code: i32,
msg: String,
}
let send_resp: SendResp = serde_json::from_str(&send_body).map_err(|e| {
ChannelError::Other(format!(
"Parse send response error: {} | body: {}",
e, &send_body
))
})?;
if send_resp.code != 0 {
return Err(ChannelError::Other(format!(
"Send multimodal message failed: code={} msg={}",
send_resp.code, send_resp.msg
)));
}
self.send_raw_message(&target, msg_type, &content).await?;
// Remove pending reaction after successfully sending
self.remove_reaction_from_metadata(&msg.metadata).await;
@ -2960,6 +2989,52 @@ mod tests {
assert_eq!(parsed.content, "ask @Alice");
}
#[tokio::test]
async fn inbound_context_replies_to_current_message_and_preserves_thread_metadata() {
let mut channel = test_channel();
channel.config.reaction_emoji.clear();
*channel.bot_open_id.write().await = Some("ou_bot".to_string());
let mut frame = inbound_frame(
"om_current",
"ou_user",
"group",
"@_user_1 hello",
serde_json::json!([
{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "PicoBot"}
]),
);
let mut payload: serde_json::Value =
serde_json::from_slice(frame.payload.as_deref().unwrap()).unwrap();
payload["event"]["message"]["root_id"] = serde_json::json!("om_root");
payload["event"]["message"]["thread_id"] = serde_json::json!("omt_thread");
frame.payload = Some(payload.to_string().into_bytes());
let bus = Arc::new(MessageBus::new(4));
channel.process_event_frame(&bus, &frame).await;
let inbound = bus.consume_inbound().await.unwrap();
assert_eq!(
inbound.channel_context.reply_to.as_deref(),
Some("om_current")
);
assert_eq!(
inbound
.channel_context
.private
.get("feishu.thread_id")
.map(String::as_str),
Some("omt_thread")
);
assert_eq!(
inbound
.channel_context
.private
.get("feishu.root_id")
.map(String::as_str),
Some("om_root")
);
}
#[test]
fn post_mentions_can_gate_group_messages_when_top_level_mentions_are_absent() {
let message = LarkMessage {
@ -2972,6 +3047,8 @@ mod tests {
})
.to_string(),
parent_id: None,
root_id: None,
thread_id: None,
mentions: vec![],
};
@ -3056,6 +3133,32 @@ mod tests {
assert_eq!(reconnect_delay(100), Duration::from_secs(64));
}
#[test]
fn send_target_uses_reply_endpoint_and_thread_flag() {
let metadata = HashMap::from([("feishu.thread_id".to_string(), "omt_thread".to_string())]);
let target = FeishuSendTarget::from_message(
"oc_chat".to_string(),
Some("om_current".to_string()),
&metadata,
);
let (url, body) = target.request("interactive", "{}");
assert!(url.ends_with("/im/v1/messages/om_current/reply"));
assert_eq!(body["reply_in_thread"], true);
assert_eq!(body["msg_type"], "interactive");
assert!(body.get("receive_id").is_none());
}
#[test]
fn proactive_thread_target_creates_inside_thread() {
let metadata = HashMap::from([("feishu.thread_id".to_string(), "omt_thread".to_string())]);
let target = FeishuSendTarget::from_message("oc_chat".to_string(), None, &metadata);
let (url, body) = target.request("text", "{}");
assert!(url.ends_with("/im/v1/messages?receive_id_type=thread_id"));
assert_eq!(body["receive_id"], "omt_thread");
}
#[tokio::test]
async fn turn_sink_creates_once_updates_same_card_and_cleans_up_at_finish() {
let state = Arc::new(Mutex::new(MockTurnState::default()));