ooodc/PicoBot
ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

refactor: get_current/save_topic 直读 DB 加载消息,移除 SessionManager 依赖

Browse Source 
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 调用。
This commit is contained in:
oudecheng 2026-07-07 14:13:31 +08:00
parent 7eecd0b6bb
commit 7652bb16e2
5 changed files with 20 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/command/handlers/get_current.rs
View File

@ -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<SessionStore>,
session_manager: Option<SessionManager>,
system_prompt_provider: Option<Arc<dyn SystemPromptProvider>>,
}
@ -21,16 +18,10 @@ impl GetCurrentSessionCommandHandler {
pub fn new(store: Arc<SessionStore>) -> 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<dyn SystemPromptProvider>) -> 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);

35
src/command/handlers/mod.rs
View File

@ -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<SessionManager>,
channel_name: &str,
chat_id: &str,
) -> Result<Vec<ChatMessage>, 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())
}
}
}

29
src/command/handlers/save_topic.rs
View File

@ -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<SessionStore>,
task_repository: Arc<dyn TaskRepository>,
system_prompt_provider: Arc<dyn SystemPromptProvider>,
session_manager: Option<SessionManager>,
}
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(

3
src/gateway/processor.rs
View File

@ -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(

3
src/gateway/ws.rs
View File

@ -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())