From 4baa8e7a6bc319039253e7e22e200646cc0d357a Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 28 Jul 2026 15:32:42 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=86=85=E5=AD=98=E5=8E=86=E5=8F=B2?= =?UTF-8?q?=E6=8C=89=20topic=5Fid=20=E9=94=AE=E5=8C=96=EF=BC=8C=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20replace=5Ftopic=5Fhistory=20=E4=BF=AE=E5=A4=8D=20DB?= =?UTF-8?q?=20=E5=8E=8B=E7=BC=A9=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SessionHistory.chat_histories 改为 topic_histories 按 topic_id 键化，消除多话题并发时内存历史互相覆盖的根因 - is_current_turn 改用 current_topic == original_topic_id 直接比较，替代失效的内存最新消息匹配判断 - 新增 ConversationRepository::replace_topic_history 按 topic 删除/插入消息，解决 replace_active_history 删除整个 session 消息的 pre-existing 问题 - compaction 路径改用 replace_topic_history，避免压缩时覆盖其他 topic 的 DB 消息 - switch_topic 不再 remove_history，不同 topic 历史独立存储互斥 - append_persisted_message 仅当 topic_id == current_topic 时更新内存，防止延迟消息污染新话题历史 --- src/gateway/compaction.rs | 20 ++-- src/gateway/execution.rs | 109 ++++++++--------- src/gateway/session.rs | 206 ++++++++++++++------------------- src/gateway/session_history.rs | 154 ++++++++++++------------ src/storage/mod.rs | 113 ++++++++++++++++++ src/storage/ports.rs | 20 ++++ 6 files changed, 365 insertions(+), 257 deletions(-) diff --git a/src/gateway/compaction.rs b/src/gateway/compaction.rs index 24465de..39c2542 100644 --- a/src/gateway/compaction.rs +++ b/src/gateway/compaction.rs @@ -13,17 +13,21 @@ use super::session::Session; /// has already finished by this point there is no response-time impact, and /// the synchronous guarantee means the next execution always starts with /// freshly compacted history. +/// +/// 按 topic_id 隔离：压缩只处理指定 topic 的历史，DB 替换也只影响该 topic。 pub(crate) async fn schedule_background_history_compaction( session: Arc>, chat_id: impl Into, + topic_id: impl Into, ) -> Result<(), AgentError> { let chat_id = chat_id.into(); + let topic_id = topic_id.into(); let mut session_guard = session.lock().await; session_guard.ensure_persistent_session(&chat_id)?; - session_guard.ensure_chat_loaded(&chat_id)?; + session_guard.ensure_chat_loaded(&chat_id, Some(&topic_id))?; - let history = session_guard.get_or_create_history(&chat_id).clone(); + let history = session_guard.get_or_create_history(&topic_id).clone(); let compressor = session_guard.compressor().clone(); if !compressor.should_compress(&history) { @@ -36,6 +40,7 @@ pub(crate) async fn schedule_background_history_compaction( tracing::info!( chat_id = %chat_id, + topic_id = %topic_id, msg_count = history.len(), "Starting synchronous two-segment compression" ); @@ -47,19 +52,20 @@ pub(crate) async fn schedule_background_history_compaction( .compress_two_segment(&history, &provider_config) .await?; - // Replace the entire history with the compressed result. - // Since we hold the lock, no concurrent modifications can occur. + // Replace only this topic's history in DB (not the entire session). + // This avoids clobbering other topics' messages during compaction. store - .replace_active_history(&session_id, &compressed) - .map_err(|e| AgentError::Other(format!("replace_active_history error: {}", e)))?; + .replace_topic_history(&session_id, &topic_id, &compressed) + .map_err(|e| AgentError::Other(format!("replace_topic_history error: {}", e)))?; tracing::info!( chat_id = %chat_id, + topic_id = %topic_id, compressed_msg_count = compressed.len(), "Two-segment compression committed" ); - session_guard.reload_chat_history(&chat_id)?; + session_guard.reload_topic_history(&chat_id, &topic_id)?; Ok(()) } diff --git a/src/gateway/execution.rs b/src/gateway/execution.rs index afea3dd..2a61c0b 100644 --- a/src/gateway/execution.rs +++ b/src/gateway/execution.rs @@ -91,13 +91,19 @@ impl AgentExecutionService { session: &mut Session, request: FinalizeAgentResultRequest<'_>, ) -> Result { - // 检查是否是最新的用户回合 - let is_current_turn = - session.matches_current_user_turn(request.chat_id, request.user_message); + // 判断是否是最新的用户回合 + // 直接比较 current_topic(chat_id) 与 original_topic_id + // 这比"检查内存历史最新消息"更可靠，且天然处理"切走又切回"的 case + let is_current_turn = match request.original_topic_id.as_deref() { + Some(orig_tid) => session.current_topic(request.chat_id).as_deref() == Some(orig_tid), + None => true, // 无 topic 时总是视为当前回合 + }; if !is_current_turn { let (latest_user_id, latest_user_preview, compression_in_flight, history_len) = - session.stale_result_diagnostics(request.chat_id); + session.stale_result_diagnostics( + request.original_topic_id.as_deref().unwrap_or(request.chat_id), + ); tracing::info!( channel = %request.channel_name, chat_id = %request.chat_id, @@ -119,36 +125,19 @@ impl AgentExecutionService { // 将结果消息保存到确定的话题 if let Some(topic_id) = target_topic_id { if is_current_turn { - // 检查当前活跃 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" - ); - } + // 话题未切换（current_topic == original_topic_id），安全更新内存历史 + if let Err(err) = session.append_persisted_messages( + topic_id, + request.result.emitted_messages.clone(), + ) { + tracing::error!( + error = %err, + topic_id = %topic_id, + "Failed to append messages to session history" + ); } } else { - // stale：只写 DB + // 话题已切换，只写 DB 不更新内存（避免污染新话题的历史） if let Err(err) = session.append_messages_to_topic( request.chat_id, topic_id, @@ -163,6 +152,7 @@ impl AgentExecutionService { } } else if is_current_turn { // 没有话题：直接更新内存历史（append_persisted_messages 会处理持久化） + // 无 topic 场景用 chat_id 作为 topic_histories 的回退 key if let Err(err) = session.append_persisted_messages( request.chat_id, request.result.emitted_messages.clone(), @@ -232,7 +222,16 @@ impl AgentExecutionService { let mut session_guard = request.session.lock().await; session_guard.ensure_persistent_session(request.chat_id)?; - session_guard.ensure_chat_loaded(request.chat_id)?; + + // 优先使用消息接收时捕获的 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()), + }; + + session_guard.ensure_chat_loaded(request.chat_id, original_topic_id.as_deref())?; session_guard.ensure_agent_prompt_before_user_message(request.chat_id)?; @@ -249,17 +248,11 @@ impl AgentExecutionService { enrich_user_content_with_media_refs(request.content, &media_refs)?; // 先计算 user_message_count（在添加新消息之前） - let history_before = session_guard.get_or_create_history(request.chat_id).clone(); + // 无 topic 时用 chat_id 作为 topic_histories 的回退 key + let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id); + let history_before = session_guard.get_or_create_history(history_key).clone(); let user_message_count = history_before.iter().filter(|m| m.role == "user").count(); - // 优先使用消息接收时捕获的 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, @@ -268,7 +261,7 @@ impl AgentExecutionService { )?; // 再获取包含新消息的完整历史记录 - let history = session_guard.get_or_create_history(request.chat_id).clone(); + let history = session_guard.get_or_create_history(history_key).clone(); session_guard.record_skill_offer(request.chat_id)?; let mut agent = session_guard.create_agent( @@ -336,21 +329,21 @@ impl AgentExecutionService { session_guard.ensure_persistent_session(request.chat_id)?; + // 复用锁前捕获的 topic_id，保证锁键与写入目标一致 + let original_topic_id = lock_time_topic_id.clone(); + // 如果 fresh_session 为 true，清理历史（内存 + 数据库） if request.fresh_session { - session_guard.clear_chat_history(request.chat_id)?; + session_guard.clear_chat_history(request.chat_id, original_topic_id.as_deref())?; tracing::info!( chat_id = %request.chat_id, "Fresh session enabled, history cleared" ); } - session_guard.ensure_chat_loaded(request.chat_id)?; + session_guard.ensure_chat_loaded(request.chat_id, original_topic_id.as_deref())?; 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( @@ -363,7 +356,8 @@ impl AgentExecutionService { )?; // 先计算 user_message_count（在添加新消息之前） - let history_before = session_guard.get_or_create_history(request.chat_id).clone(); + let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id); + let history_before = session_guard.get_or_create_history(history_key).clone(); let user_message_count = history_before.iter().filter(|m| m.role == "user").count(); let user_message = session_guard.create_user_message(request.prompt, Vec::new()); @@ -374,7 +368,7 @@ impl AgentExecutionService { )?; // 再获取包含新消息的完整历史记录 - let history = session_guard.get_or_create_history(request.chat_id).clone(); + let history = session_guard.get_or_create_history(history_key).clone(); session_guard.record_skill_offer(request.chat_id)?; let agent = session_guard.create_agent_with_provider_config( @@ -426,7 +420,7 @@ impl AgentExecutionService { metadata: request.metadata, suppress_live_tool_calls: false, execution_kind: "scheduled_task", - original_topic_id, + original_topic_id: original_topic_id.clone(), }, ) .await?; @@ -434,7 +428,8 @@ impl AgentExecutionService { // 清理内存历史，释放内存（数据库历史保留） { let mut session_guard = request.session.lock().await; - session_guard.remove_history(request.chat_id); + let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id); + session_guard.remove_history(history_key); tracing::info!( chat_id = %request.chat_id, "Scheduled task completed, memory history released" @@ -452,6 +447,7 @@ impl AgentExecutionService { let channel_name = request.channel_name.to_string(); let chat_id = request.chat_id.to_string(); let execution_kind = request.execution_kind.to_string(); + let topic_id = request.original_topic_id.clone(); let finalized_result = { let mut session_guard = session.lock().await; @@ -459,8 +455,13 @@ impl AgentExecutionService { }; if finalized_result.should_schedule_compaction { - if let Err(error) = - schedule_background_history_compaction(session.clone(), chat_id.clone()).await + let compaction_topic_id = topic_id.unwrap_or_else(|| chat_id.clone()); + if let Err(error) = schedule_background_history_compaction( + session.clone(), + chat_id.clone(), + compaction_topic_id, + ) + .await { tracing::warn!( channel = %channel_name, diff --git a/src/gateway/session.rs b/src/gateway/session.rs index 93d9dd3..5524371 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -340,27 +340,20 @@ impl Session { self.history.chat_topic(chat_id) } - /// 获取历史所对应的话题 ID（指定 chat） - pub fn history_topic(&self, chat_id: &str) -> Option<&str> { - self.history.history_topic(chat_id) - } - - /// 切换话题 - 清除当前历史并加载新话题的历史 + /// 切换话题 - 设置当前 topic 并加载新话题的历史到内存 + /// 不同 topic 的历史在 topic_histories 中独立存储，切换不互斥。 pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> { - // 清除当前历史 - self.history.remove_history(chat_id); - - // 先设置当前话题（set_history 需要这个） + // 设置当前 topic（UI 状态） self.history.set_chat_topic(chat_id, topic_id.to_string()); - // 加载新话题的历史（按 session_id 过滤，排除子智能体消息） + // 加载新 topic 的历史到内存（按 topic_id 键化） let session_id = self.persistent_session_id(chat_id); let messages = self .store .load_messages_for_topic(topic_id, Some(&session_id)) .map_err(|e| AgentError::Other(format!("load topic messages error: {}", e)))?; - self.history.set_history(chat_id, messages); + self.history.set_history(topic_id, messages); tracing::info!( topic_id = %topic_id, @@ -374,32 +367,14 @@ impl Session { self.history.ensure_persistent_session(chat_id) } - pub fn ensure_chat_loaded(&mut self, chat_id: &str) -> Result<(), AgentError> { - // 检查历史是否存在且对应正确的话题 - // 先获取 topic 信息并转换为 owned String，避免借用冲突 - let current_topic: Option = self.history.chat_topic(chat_id).map(|s| s.to_string()); - let stored_topic = self.history.history_topic(chat_id); - - if self.chat_history_exists(chat_id) { - // 如果历史已存在，但话题不匹配，需要重新加载 - if current_topic.as_deref() != stored_topic { - tracing::info!( - chat_id = %chat_id, - current_topic = ?current_topic, - stored_topic = ?stored_topic, - "Topic changed, reloading history" - ); - self.reload_chat_history(chat_id)?; - } - return Ok(()); - } - - // 历史不存在，按 topic 加载（如果设置了 topic） - self.history.ensure_chat_loaded(chat_id, current_topic.as_deref()) - } - - fn chat_history_exists(&self, chat_id: &str) -> bool { - self.history.get_history(chat_id).is_some() + /// 确保指定 topic 的历史已加载到内存。 + /// 按 topic_id 键化查找，已存在则直接返回，否则从 DB 加载。 + pub fn ensure_chat_loaded( + &mut self, + chat_id: &str, + topic_id: Option<&str>, + ) -> Result<(), AgentError> { + self.history.ensure_chat_loaded(chat_id, topic_id) } pub fn ensure_agent_prompt_before_user_message( @@ -410,27 +385,31 @@ impl Session { .ensure_agent_prompt_before_user_message(chat_id) } - /// 获取或创建指定 chat_id 的会话历史 - pub fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec { - self.history.get_or_create_history(chat_id) + /// 获取或创建指定 topic_id 的会话历史 + pub fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec { + self.history.get_or_create_history(topic_id) } - /// 获取指定 chat_id 的会话历史（不创建） - pub fn get_history(&self, chat_id: &str) -> Option<&Vec> { - self.history.get_history(chat_id) + /// 获取指定 topic_id 的会话历史（不创建） + pub fn get_history(&self, topic_id: &str) -> Option<&Vec> { + self.history.get_history(topic_id) } - /// 使用完整消息追加到历史 - pub fn add_message(&mut self, chat_id: &str, message: ChatMessage) { - self.history.add_message(chat_id, message); + /// 使用完整消息追加到指定 topic 的历史 + pub fn add_message(&mut self, topic_id: &str, message: ChatMessage) { + self.history.add_message(topic_id, message); } - pub fn remove_history(&mut self, chat_id: &str) { - self.history.remove_history(chat_id); + pub fn remove_history(&mut self, topic_id: &str) { + self.history.remove_history(topic_id); } - pub fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { - self.history.clear_chat_history(chat_id) + pub fn clear_chat_history( + &mut self, + chat_id: &str, + topic_id: Option<&str>, + ) -> Result<(), AgentError> { + self.history.clear_chat_history(chat_id, topic_id) } /// 将消息写入内存与持久化层。 @@ -457,7 +436,9 @@ impl Session { // 当用户已切换到新 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); + if let Some(ref tid) = topic_id { + self.add_message(tid, message); + } } else { tracing::info!( chat_id = %chat_id, @@ -479,13 +460,13 @@ impl Session { pub fn append_persisted_messages( &mut self, - chat_id: &str, + topic_id: &str, messages: I, ) -> Result<(), AgentError> where I: IntoIterator, { - self.history.append_persisted_messages(chat_id, messages) + self.history.append_persisted_messages(topic_id, messages) } /// 将消息保存到指定话题（直接写入数据库，不更新内存历史） @@ -507,32 +488,32 @@ impl Session { } #[cfg(test)] - fn latest_user_message_id(&self, chat_id: &str) -> Option<&str> { - self.latest_user_message(chat_id) + fn latest_user_message_id(&self, topic_id: &str) -> Option<&str> { + self.latest_user_message(topic_id) .map(|message| message.id.as_str()) } #[cfg(test)] - fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> { - self.history.latest_user_message(chat_id) + fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> { + self.history.latest_user_message(topic_id) } #[cfg(test)] - fn is_latest_user_message(&self, chat_id: &str, message_id: &str) -> bool { - self.latest_user_message_id(chat_id) + fn is_latest_user_message(&self, topic_id: &str, message_id: &str) -> bool { + self.latest_user_message_id(topic_id) .map(|current_id| current_id == message_id) .unwrap_or(false) } - pub(crate) fn matches_current_user_turn(&self, chat_id: &str, message: &ChatMessage) -> bool { - self.history.matches_current_user_turn(chat_id, message) + pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool { + self.history.matches_current_user_turn(topic_id, message) } pub(crate) fn stale_result_diagnostics( &self, - chat_id: &str, + topic_id: &str, ) -> (Option<&str>, Option, bool, usize) { - self.history.stale_result_diagnostics(chat_id) + self.history.stale_result_diagnostics(topic_id) } /// 清除所有历史 @@ -561,20 +542,13 @@ impl Session { self.history.topic_serial_lock(topic_id) } - pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { - // 如果当前有 topic，加载该 topic 的消息（按 session_id 过滤，排除子智能体消息） - if let Some(topic_id) = self.history.chat_topic(chat_id) { - let session_id = self.persistent_session_id(chat_id); - let messages = self - .store - .load_messages_for_topic(topic_id, Some(&session_id)) - .map_err(|e| AgentError::Other(format!("load topic messages error: {}", e)))?; - self.history.set_history(chat_id, messages); - } else { - // 否则加载 session 的所有消息 - self.history.reload_chat_history(chat_id)?; - } - Ok(()) + /// 按 topic_id 从 DB 重新加载历史到内存 + pub(crate) fn reload_topic_history( + &mut self, + chat_id: &str, + topic_id: &str, + ) -> Result<(), AgentError> { + self.history.reload_topic_history(chat_id, topic_id) } pub(crate) fn store(&self) -> Arc { @@ -997,7 +971,7 @@ mod tests { user_tx, tools, skills, - store, + store.clone(), 100, Arc::new(SubagentRuntime::from_config(Default::default())), ) @@ -1005,19 +979,22 @@ mod tests { .unwrap(); session.ensure_persistent_session("chat-1").unwrap(); - session.ensure_chat_loaded("chat-1").unwrap(); + let session_id = session.persistent_session_id("chat-1"); + let topic = store.create_topic(&session_id, "test topic", None).unwrap(); + let topic_id = topic.id.clone(); + session.switch_topic("chat-1", &topic_id).unwrap(); let first = session.create_user_message("first", Vec::new()); let first_id = first.id.clone(); - session.append_persisted_message("chat-1", None, first).unwrap(); - assert!(session.is_latest_user_message("chat-1", &first_id)); + session.append_persisted_message("chat-1", Some(&topic_id), first).unwrap(); + assert!(session.is_latest_user_message(&topic_id, &first_id)); let second = session.create_user_message("second", Vec::new()); let second_id = second.id.clone(); - session.append_persisted_message("chat-1", None, second).unwrap(); + session.append_persisted_message("chat-1", Some(&topic_id), second).unwrap(); - assert!(!session.is_latest_user_message("chat-1", &first_id)); - assert!(session.is_latest_user_message("chat-1", &second_id)); + assert!(!session.is_latest_user_message(&topic_id, &first_id)); + assert!(session.is_latest_user_message(&topic_id, &second_id)); } #[tokio::test] @@ -1054,46 +1031,37 @@ mod tests { .unwrap(); session.ensure_persistent_session("chat-1").unwrap(); - session.ensure_chat_loaded("chat-1").unwrap(); + let session_id = session.persistent_session_id("chat-1"); + let topic = store.create_topic(&session_id, "test topic", None).unwrap(); + let topic_id = topic.id.clone(); + session.switch_topic("chat-1", &topic_id).unwrap(); let first = session.create_user_message("first", Vec::new()); let first_id = first.id.clone(); - session.append_persisted_message("chat-1", None, first).unwrap(); + session.append_persisted_message("chat-1", Some(&topic_id), first).unwrap(); session - .append_persisted_message("chat-1", None, ChatMessage::assistant("answer-1")) + .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-1")) .unwrap(); let second = session.create_user_message("second", Vec::new()); session - .append_persisted_message("chat-1", None, second.clone()) + .append_persisted_message("chat-1", Some(&topic_id), second.clone()) .unwrap(); session - .append_persisted_message("chat-1", None, ChatMessage::assistant("answer-2")) + .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-2")) .unwrap(); - let session_id = session.persistent_session_id("chat-1"); - let snapshot_end_seq = store - .get_session(&session_id) - .unwrap() - .unwrap() - .message_count; - let preserved_messages = session.get_history("chat-1").unwrap().clone(); + let preserved_messages = session.get_history(&topic_id).unwrap().clone(); store - .compact_active_history( - &session_id, - snapshot_end_seq, - &[], - &ChatMessage::system("[Compressed History]\n\nsummary"), - &preserved_messages, - ) + .replace_topic_history(&session_id, &topic_id, &preserved_messages) .unwrap(); - session.reload_chat_history("chat-1").unwrap(); + session.reload_topic_history("chat-1", &topic_id).unwrap(); - assert!(!session.is_latest_user_message("chat-1", &first_id)); - assert!(!session.is_latest_user_message("chat-1", &second.id)); - assert!(session.matches_current_user_turn("chat-1", &second)); + assert!(!session.is_latest_user_message(&topic_id, &first_id)); + assert!(session.is_latest_user_message(&topic_id, &second.id)); + assert!(session.matches_current_user_turn(&topic_id, &second)); } async fn start_mock_openai_server() -> String { @@ -2081,7 +2049,7 @@ mod tests { .unwrap(); session.ensure_persistent_session("chat-1").unwrap(); - session.ensure_chat_loaded("chat-1").unwrap(); + session.switch_topic("chat-1", "chat-1").unwrap(); let history = session.get_history("chat-1").unwrap(); // 新设计：系统提示词不再持久化到历史记录，而是每次请求时动态注入 @@ -2122,11 +2090,14 @@ mod tests { .unwrap(); session.ensure_persistent_session("chat-1").unwrap(); - session.ensure_chat_loaded("chat-1").unwrap(); + let session_id = session.persistent_session_id("chat-1"); + let topic = store.create_topic(&session_id, "test topic", None).unwrap(); + let topic_id = topic.id.clone(); + session.switch_topic("chat-1", &topic_id).unwrap(); for turn in 0..100 { session - .append_persisted_message("chat-1", None, ChatMessage::user(format!("user-{turn}"))) + .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::user(format!("user-{turn}"))) .unwrap(); } @@ -2135,7 +2106,7 @@ mod tests { .unwrap(); // 新设计：系统提示词不再持久化到历史记录 - let history = session.get_history("chat-1").unwrap(); + let history = session.get_history(&topic_id).unwrap(); let user_messages = history .iter() .filter(|message| message.role == "user") @@ -2154,7 +2125,7 @@ mod tests { session .ensure_agent_prompt_before_user_message("chat-1") .unwrap(); - let history = session.get_history("chat-1").unwrap(); + let history = session.get_history(&topic_id).unwrap(); let user_messages = history .iter() .filter(|message| message.role == "user") @@ -2196,11 +2167,14 @@ mod tests { .unwrap(); session.ensure_persistent_session("chat-1").unwrap(); - session.ensure_chat_loaded("chat-1").unwrap(); + let session_id = session.persistent_session_id("chat-1"); + let topic = store.create_topic(&session_id, "test topic", None).unwrap(); + let topic_id = topic.id.clone(); + session.switch_topic("chat-1", &topic_id).unwrap(); for turn in 0..100 { session - .append_persisted_message("chat-1", None, ChatMessage::user(format!("user-{turn}"))) + .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::user(format!("user-{turn}"))) .unwrap(); } @@ -2209,7 +2183,7 @@ mod tests { .unwrap(); // 新设计：系统提示词不再持久化到历史记录 - let history = session.get_history("chat-1").unwrap(); + let history = session.get_history(&topic_id).unwrap(); let user_messages = history .iter() .filter(|message| message.role == "user") diff --git a/src/gateway/session_history.rs b/src/gateway/session_history.rs index 8985f85..6ebf922 100644 --- a/src/gateway/session_history.rs +++ b/src/gateway/session_history.rs @@ -17,9 +17,12 @@ fn preview_text(content: &str, max_chars: usize) -> String { pub(crate) struct SessionHistory { channel_name: String, - chat_histories: HashMap>, - chat_topic_ids: HashMap, // 每个 chat 的当前 topic - history_topic_ids: HashMap, // 每个 chat 的历史所对应的话题 + /// 按 topic_id 键化的内存历史缓存。 + /// 不同 topic 的历史独立存储，互不干扰，支持多话题并发执行。 + topic_histories: HashMap>, + /// UI 状态：每个 chat 当前活跃的 topic（按 chat_id 键）。 + chat_topic_ids: HashMap, + /// 正在压缩中的 topic_id 集合 compression_in_flight: HashSet, /// 按 topic_id 的串行化锁。 /// 同一 topic 的消息处理（agent loop + 压缩）必须串行执行， @@ -38,9 +41,8 @@ impl SessionHistory { ) -> Self { Self { channel_name: channel_name.into(), - chat_histories: HashMap::new(), + topic_histories: HashMap::new(), chat_topic_ids: HashMap::new(), - history_topic_ids: HashMap::new(), compression_in_flight: HashSet::new(), topic_serial_locks: HashMap::new(), conversations, @@ -71,38 +73,36 @@ impl SessionHistory { .map_err(|err| AgentError::Other(format!("session persistence error: {}", err))) } + /// 确保指定 topic 的历史已加载到内存。 + /// 按 topic_id 键化查找，如果已存在则直接返回，否则从 DB 加载。 pub(crate) fn ensure_chat_loaded( &mut self, chat_id: &str, topic_id: Option<&str>, ) -> Result<(), AgentError> { - if self.chat_histories.contains_key(chat_id) { + let Some(tid) = topic_id else { + return Ok(()); + }; + if self.topic_histories.contains_key(tid) { return Ok(()); } - // 如果提供了 topic_id，按 topic 加载；否则按 session 加载 - let mut history = if let Some(tid) = topic_id { - let sid = self.persistent_session_id(chat_id); - self.conversations - .load_messages_for_topic(tid, Some(&sid)) - .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))? - } else { - self.conversations - .load_messages(&self.persistent_session_id(chat_id)) - .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))? - }; + let sid = self.persistent_session_id(chat_id); + let mut history = self + .conversations + .load_messages_for_topic(tid, Some(&sid)) + .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?; - // 清理 DB 加载的历史中可能存在的不完整 tool_call 序列 let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history); if removed > 0 { tracing::warn!( - chat_id = %chat_id, + topic_id = %tid, removed_count = removed, "Sanitized incomplete tool_call sequences on history load" ); } - self.chat_histories.insert(chat_id.to_string(), history); + self.topic_histories.insert(tid.to_string(), history); Ok(()) } @@ -114,58 +114,56 @@ impl SessionHistory { Ok(()) } - pub(crate) fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec { - self.chat_histories.entry(chat_id.to_string()).or_default() + pub(crate) fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec { + self.topic_histories.entry(topic_id.to_string()).or_default() } - pub(crate) fn get_history(&self, chat_id: &str) -> Option<&Vec> { - self.chat_histories.get(chat_id) + pub(crate) fn get_history(&self, topic_id: &str) -> Option<&Vec> { + self.topic_histories.get(topic_id) } - pub(crate) fn set_history(&mut self, chat_id: &str, history: Vec) { - self.chat_histories.insert(chat_id.to_string(), history); - // 记录历史对应的话题（当前设置的话题） - if let Some(topic_id) = self.chat_topic_ids.get(chat_id) { - self.history_topic_ids.insert(chat_id.to_string(), topic_id.clone()); - } + pub(crate) fn set_history(&mut self, topic_id: &str, history: Vec) { + self.topic_histories.insert(topic_id.to_string(), history); } - /// 获取指定 chat 的历史所对应的话题 - pub(crate) fn history_topic(&self, chat_id: &str) -> Option<&str> { - self.history_topic_ids.get(chat_id).map(|s| s.as_str()) - } - - /// 设置指定 chat 的当前 topic + /// 设置指定 chat 的当前 topic（UI 状态） pub(crate) fn set_chat_topic(&mut self, chat_id: &str, topic_id: String) { self.chat_topic_ids.insert(chat_id.to_string(), topic_id); } - /// 获取指定 chat 的当前 topic + /// 获取指定 chat 的当前 topic（UI 状态） pub(crate) fn chat_topic(&self, chat_id: &str) -> Option<&str> { self.chat_topic_ids.get(chat_id).map(|s| s.as_str()) } - /// 清除指定 chat 的 topic + /// 清除指定 chat 的 topic（UI 状态） pub(crate) fn clear_chat_topic(&mut self, chat_id: &str) { self.chat_topic_ids.remove(chat_id); } - pub(crate) fn add_message(&mut self, chat_id: &str, message: ChatMessage) { - self.get_or_create_history(chat_id).push(message); + pub(crate) fn add_message(&mut self, topic_id: &str, message: ChatMessage) { + self.get_or_create_history(topic_id).push(message); } - pub(crate) fn remove_history(&mut self, chat_id: &str) { - self.chat_histories.remove(chat_id); - self.compression_in_flight.remove(chat_id); - self.history_topic_ids.remove(chat_id); + pub(crate) fn remove_history(&mut self, topic_id: &str) { + self.topic_histories.remove(topic_id); + self.compression_in_flight.remove(topic_id); } - pub(crate) fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { - if let Some(history) = self.chat_histories.get_mut(chat_id) { - let len = history.len(); - history.clear(); - #[cfg(debug_assertions)] - tracing::debug!(chat_id = %chat_id, previous_len = len, "Chat history cleared"); + /// 清空指定 chat/topic 的内存历史和 DB 消息。 + /// 内存按 topic_id 清，DB 按 session_id 清（保留原行为以兼容无 topic 场景）。 + pub(crate) fn clear_chat_history( + &mut self, + chat_id: &str, + topic_id: Option<&str>, + ) -> Result<(), AgentError> { + if let Some(tid) = topic_id { + if let Some(history) = self.topic_histories.get_mut(tid) { + let len = history.len(); + history.clear(); + #[cfg(debug_assertions)] + tracing::debug!(topic_id = %tid, previous_len = len, "Topic history cleared"); + } } self.conversations @@ -175,7 +173,7 @@ impl SessionHistory { pub(crate) fn append_persisted_messages( &mut self, - chat_id: &str, + topic_id: &str, messages: I, ) -> Result<(), AgentError> where @@ -186,13 +184,11 @@ impl SessionHistory { return Ok(()); } - // 在追加新消息前，先清理内存历史中的不完整 tool_call 序列 - // 这防止脏数据（如取消时产生的孤立 assistant(tool_calls)）在内存历史中累积 - if let Some(history) = self.chat_histories.get_mut(chat_id) { + if let Some(history) = self.topic_histories.get_mut(topic_id) { let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(history); if removed > 0 { tracing::warn!( - chat_id = %chat_id, + topic_id = %topic_id, removed_count = removed, "Sanitized in-memory history before appending persisted messages" ); @@ -200,7 +196,7 @@ impl SessionHistory { } for message in messages { - self.add_message(chat_id, message); + self.add_message(topic_id, message); } Ok(()) } @@ -219,13 +215,13 @@ impl SessionHistory { Ok(()) } - pub(crate) fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> { - self.get_history(chat_id) + pub(crate) fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> { + self.get_history(topic_id) .and_then(|history| history.iter().rev().find(|message| message.role == "user")) } - pub(crate) fn matches_current_user_turn(&self, chat_id: &str, message: &ChatMessage) -> bool { - self.latest_user_message(chat_id) + pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool { + self.latest_user_message(topic_id) .map(|current| { current.id == message.id || (current.content == message.content @@ -237,14 +233,14 @@ impl SessionHistory { pub(crate) fn stale_result_diagnostics( &self, - chat_id: &str, + topic_id: &str, ) -> (Option<&str>, Option, bool, usize) { - let latest_user = self.latest_user_message(chat_id); + let latest_user = self.latest_user_message(topic_id); let latest_user_id = latest_user.map(|message| message.id.as_str()); let latest_user_preview = latest_user.map(|message| preview_text(&message.content, 80)); - let compression_in_flight = self.compression_in_flight.contains(chat_id); + let compression_in_flight = self.compression_in_flight.contains(topic_id); let history_len = self - .get_history(chat_id) + .get_history(topic_id) .map(|history| history.len()) .unwrap_or(0); @@ -256,31 +252,29 @@ impl SessionHistory { ) } + /// 清空所有内存历史（主要用于测试全局重置）。 + /// 不遍历清 DB，生产环境如需清 DB 应由调用方显式调用。 pub(crate) fn clear_all_history(&mut self) -> Result<(), AgentError> { - let chat_ids: Vec = self.chat_histories.keys().cloned().collect(); - let total: usize = self.chat_histories.values().map(|h| h.len()).sum(); - self.chat_histories.clear(); + let total: usize = self.topic_histories.values().map(|h| h.len()).sum(); + self.topic_histories.clear(); self.compression_in_flight.clear(); #[cfg(debug_assertions)] - tracing::debug!(previous_total = total, "All chat histories cleared"); - - for chat_id in chat_ids { - self.conversations - .clear_messages(&self.persistent_session_id(&chat_id)) - .map_err(|err| { - AgentError::Other(format!("clear history persistence error: {}", err)) - })?; - } - + tracing::debug!(previous_total = total, "All topic histories cleared"); Ok(()) } - pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { + /// 按 topic_id 从 DB 重新加载历史到内存 + pub(crate) fn reload_topic_history( + &mut self, + chat_id: &str, + topic_id: &str, + ) -> Result<(), AgentError> { + let sid = self.persistent_session_id(chat_id); let history = self .conversations - .load_messages(&self.persistent_session_id(chat_id)) + .load_messages_for_topic(topic_id, Some(&sid)) .map_err(|err| AgentError::Other(format!("session history reload error: {}", err)))?; - self.chat_histories.insert(chat_id.to_string(), history); + self.topic_histories.insert(topic_id.to_string(), history); Ok(()) } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index ffa22fb..4ed6816 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -875,6 +875,74 @@ impl SessionStore { Ok(()) } + /// Replace the entire history for a specific topic. + /// + /// Deletes only messages belonging to the given topic_id (preserving + /// other topics' messages), then inserts the new messages with topic_id + /// set correctly. Used by the compressor when it has produced a + /// complete, validated message list for a single topic. + /// + /// Seq numbers continue from the current session-wide max (not reset to + /// 1) so we don't collide with other topics' messages. Gaps in seq + /// (from the deleted old messages) are harmless — per-topic loading + /// orders by seq and gaps don't affect ordering. + pub fn replace_topic_history( + &self, + session_id: &str, + topic_id: &str, + messages: &[ChatMessage], + ) -> Result<(), StorageError> { + let conn = self.pool.get()?; + let tx = conn.unchecked_transaction()?; + let now = current_timestamp(); + + // Delete only messages belonging to this topic — other topics' + // messages are preserved (the pre-existing `replace_active_history` + // clobbered the entire session, which broke multi-topic isolation). + tx.execute( + "DELETE FROM messages WHERE session_id = ?1 AND topic_id = ?2", + params![session_id, topic_id], + )?; + + // Continue seq from the session-wide max so we don't violate + // UNIQUE(session_id, seq). Other topics' messages keep their seqs. + let start_seq: i64 = tx.query_row( + "SELECT COALESCE(MAX(seq), 0) + 1 FROM messages WHERE session_id = ?1", + params![session_id], + |row| row.get(0), + )?; + + for (i, message) in messages.iter().enumerate() { + let seq = start_seq + i as i64; + insert_message_with_topic_seq(&tx, session_id, topic_id, seq, message)?; + } + + // Update this topic's message_count and timestamps. + tx.execute( + "UPDATE topics SET message_count = ?2, last_active_at = ?3, updated_at = ?3 WHERE id = ?1", + params![topic_id, messages.len() as i64, now], + )?; + + // Recompute session-wide counts from the messages table so they stay + // consistent after a partial replacement (we only touched one topic, + // so we can't just set the session count to `messages.len()`). + let (total_count, user_turn_count): (i64, i64) = tx.query_row( + "SELECT COUNT(*), COALESCE(SUM(CASE WHEN role = 'user' THEN 1 ELSE 0 END), 0) \ + FROM messages WHERE session_id = ?1", + params![session_id], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + tx.execute( + "UPDATE sessions SET message_count = ?2, user_turn_count = ?3, \ + updated_at = ?4, last_active_at = ?4, archived_at = NULL \ + WHERE id = ?1 AND deleted_at IS NULL", + params![session_id, total_count, user_turn_count, now], + )?; + + tx.commit()?; + Ok(()) + } + pub fn mark_agent_prompt_reinjected(&self, session_id: &str) -> Result<(), StorageError> { let now = current_timestamp(); let conn = self.pool.get()?; @@ -1667,6 +1735,51 @@ fn insert_message_with_seq( Ok(()) } +/// Insert a message with an explicit `topic_id` and `seq`. +/// +/// Used by `replace_topic_history` to insert compressed messages while +/// preserving topic association (the plain `insert_message_with_seq` would +/// set topic_id to NULL). +fn insert_message_with_topic_seq( + conn: &rusqlite::Transaction<'_>, + session_id: &str, + topic_id: &str, + seq: i64, + message: &ChatMessage, +) -> Result<(), StorageError> { + let media_refs_json = serde_json::to_string(&message.media_refs)?; + let tool_calls_json = message + .tool_calls + .as_ref() + .map(serde_json::to_string) + .transpose()?; + conn.execute( + " + INSERT INTO messages ( + id, session_id, topic_id, seq, role, content, + system_context, reasoning_content, media_refs_json, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + ", + params![ + message.id, + session_id, + topic_id, + seq, + message.role, + message.content, + message.system_context, + message.reasoning_content, + media_refs_json, + message.tool_call_id, + message.tool_name, + tool_calls_json, + message.tool_duration_ms.map(|v| v as i64), + message.timestamp, + ], + )?; + Ok(()) +} + fn clone_message_for_compaction(message: &ChatMessage, timestamp: i64) -> ChatMessage { ChatMessage { id: uuid::Uuid::new_v4().to_string(), diff --git a/src/storage/ports.rs b/src/storage/ports.rs index 0a7aeba..84ee6df 100644 --- a/src/storage/ports.rs +++ b/src/storage/ports.rs @@ -63,6 +63,17 @@ pub trait ConversationRepository: Send + Sync + 'static { session_id: &str, messages: &[ChatMessage], ) -> Result<(), StorageError>; + + /// Replace the entire history for a specific topic. + /// Deletes only messages belonging to the given topic_id, then inserts + /// the new messages with topic_id set correctly. Used by compressor when + /// it has produced a complete, validated message list for a single topic. + fn replace_topic_history( + &self, + session_id: &str, + topic_id: &str, + messages: &[ChatMessage], + ) -> Result<(), StorageError>; } pub trait PromptInjectionRepository: Send + Sync + 'static { @@ -252,6 +263,15 @@ impl ConversationRepository for super::SessionStore { ) -> Result<(), StorageError> { super::SessionStore::replace_active_history(self, session_id, messages) } + + fn replace_topic_history( + &self, + session_id: &str, + topic_id: &str, + messages: &[ChatMessage], + ) -> Result<(), StorageError> { + super::SessionStore::replace_topic_history(self, session_id, topic_id, messages) + } } impl PromptInjectionRepository for super::SessionStore {