diff --git a/src/gateway/execution.rs b/src/gateway/execution.rs index 2e1b58b..afea3dd 100644 --- a/src/gateway/execution.rs +++ b/src/gateway/execution.rs @@ -64,6 +64,8 @@ pub(crate) struct MessageExecutionRequest<'a> { pub(crate) content: &'a str, pub(crate) media: Vec, pub(crate) live_emitter: Option>, + /// 消息接收时捕获的 topic_id，全程显式传递避免从共享状态重复读取竞态 + pub(crate) topic_id: Option, } pub(crate) struct ScheduledExecutionRequest<'a> { @@ -111,29 +113,42 @@ impl AgentExecutionService { } // 确定保存消息的话题 ID - // 如果是最新回合，使用当前话题；否则使用原始话题 - let target_topic_id = if is_current_turn { - session.current_topic(request.chat_id) - } else { - request.original_topic_id.as_deref() - }; + // 始终使用执行开始时捕获的 original_topic_id，避免从共享状态重复读取竞态 + let target_topic_id = request.original_topic_id.as_deref(); // 将结果消息保存到确定的话题 if let Some(topic_id) = target_topic_id { if is_current_turn { - // 如果是最新回合，使用 append_persisted_messages 保存到数据库并更新内存历史 - if let Err(err) = session.append_persisted_messages( - request.chat_id, - request.result.emitted_messages.clone(), - ) { - tracing::error!( - error = %err, - chat_id = %request.chat_id, - "Failed to append messages to session history" - ); + // 检查当前活跃 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(), + ) { + tracing::error!( + error = %err, + chat_id = %request.chat_id, + "Failed to append messages to session history" + ); + } + } 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 - let original_topic_id = session_guard - .current_topic(request.chat_id) - .map(|s| s.to_string()); + // 优先使用消息接收时捕获的 topic_id，消除 #1 与 #2 之间的竞态 + 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()), + }; 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, 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 diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 13daab7..12a21e0 100644 --- a/src/gateway/processor.rs +++ b/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 { diff --git a/src/gateway/session.rs b/src/gateway/session.rs index b73afc4..93d9dd3 100644 --- a/src/gateway/session.rs +++ b/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 - .insert(chat_id.to_string(), receiver); + let key = topic_id.unwrap_or(chat_id).to_string(); + 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)) })?; - self.add_message(chat_id, message); + + // 只有当写入的 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 { 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 { + // 优先使用显式传入的 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); - let topic_id = self.current_topic(session_chat_id).map(|s| s.to_string()); + // 优先按 topic_id 查找；无 topic 时回退 chat_id + 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, live_emitter: Option>, + topic_id: Option<&str>, ) -> Result, 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(); } diff --git a/src/gateway/session_message_service.rs b/src/gateway/session_message_service.rs index 855cbda..4733e61 100644 --- a/src/gateway/session_message_service.rs +++ b/src/gateway/session_message_service.rs @@ -28,6 +28,7 @@ impl SessionMessageService { content: &str, media: Vec, live_emitter: Option>, + topic_id: Option<&str>, ) -> Result, AgentError> { #[cfg(debug_assertions)] { @@ -54,6 +55,7 @@ impl SessionMessageService { content, media, live_emitter, + topic_id: topic_id.map(|s| s.to_string()), }) .await?;