fix: 话题隔离 - topic_id 全程显式传递，修复消息错投与并发阻塞

问题：WS 端一个话题执行中时新建另一话题发消息，新会话无响应
（串行锁按 chat_id 阻塞），且首个话题完成后用户消息错误出现在
旧话题而非新话题（执行路径多次从共享 UI 状态读取 topic_id 产生竞态）。

核心修复（第一性原则）：
执行上下文应在消息接收时一次性捕获，全程显式传递，不从共享可变
状态重复读取。

1. processor.rs: process_one 入口捕获 current_topic，传入 handle_message
   和 set_agent_cancel_token
2. session_message_service.rs: handle_message 签名加 topic_id 参数，
   透传给 MessageExecutionRequest
3. execution.rs: MessageExecutionRequest 加 topic_id 字段；
   - 串行锁键改用 topic_id（不同 topic 并发，同 topic 串行）
   - original_topic_id 优先用传入值，消除锁等待期间 topic 切换竞态
   - append_persisted_message 调用传入 original_topic_id
   - create_agent 调用传入 original_topic_id
4. session.rs: append_persisted_message 加 explicit_topic_id 参数；
   create_agent/create_agent_with_provider_config 加 explicit_topic_id；
   set_cancel_receiver/set_agent_cancel_token 加 topic_id 参数；
   pending_cancel_tokens 查找改为优先 topic_id（避免并发 topic 执行时
   cancel token 互相覆盖）

对抗性审查补丁：
- append_persisted_message: 仅当写入 topic 匹配当前活跃 topic 时才更新
  内存历史，避免旧 topic 的排队消息污染已切换到的新 topic 内存历史
- prepare_and_execute_scheduled_task: 锁前一次性捕获 topic_id，锁后
  复用同一值作为 original_topic_id，保证锁键与写入目标一致

已验证：cargo check 通过，gateway 模块 48 个测试通过（1 个预存在的
prompt 模板测试失败，与本次修改无关）。

src/gateway/execution.rs

@@ -64,6 +64,8 @@ pub(crate) struct MessageExecutionRequest<'a> {
     pub(crate) content: &'a str,
     pub(crate) media: Vec<MediaItem>,
     pub(crate) live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
+    /// 消息接收时捕获的 topic_id，全程显式传递避免从共享状态重复读取竞态
+    pub(crate) topic_id: Option<String>,
 }
 
 pub(crate) struct ScheduledExecutionRequest<'a> {
@@ -111,17 +113,16 @@ impl AgentExecutionService {
         }
 
         // 确定保存消息的话题 ID
-        // 如果是最新回合，使用当前话题；否则使用原始话题
+        // 始终使用执行开始时捕获的 original_topic_id，避免从共享状态重复读取竞态
-        let target_topic_id = if is_current_turn {
+        let target_topic_id = request.original_topic_id.as_deref();
-            session.current_topic(request.chat_id)
-        } else {
-            request.original_topic_id.as_deref()
-        };
 
         // 将结果消息保存到确定的话题
         if let Some(topic_id) = target_topic_id {
             if is_current_turn {
-                // 如果是最新回合，使用 append_persisted_messages 保存到数据库并更新内存历史
+                // 检查当前活跃 topic 是否仍是 original_topic_id
+                let current_tid = session.current_topic(request.chat_id);
+                if current_tid.as_deref() == Some(topic_id) {
+                    // 话题未切换，安全更新内存历史
                     if let Err(err) = session.append_persisted_messages(
                         request.chat_id,
                         request.result.emitted_messages.clone(),
@@ -133,7 +134,21 @@ impl AgentExecutionService {
                         );
                     }
                 } else {
-                // 如果用户已切换话题，只保存到原始话题（不更新内存历史）
+                    // 话题已切换，只写 DB 不更新内存（避免污染新话题的历史）
+                    if let Err(err) = session.append_messages_to_topic(
+                        request.chat_id,
+                        topic_id,
+                        &request.result.emitted_messages,
+                    ) {
+                        tracing::error!(
+                            error = %err,
+                            topic_id = %topic_id,
+                            "Failed to append messages to topic"
+                        );
+                    }
+                }
+            } else {
+                // stale：只写 DB
                 if let Err(err) = session.append_messages_to_topic(
                     request.chat_id,
                     topic_id,
@@ -147,7 +162,7 @@ impl AgentExecutionService {
                 }
             }
         } else if is_current_turn {
-            // 如果没有话题，直接更新内存历史（append_persisted_messages 会处理持久化）
+            // 没有话题：直接更新内存历史（append_persisted_messages 会处理持久化）
             if let Err(err) = session.append_persisted_messages(
                 request.chat_id,
                 request.result.emitted_messages.clone(),
@@ -205,7 +220,8 @@ impl AgentExecutionService {
         // 不同 topic 之间互不阻塞，支持多话题并发执行
         let serial_lock = {
             let mut session_guard = request.session.lock().await;
-            session_guard.topic_serial_lock(request.chat_id)
+            let lock_key = request.topic_id.as_deref().unwrap_or(request.chat_id);
+            session_guard.topic_serial_lock(lock_key)
         };
 
         // 等待该 topic 的前一条消息处理完成（含压缩）
@@ -236,13 +252,20 @@ impl AgentExecutionService {
             let history_before = session_guard.get_or_create_history(request.chat_id).clone();
             let user_message_count = history_before.iter().filter(|m| m.role == "user").count();
 
-            // 在添加用户消息前，记录当前话题 ID
+            // 优先使用消息接收时捕获的 topic_id，消除 #1 与 #2 之间的竞态
-            let original_topic_id = session_guard
+            let original_topic_id = match &request.topic_id {
+                Some(tid) => Some(tid.clone()),
+                None => session_guard
                     .current_topic(request.chat_id)
-                .map(|s| s.to_string());
+                    .map(|s| s.to_string()),
+            };
 
             let user_message = session_guard.create_user_message(&enriched_content, media_refs);
-            session_guard.append_persisted_message(request.chat_id, user_message.clone())?;
+            session_guard.append_persisted_message(
+                request.chat_id,
+                original_topic_id.as_deref(),
+                user_message.clone(),
+            )?;
 
             // 再获取包含新消息的完整历史记录
             let history = session_guard.get_or_create_history(request.chat_id).clone();
@@ -252,6 +275,7 @@ impl AgentExecutionService {
                 request.chat_id,
                 Some(request.sender_id),
                 Some(&user_message.id),
+                original_topic_id.as_deref(),
             )?;
             if let Some(handler) = request.live_emitter.clone() {
                 agent = agent.with_emitted_message_handler(handler);
@@ -293,9 +317,15 @@ impl AgentExecutionService {
         request: ScheduledExecutionRequest<'_>,
     ) -> Result<Vec<OutboundMessage>, AgentError> {
         // 获取该 topic 的串行锁（与普通消息路径共享，保证串行执行）
-        let serial_lock = {
+        // 定时任务由调度器触发，无用户消息竞态；在锁前一次性捕获 topic_id，
+        // 锁后复用同一值作为 original_topic_id，保证锁键与写入目标一致。
+        let (serial_lock, lock_time_topic_id) = {
             let mut session_guard = request.session.lock().await;
-            session_guard.topic_serial_lock(request.chat_id)
+            let tid = session_guard
+                .current_topic(request.chat_id)
+                .map(|s| s.to_string());
+            let lock_key = tid.as_deref().unwrap_or(request.chat_id);
+            (session_guard.topic_serial_lock(lock_key), tid)
         };
 
         // 等待该 topic 的前一条消息处理完成（含压缩）
@@ -318,10 +348,14 @@ impl AgentExecutionService {
             session_guard.ensure_chat_loaded(request.chat_id)?;
             session_guard.ensure_agent_prompt_before_user_message(request.chat_id)?;
 
+            // 复用锁前捕获的 topic_id，保证锁键与写入目标一致
+            let original_topic_id = lock_time_topic_id.clone();
+
             let scheduled_system_prompt =
                 compose_scheduled_task_system_prompt(request.system_prompt);
             session_guard.append_persisted_message(
                 request.chat_id,
+                original_topic_id.as_deref(),
                 ChatMessage::system_with_context(
                     &scheduled_system_prompt,
                     Some(SYSTEM_CONTEXT_SCHEDULED_PROMPT.to_string()),
@@ -332,13 +366,12 @@ impl AgentExecutionService {
             let history_before = session_guard.get_or_create_history(request.chat_id).clone();
             let user_message_count = history_before.iter().filter(|m| m.role == "user").count();
 
-            // 在添加用户消息前，记录当前话题 ID
-            let original_topic_id = session_guard
-                .current_topic(request.chat_id)
-                .map(|s| s.to_string());
-
             let user_message = session_guard.create_user_message(request.prompt, Vec::new());
-            session_guard.append_persisted_message(request.chat_id, user_message.clone())?;
+            session_guard.append_persisted_message(
+                request.chat_id,
+                original_topic_id.as_deref(),
+                user_message.clone(),
+            )?;
 
             // 再获取包含新消息的完整历史记录
             let history = session_guard.get_or_create_history(request.chat_id).clone();
@@ -350,6 +383,7 @@ impl AgentExecutionService {
                 Some(request.sender_id),
                 Some(&user_message.id),
                 request.provider_config.clone(),
+                original_topic_id.as_deref(),
             )?;
 
             // 获取 store 和 session_id，用于构造消息持久化 handler

src/gateway/processor.rs

@@ -269,7 +269,7 @@ impl InboundProcessor {
         if let Some(ref topic_id) = current_topic {
             let cancel_rx = self.cancel_manager.register(topic_id).await;
             self.session_manager
-                .set_agent_cancel_token(&channel, &chat_id, cancel_rx)
+                .set_agent_cancel_token(&channel, &chat_id, Some(topic_id.as_str()), cancel_rx)
                 .await;
         }
 
@@ -282,6 +282,7 @@ impl InboundProcessor {
                 &inbound.content,
                 inbound.media,
                 Some(live_emitter),
+                current_topic.as_deref(),
             )
             .await
         {

src/gateway/session.rs

@@ -323,14 +323,16 @@ impl Session {
     /// 存入待使用的取消信号接收端。
     ///
     /// 在 Agent 执行前由处理器调用，Agent 构建时（create_agent）自动消费。
-    /// 每个 chat_id 同时只允许一个 pending token；新 token 会替换旧 token。
+    /// 优先按 topic_id 键化（不同 topic 的 token 互不覆盖）；
+    /// 无 topic 时回退到 chat_id。
     pub fn set_cancel_receiver(
         &mut self,
         chat_id: &str,
+        topic_id: Option<&str>,
         receiver: tokio::sync::watch::Receiver<()>,
     ) {
-        self.pending_cancel_tokens
+        let key = topic_id.unwrap_or(chat_id).to_string();
-            .insert(chat_id.to_string(), receiver);
+        self.pending_cancel_tokens.insert(key, receiver);
     }
 
     /// 获取当前话题 ID（指定 chat）
@@ -431,20 +433,39 @@ impl Session {
         self.history.clear_chat_history(chat_id)
     }
 
-    /// 将消息写入内存与持久化层（使用当前 topic）
+    /// 将消息写入内存与持久化层。
+    /// 优先使用显式传入的 topic_id；未传入时回退到当前 chat 的活跃 topic。
+    /// 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史，
+    /// 避免旧 topic 的消息污染已切换到的新 topic 的内存历史。
     pub fn append_persisted_message(
         &mut self,
         chat_id: &str,
+        explicit_topic_id: Option<&str>,
         message: ChatMessage,
     ) -> Result<(), AgentError> {
         let session_id = self.persistent_session_id(chat_id);
-        let topic_id = self.history.chat_topic(chat_id).map(|s| s.to_string());
+        let topic_id = explicit_topic_id
+            .map(|s| s.to_string())
+            .or_else(|| self.history.chat_topic(chat_id).map(|s| s.to_string()));
         self.store
             .append_message_with_topic(&session_id, topic_id.as_deref(), &message)
             .map_err(|err| {
                 AgentError::Other(format!("append message persistence error: {}", err))
             })?;
+
+        // 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史。
+        // 当用户已切换到新 topic 时，旧 topic 的排队消息不应污染新 topic 的内存历史。
+        let current_chat_topic = self.history.chat_topic(chat_id);
+        if topic_id.as_deref() == current_chat_topic {
             self.add_message(chat_id, message);
+        } else {
+            tracing::info!(
+                chat_id = %chat_id,
+                write_topic_id = ?topic_id,
+                current_topic_id = ?current_chat_topic,
+                "Skipping memory history update: message belongs to a different topic"
+            );
+        }
 
         // 更新 topic 的最后活跃时间
         if let Some(ref topic_id) = topic_id {
@@ -579,6 +600,7 @@ impl Session {
         chat_id: &str,
         sender_id: Option<&str>,
         message_id: Option<&str>,
+        explicit_topic_id: Option<&str>,
     ) -> Result<AgentLoop, AgentError> {
         self.create_agent_with_provider_config(
             chat_id,
@@ -586,6 +608,7 @@ impl Session {
             sender_id,
             message_id,
             self.provider_config.clone(),
+            explicit_topic_id,
         )
     }
 
@@ -596,10 +619,21 @@ impl Session {
         sender_id: Option<&str>,
         message_id: Option<&str>,
         provider_config: LLMProviderConfig,
+        explicit_topic_id: Option<&str>,
     ) -> Result<AgentLoop, AgentError> {
+        // 优先使用显式传入的 topic_id；回退到当前 chat 的活跃 topic
+        let topic_id = explicit_topic_id
+            .map(|s| s.to_string())
+            .or_else(|| self.current_topic(session_chat_id).map(|s| s.to_string()));
+
         // 消费 pending 的取消信号接收端（如果存在）
-        let cancel_token = self.pending_cancel_tokens.remove(session_chat_id);
+        // 优先按 topic_id 查找；无 topic 时回退 chat_id
-        let topic_id = self.current_topic(session_chat_id).map(|s| s.to_string());
+        let cancel_token = match &topic_id {
+            Some(tid) => self.pending_cancel_tokens.remove(tid)
+                .or_else(|| self.pending_cancel_tokens.remove(session_chat_id)),
+            None => self.pending_cancel_tokens.remove(session_chat_id),
+        };
+
         self.agent_factory.create(AgentBuildRequest {
             channel_name: &self.channel_name,
             session_chat_id,
@@ -819,10 +853,11 @@ impl SessionManager {
         &self,
         channel_name: &str,
         chat_id: &str,
+        topic_id: Option<&str>,
         token: tokio::sync::watch::Receiver<()>,
     ) {
         if let Some(session) = self.get(channel_name).await {
-            session.lock().await.set_cancel_receiver(chat_id, token);
+            session.lock().await.set_cancel_receiver(chat_id, topic_id, token);
         }
     }
 
@@ -844,6 +879,7 @@ impl SessionManager {
         content: &str,
         media: Vec<crate::bus::MediaItem>,
         live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
+        topic_id: Option<&str>,
     ) -> Result<Vec<OutboundMessage>, AgentError> {
         self.messages
             .handle_message(
@@ -853,6 +889,7 @@ impl SessionManager {
                 content,
                 media,
                 live_emitter,
+                topic_id,
             )
             .await
     }
@@ -972,12 +1009,12 @@ mod tests {
 
         let first = session.create_user_message("first", Vec::new());
         let first_id = first.id.clone();
-        session.append_persisted_message("chat-1", first).unwrap();
+        session.append_persisted_message("chat-1", None, first).unwrap();
         assert!(session.is_latest_user_message("chat-1", &first_id));
 
         let second = session.create_user_message("second", Vec::new());
         let second_id = second.id.clone();
-        session.append_persisted_message("chat-1", second).unwrap();
+        session.append_persisted_message("chat-1", None, second).unwrap();
 
         assert!(!session.is_latest_user_message("chat-1", &first_id));
         assert!(session.is_latest_user_message("chat-1", &second_id));
@@ -1021,17 +1058,17 @@ mod tests {
 
         let first = session.create_user_message("first", Vec::new());
         let first_id = first.id.clone();
-        session.append_persisted_message("chat-1", first).unwrap();
+        session.append_persisted_message("chat-1", None, first).unwrap();
         session
-            .append_persisted_message("chat-1", ChatMessage::assistant("answer-1"))
+            .append_persisted_message("chat-1", None, ChatMessage::assistant("answer-1"))
             .unwrap();
 
         let second = session.create_user_message("second", Vec::new());
         session
-            .append_persisted_message("chat-1", second.clone())
+            .append_persisted_message("chat-1", None, second.clone())
             .unwrap();
         session
-            .append_persisted_message("chat-1", ChatMessage::assistant("answer-2"))
+            .append_persisted_message("chat-1", None, ChatMessage::assistant("answer-2"))
             .unwrap();
 
         let session_id = session.persistent_session_id("chat-1");
@@ -1216,7 +1253,7 @@ mod tests {
         .unwrap();
 
         let outbound = session_manager
-            .handle_message("test-channel", "user-1", "chat-1", "hello", Vec::new(), None)
+            .handle_message("test-channel", "user-1", "chat-1", "hello", Vec::new(), None, None)
             .await
             .unwrap();
 
@@ -2089,7 +2126,7 @@ mod tests {
 
         for turn in 0..100 {
             session
-                .append_persisted_message("chat-1", ChatMessage::user(format!("user-{turn}")))
+                .append_persisted_message("chat-1", None, ChatMessage::user(format!("user-{turn}")))
                 .unwrap();
         }
 
@@ -2163,7 +2200,7 @@ mod tests {
 
         for turn in 0..100 {
             session
-                .append_persisted_message("chat-1", ChatMessage::user(format!("user-{turn}")))
+                .append_persisted_message("chat-1", None, ChatMessage::user(format!("user-{turn}")))
                 .unwrap();
         }
 

src/gateway/session_message_service.rs

@@ -28,6 +28,7 @@ impl SessionMessageService {
         content: &str,
         media: Vec<MediaItem>,
         live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
+        topic_id: Option<&str>,
     ) -> Result<Vec<OutboundMessage>, AgentError> {
         #[cfg(debug_assertions)]
         {
@@ -54,6 +55,7 @@ impl SessionMessageService {
                 content,
                 media,
                 live_emitter,
+                topic_id: topic_id.map(|s| s.to_string()),
             })
             .await?;