From 76685c3983d3e83f92647c8b999a075071406754 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Sun, 19 Jul 2026 13:42:43 +0800 Subject: [PATCH] refactor: normalize inbound channel context --- AGENTS.md | 1 + docs/ARCHITECTURE.md | 1 + src/bus/message.rs | 17 ++++-- src/bus/mod.rs | 4 +- src/channels/cli_chat.rs | 5 +- src/channels/feishu.rs | 18 +++---- src/gateway/router.rs | 46 +++++++++++----- src/session/session.rs | 111 ++++++++++++++++++++++----------------- 8 files changed, 122 insertions(+), 81 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 69e9476..7b5cd7a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del ### Functional Boundaries - **Channels** publish inbound messages through `MessageBus`; outbound writes arrive through `OutboundDispatcher` or a per-turn `TurnSink`. They know nothing about sessions or LLM +- **Inbound contract** carries normalized sender/time/media plus `ChannelContext`; core routing may interpret `reply_to` but must treat platform-private context as opaque reply data - **MessageBus** owns bounded inbound/outbound/control queues; outbound routing and retries belong to `OutboundDispatcher`, not the queue - **SessionManager** owns session state, dialog operations, context construction, per-session work queues, and persistence coordination - **TurnController** is the only owner of active Turn state; snapshots are complete latest-wins values, not token queues, and `Completed` is published only after atomic persistence succeeds diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8fe77e7..322afd2 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -125,6 +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。 - 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 历史。 diff --git a/src/bus/message.rs b/src/bus/message.rs index 21c5659..4085d71 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -154,6 +154,8 @@ pub struct ChatMessage { #[derive(Debug, Clone, Serialize, Deserialize)] pub enum SourceKind { + #[serde(rename = "user_input")] + UserInput, #[serde(rename = "system_notification")] SystemNotification, #[serde(rename = "cross_channel")] @@ -364,18 +366,23 @@ mod conversation_message_tests { // InboundMessage - Message from Channel to Bus (user input) // ============================================================================ +/// Opaque channel-owned context that may be carried to the corresponding reply. +/// Core routing understands `reply_to`; all other platform data remains private. +#[derive(Debug, Clone, Default)] +pub struct ChannelContext { + pub reply_to: Option, + pub private: HashMap, +} + #[derive(Debug, Clone)] pub struct InboundMessage { pub channel: String, pub sender_id: String, pub chat_id: String, pub content: String, - pub timestamp: i64, + pub received_at: i64, pub media: Vec, - /// Channel-specific data used internally by the channel (not forwarded). - pub metadata: HashMap, - /// Data forwarded from inbound to outbound (copied to OutboundMessage.metadata by gateway). - pub forwarded_metadata: HashMap, + pub channel_context: ChannelContext, } // ============================================================================ diff --git a/src/bus/mod.rs b/src/bus/mod.rs index f54fd1d..4050e8f 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -3,8 +3,8 @@ pub mod message; pub use dispatcher::OutboundDispatcher; pub use message::{ - ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage, MediaItem, - MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind, + ChannelContext, ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage, + MediaItem, MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind, }; use std::sync::Arc; diff --git a/src/channels/cli_chat.rs b/src/channels/cli_chat.rs index 337eb31..a1f625b 100644 --- a/src/channels/cli_chat.rs +++ b/src/channels/cli_chat.rs @@ -211,10 +211,9 @@ impl CliChatChannel { sender_id: "cli".to_string(), chat_id: target_chat_id, content, - timestamp: crate::bus::message::current_timestamp(), + received_at: crate::bus::message::current_timestamp(), media, - metadata: Default::default(), - forwarded_metadata: Default::default(), + channel_context: Default::default(), }; if let Err(error) = bus.publish_inbound(msg).await { self.uploads.restore(uploads).await; diff --git a/src/channels/feishu.rs b/src/channels/feishu.rs index 30d1d02..a549a64 100644 --- a/src/channels/feishu.rs +++ b/src/channels/feishu.rs @@ -1252,14 +1252,10 @@ impl FeishuChannel { } }; - // forwarded_metadata is copied to OutboundMessage.metadata by the gateway. - let mut forwarded_metadata = std::collections::HashMap::new(); - forwarded_metadata.insert("feishu.message_id".to_string(), message_id.clone()); + let mut private_context = std::collections::HashMap::new(); + private_context.insert("feishu.message_id".to_string(), message_id.clone()); if let Some(ref rid) = reaction_id { - forwarded_metadata.insert("feishu.reaction_id".to_string(), rid.clone()); - } - if let Some(ref pid) = parsed.parent_id { - forwarded_metadata.insert("feishu.parent_id".to_string(), pid.clone()); + private_context.insert("feishu.reaction_id".to_string(), rid.clone()); } #[cfg(debug_assertions)] @@ -1269,10 +1265,12 @@ impl FeishuChannel { sender_id: parsed.open_id.clone(), chat_id: parsed.chat_id.clone(), content: parsed.content.clone(), - timestamp: crate::bus::message::current_timestamp(), + received_at: crate::bus::message::current_timestamp(), media: parsed.media.clone(), - metadata: std::collections::HashMap::new(), - forwarded_metadata, + channel_context: crate::bus::ChannelContext { + reply_to: parsed.parent_id.clone(), + private: private_context, + }, }; if let Err(e) = self.handle_and_publish(&bus, &msg).await { tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus"); diff --git a/src/gateway/router.rs b/src/gateway/router.rs index bc69fe7..d7868ac 100644 --- a/src/gateway/router.rs +++ b/src/gateway/router.rs @@ -154,16 +154,7 @@ async fn process_inbound( session_manager: Arc, inbound: InboundMessage, ) { - let result = session_manager - .handle_message( - &inbound.channel, - &inbound.sender_id, - &inbound.chat_id, - &inbound.content, - inbound.media.clone(), - inbound.forwarded_metadata.clone(), - ) - .await; + let result = session_manager.handle_message(&inbound).await; match result { Ok(crate::session::session::HandleResult::AgentResponse(content)) => { @@ -189,7 +180,7 @@ async fn publish_command_output(bus: &MessageBus, inbound: InboundMessage, conte } async fn publish_output(bus: &MessageBus, inbound: InboundMessage, content: String, command: bool) { - let mut metadata = inbound.forwarded_metadata; + let mut metadata = inbound.channel_context.private; if command { metadata.insert("_type".to_string(), "command".to_string()); } @@ -197,7 +188,7 @@ async fn publish_output(bus: &MessageBus, inbound: InboundMessage, content: Stri channel: inbound.channel, chat_id: inbound.chat_id, content, - reply_to: None, + reply_to: inbound.channel_context.reply_to, media: vec![], metadata, delivery: None, @@ -353,6 +344,7 @@ fn is_priority_stop(content: &str) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::bus::ChannelContext; use std::collections::HashSet; use tokio::sync::Notify; @@ -374,6 +366,36 @@ mod tests { assert_eq!(keys.len(), 3); } + #[tokio::test] + async fn routed_output_preserves_reply_target_and_private_context() { + let bus = MessageBus::new(2); + let inbound = InboundMessage { + channel: "test".to_string(), + sender_id: "user".to_string(), + chat_id: "chat".to_string(), + content: "hello".to_string(), + received_at: 123, + media: vec![], + channel_context: ChannelContext { + reply_to: Some("parent".to_string()), + private: HashMap::from([("opaque".to_string(), "value".to_string())]), + }, + }; + + publish_command_output(&bus, inbound, "done".to_string()).await; + let output = bus.consume_outbound().await.unwrap(); + + assert_eq!(output.reply_to.as_deref(), Some("parent")); + assert_eq!( + output.metadata.get("opaque").map(String::as_str), + Some("value") + ); + assert_eq!( + output.metadata.get("_type").map(String::as_str), + Some("command") + ); + } + #[tokio::test] async fn slow_conversation_lane_does_not_block_another_lane() { let (slow_tx, slow_rx) = mpsc::channel(2); diff --git a/src/session/session.rs b/src/session/session.rs index 4d851d8..19f66e1 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -7,7 +7,8 @@ use super::persistence::{append_persisted_messages, finalize_turn_after_persiste use super::turn::{TurnBlock, TurnController, TurnSnapshot}; use super::turn_input::prepare_turn_input; use crate::bus::{ - ChatMessage, CompletionStatus, MediaItem, MediaRef, MessageSource, OutboundMessage, SourceKind, + ChannelContext, ChatMessage, CompletionStatus, InboundMessage, MediaItem, MediaRef, + MessageSource, OutboundMessage, SourceKind, }; use crate::mcp::get_mcp_status; use crate::storage::{Storage, StorageError}; @@ -28,9 +29,9 @@ fn outbound_session_metadata(session_id: &str) -> HashMap { fn outbound_turn_metadata( session_id: &str, - forwarded: &HashMap, + private_context: &HashMap, ) -> HashMap { - let mut metadata = forwarded.clone(); + let mut metadata = private_context.clone(); metadata.insert("_session_id".to_string(), session_id.to_string()); metadata } @@ -115,31 +116,28 @@ fn terminal_fallback_content(snapshot: &TurnSnapshot) -> Option { async fn deliver_terminal_fallback( handle: TurnDeliveryHandle, bus: &MessageBus, - channel: &str, - chat_id: &str, - session_id: &str, - forwarded_metadata: &HashMap, + target: &crate::channels::TurnTarget, controller: &TurnController, ) { let Err(error) = handle.wait().await else { return; }; - tracing::error!(channel, chat_id, error = %error, "Turn sink terminal delivery failed; using ordinary outbound fallback"); + tracing::error!(channel = %target.channel, chat_id = %target.chat_id, error = %error, "Turn sink terminal delivery failed; using ordinary outbound fallback"); let snapshot = controller.snapshot(); let Some(content) = terminal_fallback_content(&snapshot) else { return; }; let outbound = OutboundMessage { - channel: channel.to_string(), - chat_id: chat_id.to_string(), + channel: target.channel.clone(), + chat_id: target.chat_id.clone(), content, - reply_to: None, + reply_to: target.reply_to.clone(), media: vec![], - metadata: outbound_turn_metadata(session_id, forwarded_metadata), + metadata: target.metadata.clone(), delivery: None, }; if let Err(fallback_error) = bus.deliver_outbound(outbound).await { - tracing::error!(channel, chat_id, error = %fallback_error, "Ordinary terminal fallback delivery failed"); + tracing::error!(channel = %target.channel, chat_id = %target.chat_id, error = %fallback_error, "Ordinary terminal fallback delivery failed"); } } @@ -298,13 +296,17 @@ mod cancelled_partial_tests { let (sender, completion) = oneshot::channel(); sender.send(Err(DeliveryError::FinalTimedOut)).unwrap(); + let target = crate::channels::TurnTarget { + channel: "recording".to_string(), + chat_id: "chat".to_string(), + session_id: "session".to_string(), + reply_to: None, + metadata: HashMap::new(), + }; deliver_terminal_fallback( TurnDeliveryHandle { completion }, &bus, - "recording", - "chat", - "session", - &HashMap::new(), + &target, &controller, ) .await; @@ -399,10 +401,12 @@ struct ActiveTurnEmitter { /// A task to be processed by the per-session agent worker struct AgentTask { channel: String, + sender_id: String, chat_id: String, content: String, + received_at: i64, media: Vec, - forwarded_metadata: HashMap, + channel_context: ChannelContext, } #[derive(Clone)] @@ -2323,13 +2327,12 @@ impl SessionManager { pub async fn handle_message( &self, - channel: &str, - _sender_id: &str, - chat_id: &str, - content: &str, - media: Vec, - forwarded_metadata: HashMap, + inbound: &InboundMessage, ) -> Result { + let channel = inbound.channel.as_str(); + let sender_id = inbound.sender_id.as_str(); + let chat_id = inbound.chat_id.as_str(); + let content = inbound.content.as_str(); let unified_id = self.resolve_dialog_id(channel, chat_id).await?; tracing::debug!(unified_id = %unified_id, "handle_message resolved unified_id"); let session = self.get_or_create_session(&unified_id).await?; @@ -2363,10 +2366,12 @@ impl SessionManager { // Normal message: enqueue to per-session worker for serial processing. let task = AgentTask { channel: channel.to_string(), + sender_id: sender_id.to_string(), chat_id: chat_id.to_string(), content: content.to_string(), - media, - forwarded_metadata, + received_at: inbound.received_at, + media: inbound.media.clone(), + channel_context: inbound.channel_context.clone(), }; let session_clone = session.clone(); let unified_str = unified_id.to_string(); @@ -2534,7 +2539,8 @@ fn spawn_agent_worker( 'tasks: while let Some(task) = task_rx.recv().await { let task_chan = task.channel.clone(); let task_cid = task.chat_id.clone(); - let task_metadata = task.forwarded_metadata.clone(); + let task_metadata = task.channel_context.private.clone(); + let task_reply_to = task.channel_context.reply_to.clone(); // Phase 1: capture a stable session snapshot under lock. // Memory recall and compression happen outside this block so // /stop and other commands are not blocked behind slow I/O or @@ -2547,7 +2553,18 @@ fn spawn_agent_worker( } let media_refs: Vec = task.media.iter().map(MediaItem::to_media_ref).collect(); - guard.create_user_message(&task.content, media_refs) + let source = MessageSource { + kind: SourceKind::UserInput, + from_channel: Some(task.channel.clone()), + from_session: None, + from_user_id: Some(task.sender_id.clone()), + system_name: None, + task_id: None, + }; + let mut message = + guard.create_user_message_with_source(&task.content, media_refs, source); + message.timestamp = task.received_at; + message }; if let Err(e) = append_persisted_messages(&session, vec![user_message]).await { tracing::error!(error = %e, "Failed to persist user message"); @@ -2555,7 +2572,7 @@ fn spawn_agent_worker( channel: task_chan.clone(), chat_id: task_cid.clone(), content: "Failed to save your message, please try again.".to_string(), - reply_to: None, + reply_to: task_reply_to.clone(), media: vec![], metadata: outbound_turn_metadata(&unified_str, &task_metadata), delivery: None, @@ -2589,7 +2606,7 @@ fn spawn_agent_worker( chat_id: task_cid.clone(), content: "Agent creation failed, please try again." .to_string(), - reply_to: None, + reply_to: task_reply_to.clone(), media: vec![], metadata: outbound_turn_metadata(&unified_str, &task_metadata), delivery: None, @@ -2660,17 +2677,15 @@ fn spawn_agent_worker( ); let initial_turn = turn_controller.snapshot(); let active_turn_id = initial_turn.id.0.clone(); + let turn_target = crate::channels::TurnTarget { + channel: task_chan.clone(), + chat_id: task_cid.clone(), + session_id: unified_str.clone(), + reply_to: task_reply_to.clone(), + metadata: outbound_turn_metadata(&unified_str, &task_metadata), + }; let delivery_handle = match turn_delivery - .start( - crate::channels::TurnTarget { - channel: task_chan.clone(), - chat_id: task_cid.clone(), - session_id: unified_str.clone(), - reply_to: None, - metadata: outbound_turn_metadata(&unified_str, &task_metadata), - }, - turn_receiver, - ) + .start(turn_target.clone(), turn_receiver) .await { Ok(handle) => Some(handle), @@ -2712,6 +2727,7 @@ fn spawn_agent_worker( let cid2 = task_cid.clone(); let unified_str2 = unified_str.clone(); let task_metadata2 = task_metadata.clone(); + let task_reply_to2 = task_reply_to.clone(); let title_supervisor = worker_supervisor.clone(); let turn_lifecycle = &turn_controller; let process_future = async move { @@ -2763,7 +2779,7 @@ fn spawn_agent_worker( chat_id: cid2, content: "Context overflow handling failed." .to_string(), - reply_to: None, + reply_to: task_reply_to2.clone(), media: vec![], metadata: outbound_turn_metadata( &response_session_id, @@ -2826,7 +2842,7 @@ fn spawn_agent_worker( channel: chan2, chat_id: cid2, content: format!("Processing error: {}", e), - reply_to: None, + reply_to: task_reply_to2.clone(), media: vec![], metadata: outbound_turn_metadata( &response_session_id, @@ -2853,7 +2869,7 @@ fn spawn_agent_worker( channel: chan2, chat_id: cid2, content: format!("Processing error: {}", e), - reply_to: None, + reply_to: task_reply_to2.clone(), media: vec![], metadata: outbound_turn_metadata( &response_session_id, @@ -2907,7 +2923,7 @@ fn spawn_agent_worker( chat_id: cid2, content: "Failed to save the agent response, please try again." .to_string(), - reply_to: None, + reply_to: task_reply_to2.clone(), media: vec![], metadata: outbound_turn_metadata( &response_session_id, @@ -2933,7 +2949,7 @@ fn spawn_agent_worker( channel: chan2, chat_id: cid2, content: response, - reply_to: None, + reply_to: task_reply_to2.clone(), media: vec![], metadata: outbound_turn_metadata( &response_session_id, @@ -2976,10 +2992,7 @@ fn spawn_agent_worker( deliver_terminal_fallback( handle, &bus, - &task_chan, - &task_cid, - &unified_str, - &task_metadata, + &turn_target, &turn_controller, ) .await;