From 7652bb16e2e0270f88c10f2f762d63cce7f52c27 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 7 Jul 2026 14:13:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20get=5Fcurrent/save=5Ftopic=20?= =?UTF-8?q?=E7=9B=B4=E8=AF=BB=20DB=20=E5=8A=A0=E8=BD=BD=E6=B6=88=E6=81=AF,?= =?UTF-8?q?=E7=A7=BB=E9=99=A4=20SessionManager=20=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_current.rs 和 save_topic.rs 改为直读 DB(load_messages_for_topic),不依赖 SessionManager 内存,重启后也能正确获取历史。 mod.rs: 删除 get_messages_from_session 辅助函数。 processor.rs/ws.rs: 移除 with_session_manager 调用。 --- src/command/handlers/get_current.rs | 20 +++++------------ src/command/handlers/mod.rs | 35 ----------------------------- src/command/handlers/save_topic.rs | 29 +++++++++++------------- src/gateway/processor.rs | 3 +-- src/gateway/ws.rs | 3 +-- 5 files changed, 20 insertions(+), 70 deletions(-) diff --git a/src/command/handlers/get_current.rs b/src/command/handlers/get_current.rs index b1d1884..baf5145 100644 --- a/src/command/handlers/get_current.rs +++ b/src/command/handlers/get_current.rs @@ -2,10 +2,8 @@ use crate::agent::context_compressor::estimate_tokens; use crate::agent::{SystemPromptContext, SystemPromptProvider}; use crate::command::context::CommandContext; use crate::command::handler::{CommandHandler, CommandMetadata}; -use crate::command::handlers::get_messages_from_session; use crate::command::response::{CommandError, CommandResponse, MessageKind}; use crate::command::Command; -use crate::gateway::session::SessionManager; use crate::storage::SessionStore; use async_trait::async_trait; use std::sync::Arc; @@ -13,7 +11,6 @@ use std::sync::Arc; /// 获取当前话题命令处理器 pub struct GetCurrentSessionCommandHandler { store: Arc, - session_manager: Option, system_prompt_provider: Option>, } @@ -21,16 +18,10 @@ impl GetCurrentSessionCommandHandler { pub fn new(store: Arc) -> Self { Self { store, - session_manager: None, system_prompt_provider: None, } } - pub fn with_session_manager(mut self, session_manager: SessionManager) -> Self { - self.session_manager = Some(session_manager); - self - } - pub fn with_system_prompt_provider(mut self, provider: Arc) -> Self { self.system_prompt_provider = Some(provider); self @@ -79,12 +70,11 @@ async fn handle_get_current_session( .map_err(|e| CommandError::new("GET_TOPIC_ERROR", e.to_string()))? .ok_or_else(|| CommandError::new("TOPIC_NOT_FOUND", format!("Topic not found: {}", topic_id)))?; - // Load messages from session memory - let messages = get_messages_from_session( - &handler.session_manager, - &ctx.channel_name, - chat_id, - ).await?; + // 直读 DB 按话题加载消息（不依赖 SessionManager 内存，重启后也能正确获取历史） + let messages = handler + .store + .load_messages_for_topic(topic_id, Some(&topic.session_id)) + .map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?; let actual_message_count = messages.len(); let message_tokens = estimate_tokens(&messages); diff --git a/src/command/handlers/mod.rs b/src/command/handlers/mod.rs index d11d767..1e8e8cf 100644 --- a/src/command/handlers/mod.rs +++ b/src/command/handlers/mod.rs @@ -26,38 +26,3 @@ pub use save_session::{ generate_subagent_tasks_markdown, load_subagent_data, SubagentTaskData, }; -use crate::bus::ChatMessage; -use crate::command::response::CommandError; -use crate::gateway::session::SessionManager; - -/// 从 Session 内存获取消息历史（供命令使用） -pub async fn get_messages_from_session( - session_manager: &Option, - channel_name: &str, - chat_id: &str, -) -> Result, CommandError> { - let session_manager = session_manager.as_ref().ok_or_else(|| { - CommandError::new( - "SESSION_MANAGER_NOT_SET", - "Session manager not configured".to_string(), - ) - })?; - - match session_manager.get(channel_name).await { - Some(session) => { - let guard = session.lock().await; - Ok(guard - .get_history(chat_id) - .map(|m| m.clone()) - .unwrap_or_default()) - } - None => { - tracing::warn!( - channel = %channel_name, - chat_id = %chat_id, - "No in-memory session, returning empty message list" - ); - Ok(Vec::new()) - } - } -} diff --git a/src/command/handlers/save_topic.rs b/src/command/handlers/save_topic.rs index 003e088..e70ff14 100644 --- a/src/command/handlers/save_topic.rs +++ b/src/command/handlers/save_topic.rs @@ -5,11 +5,10 @@ use crate::command::handler::{CommandHandler, CommandMetadata}; use crate::command::handlers::{ escape_yaml_string, format_timestamp, generate_messages_markdown, generate_subagent_tasks_markdown, generate_system_prompt_markdown, - get_messages_from_session, load_subagent_data, SubagentTaskData, + load_subagent_data, SubagentTaskData, }; use crate::command::response::{CommandError, CommandResponse, MessageKind}; use crate::command::Command; -use crate::gateway::session::SessionManager; use crate::storage::{SessionStore, TopicRecord}; use crate::tools::task::repository::TaskRepository; use async_trait::async_trait; @@ -175,7 +174,6 @@ pub struct SaveTopicCommandHandler { store: Arc, task_repository: Arc, system_prompt_provider: Arc, - session_manager: Option, } impl SaveTopicCommandHandler { @@ -188,14 +186,8 @@ impl SaveTopicCommandHandler { store, task_repository, system_prompt_provider, - session_manager: None, } } - - pub fn with_session_manager(mut self, session_manager: SessionManager) -> Self { - self.session_manager = Some(session_manager); - self - } } #[async_trait] @@ -252,14 +244,19 @@ async fn handle_save_topic( tracing::debug!(topic_id = %topic_id, chat_id = %chat_id, "Attempting to save topic"); - // 从 Session 获取当前 history（包含已压缩的消息） - let messages = get_messages_from_session( - &handler.session_manager, - &ctx.channel_name, - chat_id, - ).await?; + // 直读 DB 按话题加载消息（不依赖 SessionManager 内存，重启后也能正确获取历史） + let topic_record = handler + .store + .get_topic(topic_id) + .map_err(|e| CommandError::new("GET_TOPIC_ERROR", e.to_string()))? + .ok_or_else(|| CommandError::new("TOPIC_NOT_FOUND", format!("Topic not found: {}", topic_id)))?; - tracing::debug!(message_count = messages.len(), "Got messages from session"); + let messages = handler + .store + .load_messages_for_topic(topic_id, Some(&topic_record.session_id)) + .map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?; + + tracing::debug!(message_count = messages.len(), "Loaded messages from DB for topic"); // 调用保存函数 let output_path = save_topic_to_file( diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index b86765a..13daab7 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -78,7 +78,6 @@ impl InboundProcessor { // 注册 get_current 处理器 command_router.register(Box::new( GetCurrentSessionCommandHandler::new(store.clone()) - .with_session_manager(session_manager.clone()) .with_system_prompt_provider(system_prompt_provider.clone()) )); @@ -97,7 +96,7 @@ impl InboundProcessor { store.clone(), session_manager.task_repository(), system_prompt_provider, - ).with_session_manager(session_manager.clone()))); + ))); // 注册 delete_topic 处理器 command_router.register(Box::new( diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index 0336a02..9cd5569 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -429,7 +429,6 @@ async fn handle_inbound( // 注册 get_current 处理器 router.register(Box::new( GetCurrentSessionCommandHandler::new(store.clone()) - .with_session_manager(state.session_manager.clone()) .with_system_prompt_provider(system_prompt_provider.clone()), )); // 注册 load_topic 处理器 @@ -449,7 +448,7 @@ async fn handle_inbound( store.clone(), state.task_repository.clone(), system_prompt_provider.clone(), - ).with_session_manager(state.session_manager.clone()))); + ))); // 注册 delete_topic 处理器 router.register(Box::new( DeleteTopicCommandHandler::new(store.clone())