refactor: normalize inbound channel context

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

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 历史。

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<String>,
+    pub private: HashMap<String, String>,
+}
+
 #[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<MediaItem>,
-    /// Channel-specific data used internally by the channel (not forwarded).
+    pub channel_context: ChannelContext,
-    pub metadata: HashMap<String, String>,
-    /// Data forwarded from inbound to outbound (copied to OutboundMessage.metadata by gateway).
-    pub forwarded_metadata: HashMap<String, String>,
 }
 
 // ============================================================================

src/bus/mod.rs

@@ -3,8 +3,8 @@ pub mod message;
 
 pub use dispatcher::OutboundDispatcher;
 pub use message::{
-    ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage, MediaItem,
+    ChannelContext, ChatMessage, CompletionStatus, ContentBlock, ControlMessage, InboundMessage,
-    MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind,
+    MediaItem, MediaRef, MessageSource, OutboundMessage, ProviderReasoningState, SourceKind,
 };
 
 use std::sync::Arc;

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(),
+                    channel_context: Default::default(),
-                    forwarded_metadata: Default::default(),
                 };
                 if let Err(error) = bus.publish_inbound(msg).await {
                     self.uploads.restore(uploads).await;

src/channels/feishu.rs

@@ -1252,14 +1252,10 @@ impl FeishuChannel {
                                             }
                                         };
 
-                                        // forwarded_metadata is copied to OutboundMessage.metadata by the gateway.
+                                        let mut private_context = std::collections::HashMap::new();
-                                        let mut forwarded_metadata = std::collections::HashMap::new();
+                                        private_context.insert("feishu.message_id".to_string(), message_id.clone());
-                                        forwarded_metadata.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());
+                                            private_context.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());
                                         }
 
                                         #[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(),
+                                            channel_context: crate::bus::ChannelContext {
-                                            forwarded_metadata,
+                                                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");

src/gateway/router.rs

@@ -154,16 +154,7 @@ async fn process_inbound(
     session_manager: Arc<SessionManager>,
     inbound: InboundMessage,
 ) {
-    let result = session_manager
+    let result = session_manager.handle_message(&inbound).await;
-        .handle_message(
-            &inbound.channel,
-            &inbound.sender_id,
-            &inbound.chat_id,
-            &inbound.content,
-            inbound.media.clone(),
-            inbound.forwarded_metadata.clone(),
-        )
-        .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);

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<String, String> {
 
 fn outbound_turn_metadata(
     session_id: &str,
-    forwarded: &HashMap<String, String>,
+    private_context: &HashMap<String, String>,
 ) -> HashMap<String, String> {
-    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<String> {
 async fn deliver_terminal_fallback(
     handle: TurnDeliveryHandle,
     bus: &MessageBus,
-    channel: &str,
+    target: &crate::channels::TurnTarget,
-    chat_id: &str,
-    session_id: &str,
-    forwarded_metadata: &HashMap<String, String>,
     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(),
+        channel: target.channel.clone(),
-        chat_id: chat_id.to_string(),
+        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",
+            &target,
-            "chat",
-            "session",
-            &HashMap::new(),
             &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<MediaItem>,
-    forwarded_metadata: HashMap<String, String>,
+    channel_context: ChannelContext,
 }
 
 #[derive(Clone)]
@@ -2323,13 +2327,12 @@ impl SessionManager {
 
     pub async fn handle_message(
         &self,
-        channel: &str,
+        inbound: &InboundMessage,
-        _sender_id: &str,
-        chat_id: &str,
-        content: &str,
-        media: Vec<MediaItem>,
-        forwarded_metadata: HashMap<String, String>,
     ) -> Result<HandleResult, AgentError> {
+        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,
+                    received_at: inbound.received_at,
-                    forwarded_metadata,
+                    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<MediaRef> =
                         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(
+                    .start(turn_target.clone(), turn_receiver)
-                        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,
-                    )
                     .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,
+                        &turn_target,
-                        &task_cid,
-                        &unified_str,
-                        &task_metadata,
                         &turn_controller,
                     )
                     .await;