diff --git a/README.md b/README.md index 6186d8e..9c00ecf 100644 --- a/README.md +++ b/README.md @@ -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。 ### 会话 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 66613a1..95cc01f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -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 终态与独立消息并发写入同一目标。 ### 出站投递 diff --git a/src/channels/feishu.rs b/src/channels/feishu.rs index ed737d3..d43871b 100644 --- a/src/channels/feishu.rs +++ b/src/channels/feishu.rs @@ -159,6 +159,10 @@ struct LarkMessage { #[serde(default)] parent_id: Option, #[serde(default)] + root_id: Option, + #[serde(default)] + thread_id: Option, + #[serde(default)] mentions: Vec, } @@ -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, + root_id: Option, + thread_id: Option, + chat_type: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct FeishuSendTarget { + receive_id: String, + receive_id_type: &'static str, + reply_to: Option, + reply_in_thread: bool, +} + +impl FeishuSendTarget { + fn from_message( + chat_id: String, + reply_to: Option, + metadata: &HashMap, + ) -> 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, 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 { 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, + } + #[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 { - 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, + } 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, } @@ -2256,7 +2347,7 @@ impl FeishuTurnApi for FeishuTurnBackend { async fn create_card(&mut self, markdown: &str) -> Result { 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, 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()));