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

fix: 内存历史按 topic_id 键化，新增 replace_topic_history 修复 DB 压缩覆盖

Browse Source 
- 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 时更新内存，防止延迟消息污染新话题历史
This commit is contained in:
oudecheng 2026-07-28 15:32:42 +08:00
parent b042b45ac7
commit 4baa8e7a6b
6 changed files with 365 additions and 257 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/gateway/compaction.rs
View File

@ -13,17 +13,21 @@ use super::session::Session;
/// has already finished by this point there is no response-time impact, and /// has already finished by this point there is no response-time impact, and
/// the synchronous guarantee means the next execution always starts with /// the synchronous guarantee means the next execution always starts with
/// freshly compacted history. /// freshly compacted history.
///
/// 按 topic_id 隔离：压缩只处理指定 topic 的历史，DB 替换也只影响该 topic。
pub(crate) async fn schedule_background_history_compaction( pub(crate) async fn schedule_background_history_compaction(
session: Arc<Mutex<Session>>, session: Arc<Mutex<Session>>,
chat_id: impl Into<String>, chat_id: impl Into<String>,
topic_id: impl Into<String>,
) -> Result<(), AgentError> { ) -> Result<(), AgentError> {
let chat_id = chat_id.into(); let chat_id = chat_id.into();
let topic_id = topic_id.into();
let mut session_guard = session.lock().await; let mut session_guard = session.lock().await;
session_guard.ensure_persistent_session(&chat_id)?; 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(); let compressor = session_guard.compressor().clone();
if !compressor.should_compress(&history) { if !compressor.should_compress(&history) {
@ -36,6 +40,7 @@ pub(crate) async fn schedule_background_history_compaction(
tracing::info!( tracing::info!(
chat_id = %chat_id, chat_id = %chat_id,
topic_id = %topic_id,
msg_count = history.len(), msg_count = history.len(),
"Starting synchronous two-segment compression" "Starting synchronous two-segment compression"
); );
@ -47,19 +52,20 @@ pub(crate) async fn schedule_background_history_compaction(
.compress_two_segment(&history, &provider_config) .compress_two_segment(&history, &provider_config)
.await?; .await?;
// Replace the entire history with the compressed result. // Replace only this topic's history in DB (not the entire session).
// Since we hold the lock, no concurrent modifications can occur. // This avoids clobbering other topics' messages during compaction.
store store
.replace_active_history(&session_id, &compressed) .replace_topic_history(&session_id, &topic_id, &compressed)
.map_err(|e| AgentError::Other(format!("replace_active_history error: {}", e)))?; .map_err(|e| AgentError::Other(format!("replace_topic_history error: {}", e)))?;
tracing::info!( tracing::info!(
chat_id = %chat_id, chat_id = %chat_id,
topic_id = %topic_id,
compressed_msg_count = compressed.len(), compressed_msg_count = compressed.len(),
"Two-segment compression committed" "Two-segment compression committed"
); );
session_guard.reload_chat_history(&chat_id)?; session_guard.reload_topic_history(&chat_id, &topic_id)?;
Ok(()) Ok(())
} }

109
src/gateway/execution.rs
View File

@ -91,13 +91,19 @@ impl AgentExecutionService {
session: &mut Session, session: &mut Session,
request: FinalizeAgentResultRequest<'_>, request: FinalizeAgentResultRequest<'_>,
) -> Result<FinalizedAgentResult, AgentError> { ) -> Result<FinalizedAgentResult, AgentError> {
// 检查是否是最新的用户回合 // 判断是否是最新的用户回合
let is_current_turn = // 直接比较 current_topic(chat_id) 与 original_topic_id
session.matches_current_user_turn(request.chat_id, request.user_message); // 这比"检查内存历史最新消息"更可靠，且天然处理"切走又切回"的 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 { if !is_current_turn {
let (latest_user_id, latest_user_preview, compression_in_flight, history_len) = 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!( tracing::info!(
channel = %request.channel_name, channel = %request.channel_name,
chat_id = %request.chat_id, chat_id = %request.chat_id,
@ -119,36 +125,19 @@ impl AgentExecutionService {
// 将结果消息保存到确定的话题 // 将结果消息保存到确定的话题
if let Some(topic_id) = target_topic_id { if let Some(topic_id) = target_topic_id {
if is_current_turn { if is_current_turn {
// 检查当前活跃 topic 是否仍是 original_topic_id // 话题未切换（current_topic == original_topic_id），安全更新内存历史
let current_tid = session.current_topic(request.chat_id); if let Err(err) = session.append_persisted_messages(
if current_tid.as_deref() == Some(topic_id) { topic_id,
// 话题未切换，安全更新内存历史 request.result.emitted_messages.clone(),
if let Err(err) = session.append_persisted_messages( ) {
request.chat_id, tracing::error!(
request.result.emitted_messages.clone(), error = %err,
) { topic_id = %topic_id,
tracing::error!( "Failed to append messages to session history"
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 { } else {
// stale：只写 DB // 话题已切换，只写 DB 不更新内存（避免污染新话题的历史）
if let Err(err) = session.append_messages_to_topic( if let Err(err) = session.append_messages_to_topic(
request.chat_id, request.chat_id,
topic_id, topic_id,
@ -163,6 +152,7 @@ impl AgentExecutionService {
} }
} else if is_current_turn { } else if is_current_turn {
// 没有话题：直接更新内存历史（append_persisted_messages 会处理持久化） // 没有话题：直接更新内存历史（append_persisted_messages 会处理持久化）
// 无 topic 场景用 chat_id 作为 topic_histories 的回退 key
if let Err(err) = session.append_persisted_messages( if let Err(err) = session.append_persisted_messages(
request.chat_id, request.chat_id,
request.result.emitted_messages.clone(), request.result.emitted_messages.clone(),
@ -232,7 +222,16 @@ impl AgentExecutionService {
let mut session_guard = request.session.lock().await; let mut session_guard = request.session.lock().await;
session_guard.ensure_persistent_session(request.chat_id)?; 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)?; 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)?; enrich_user_content_with_media_refs(request.content, &media_refs)?;
// 先计算 user_message_count（在添加新消息之前） // 先计算 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(); 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); let user_message = session_guard.create_user_message(&enriched_content, media_refs);
session_guard.append_persisted_message( session_guard.append_persisted_message(
request.chat_id, 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)?; session_guard.record_skill_offer(request.chat_id)?;
let mut agent = session_guard.create_agent( let mut agent = session_guard.create_agent(
@ -336,21 +329,21 @@ impl AgentExecutionService {
session_guard.ensure_persistent_session(request.chat_id)?; session_guard.ensure_persistent_session(request.chat_id)?;
// 复用锁前捕获的 topic_id，保证锁键与写入目标一致
let original_topic_id = lock_time_topic_id.clone();
// 如果 fresh_session 为 true，清理历史（内存 + 数据库） // 如果 fresh_session 为 true，清理历史（内存 + 数据库）
if request.fresh_session { 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!( tracing::info!(
chat_id = %request.chat_id, chat_id = %request.chat_id,
"Fresh session enabled, history cleared" "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)?; 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 = let scheduled_system_prompt =
compose_scheduled_task_system_prompt(request.system_prompt); compose_scheduled_task_system_prompt(request.system_prompt);
session_guard.append_persisted_message( session_guard.append_persisted_message(
@ -363,7 +356,8 @@ impl AgentExecutionService {
)?; )?;
// 先计算 user_message_count（在添加新消息之前） // 先计算 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_count = history_before.iter().filter(|m| m.role == "user").count();
let user_message = session_guard.create_user_message(request.prompt, Vec::new()); 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)?; session_guard.record_skill_offer(request.chat_id)?;
let agent = session_guard.create_agent_with_provider_config( let agent = session_guard.create_agent_with_provider_config(
@ -426,7 +420,7 @@ impl AgentExecutionService {
metadata: request.metadata, metadata: request.metadata,
suppress_live_tool_calls: false, suppress_live_tool_calls: false,
execution_kind: "scheduled_task", execution_kind: "scheduled_task",
original_topic_id, original_topic_id: original_topic_id.clone(),
}, },
) )
.await?; .await?;
@ -434,7 +428,8 @@ impl AgentExecutionService {
// 清理内存历史，释放内存（数据库历史保留） // 清理内存历史，释放内存（数据库历史保留）
{ {
let mut session_guard = request.session.lock().await; 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!( tracing::info!(
chat_id = %request.chat_id, chat_id = %request.chat_id,
"Scheduled task completed, memory history released" "Scheduled task completed, memory history released"
@ -452,6 +447,7 @@ impl AgentExecutionService {
let channel_name = request.channel_name.to_string(); let channel_name = request.channel_name.to_string();
let chat_id = request.chat_id.to_string(); let chat_id = request.chat_id.to_string();
let execution_kind = request.execution_kind.to_string(); let execution_kind = request.execution_kind.to_string();
let topic_id = request.original_topic_id.clone();
let finalized_result = { let finalized_result = {
let mut session_guard = session.lock().await; let mut session_guard = session.lock().await;
@ -459,8 +455,13 @@ impl AgentExecutionService {
}; };
if finalized_result.should_schedule_compaction { if finalized_result.should_schedule_compaction {
if let Err(error) = let compaction_topic_id = topic_id.unwrap_or_else(|| chat_id.clone());
schedule_background_history_compaction(session.clone(), chat_id.clone()).await if let Err(error) = schedule_background_history_compaction(
session.clone(),
chat_id.clone(),
compaction_topic_id,
)
.await
{ {
tracing::warn!( tracing::warn!(
channel = %channel_name, channel = %channel_name,

206
src/gateway/session.rs
View File

@ -340,27 +340,20 @@ impl Session {
self.history.chat_topic(chat_id) self.history.chat_topic(chat_id)
} }
/// 获取历史所对应的话题 ID（指定 chat） /// 切换话题 - 设置当前 topic 并加载新话题的历史到内存
pub fn history_topic(&self, chat_id: &str) -> Option<&str> { /// 不同 topic 的历史在 topic_histories 中独立存储，切换不互斥。
self.history.history_topic(chat_id)
}
/// 切换话题 - 清除当前历史并加载新话题的历史
pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> { pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> {
// 清除当前历史 // 设置当前 topic（UI 状态）
self.history.remove_history(chat_id);
// 先设置当前话题（set_history 需要这个）
self.history.set_chat_topic(chat_id, topic_id.to_string()); 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 session_id = self.persistent_session_id(chat_id);
let messages = self let messages = self
.store .store
.load_messages_for_topic(topic_id, Some(&session_id)) .load_messages_for_topic(topic_id, Some(&session_id))
.map_err(|e| AgentError::Other(format!("load topic messages error: {}", e)))?; .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!( tracing::info!(
topic_id = %topic_id, topic_id = %topic_id,
@ -374,32 +367,14 @@ impl Session {
self.history.ensure_persistent_session(chat_id) self.history.ensure_persistent_session(chat_id)
} }
pub fn ensure_chat_loaded(&mut self, chat_id: &str) -> Result<(), AgentError> { /// 确保指定 topic 的历史已加载到内存。
// 检查历史是否存在且对应正确的话题 /// 按 topic_id 键化查找，已存在则直接返回，否则从 DB 加载。
// 先获取 topic 信息并转换为 owned String，避免借用冲突 pub fn ensure_chat_loaded(
let current_topic: Option<String> = self.history.chat_topic(chat_id).map(|s| s.to_string()); &mut self,
let stored_topic = self.history.history_topic(chat_id); chat_id: &str,
topic_id: Option<&str>,
if self.chat_history_exists(chat_id) { ) -> Result<(), AgentError> {
// 如果历史已存在，但话题不匹配，需要重新加载 self.history.ensure_chat_loaded(chat_id, topic_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()
} }
pub fn ensure_agent_prompt_before_user_message( pub fn ensure_agent_prompt_before_user_message(
@ -410,27 +385,31 @@ impl Session {
.ensure_agent_prompt_before_user_message(chat_id) .ensure_agent_prompt_before_user_message(chat_id)
} }
/// 获取或创建指定 chat_id 的会话历史 /// 获取或创建指定 topic_id 的会话历史
pub fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec<ChatMessage> { pub fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec<ChatMessage> {
self.history.get_or_create_history(chat_id) self.history.get_or_create_history(topic_id)
} }
/// 获取指定 chat_id 的会话历史（不创建） /// 获取指定 topic_id 的会话历史（不创建）
pub fn get_history(&self, chat_id: &str) -> Option<&Vec<ChatMessage>> { pub fn get_history(&self, topic_id: &str) -> Option<&Vec<ChatMessage>> {
self.history.get_history(chat_id) self.history.get_history(topic_id)
} }
/// 使用完整消息追加到历史 /// 使用完整消息追加到指定 topic 的历史
pub fn add_message(&mut self, chat_id: &str, message: ChatMessage) { pub fn add_message(&mut self, topic_id: &str, message: ChatMessage) {
self.history.add_message(chat_id, message); self.history.add_message(topic_id, message);
} }
pub fn remove_history(&mut self, chat_id: &str) { pub fn remove_history(&mut self, topic_id: &str) {
self.history.remove_history(chat_id); self.history.remove_history(topic_id);
} }
pub fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { pub fn clear_chat_history(
self.history.clear_chat_history(chat_id) &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 的内存历史。 // 当用户已切换到新 topic 时，旧 topic 的排队消息不应污染新 topic 的内存历史。
let current_chat_topic = self.history.chat_topic(chat_id); let current_chat_topic = self.history.chat_topic(chat_id);
if topic_id.as_deref() == current_chat_topic { 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 { } else {
tracing::info!( tracing::info!(
chat_id = %chat_id, chat_id = %chat_id,
@ -479,13 +460,13 @@ impl Session {
pub fn append_persisted_messages<I>( pub fn append_persisted_messages<I>(
&mut self, &mut self,
chat_id: &str, topic_id: &str,
messages: I, messages: I,
) -> Result<(), AgentError> ) -> Result<(), AgentError>
where where
I: IntoIterator<Item = ChatMessage>, I: IntoIterator<Item = ChatMessage>,
{ {
self.history.append_persisted_messages(chat_id, messages) self.history.append_persisted_messages(topic_id, messages)
} }
/// 将消息保存到指定话题（直接写入数据库，不更新内存历史） /// 将消息保存到指定话题（直接写入数据库，不更新内存历史）
@ -507,32 +488,32 @@ impl Session {
} }
#[cfg(test)] #[cfg(test)]
fn latest_user_message_id(&self, chat_id: &str) -> Option<&str> { fn latest_user_message_id(&self, topic_id: &str) -> Option<&str> {
self.latest_user_message(chat_id) self.latest_user_message(topic_id)
.map(|message| message.id.as_str()) .map(|message| message.id.as_str())
} }
#[cfg(test)] #[cfg(test)]
fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> { fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> {
self.history.latest_user_message(chat_id) self.history.latest_user_message(topic_id)
} }
#[cfg(test)] #[cfg(test)]
fn is_latest_user_message(&self, chat_id: &str, message_id: &str) -> bool { fn is_latest_user_message(&self, topic_id: &str, message_id: &str) -> bool {
self.latest_user_message_id(chat_id) self.latest_user_message_id(topic_id)
.map(|current_id| current_id == message_id) .map(|current_id| current_id == message_id)
.unwrap_or(false) .unwrap_or(false)
} }
pub(crate) fn matches_current_user_turn(&self, chat_id: &str, message: &ChatMessage) -> bool { pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool {
self.history.matches_current_user_turn(chat_id, message) self.history.matches_current_user_turn(topic_id, message)
} }
pub(crate) fn stale_result_diagnostics( pub(crate) fn stale_result_diagnostics(
&self, &self,
chat_id: &str, topic_id: &str,
) -> (Option<&str>, Option<String>, bool, usize) { ) -> (Option<&str>, Option<String>, 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) self.history.topic_serial_lock(topic_id)
} }
pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { /// 按 topic_id 从 DB 重新加载历史到内存
// 如果当前有 topic，加载该 topic 的消息（按 session_id 过滤，排除子智能体消息） pub(crate) fn reload_topic_history(
if let Some(topic_id) = self.history.chat_topic(chat_id) { &mut self,
let session_id = self.persistent_session_id(chat_id); chat_id: &str,
let messages = self topic_id: &str,
.store ) -> Result<(), AgentError> {
.load_messages_for_topic(topic_id, Some(&session_id)) self.history.reload_topic_history(chat_id, topic_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(())
} }
pub(crate) fn store(&self) -> Arc<dyn ConversationRepository> { pub(crate) fn store(&self) -> Arc<dyn ConversationRepository> {
@ -997,7 +971,7 @@ mod tests {
user_tx, user_tx,
tools, tools,
skills, skills,
store, store.clone(),
100, 100,
Arc::new(SubagentRuntime::from_config(Default::default())), Arc::new(SubagentRuntime::from_config(Default::default())),
) )
@ -1005,19 +979,22 @@ mod tests {
.unwrap(); .unwrap();
session.ensure_persistent_session("chat-1").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 = session.create_user_message("first", Vec::new());
let first_id = first.id.clone(); 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();
assert!(session.is_latest_user_message("chat-1", &first_id)); assert!(session.is_latest_user_message(&topic_id, &first_id));
let second = session.create_user_message("second", Vec::new()); let second = session.create_user_message("second", Vec::new());
let second_id = second.id.clone(); 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(&topic_id, &first_id));
assert!(session.is_latest_user_message("chat-1", &second_id)); assert!(session.is_latest_user_message(&topic_id, &second_id));
} }
#[tokio::test] #[tokio::test]
@ -1054,46 +1031,37 @@ mod tests {
.unwrap(); .unwrap();
session.ensure_persistent_session("chat-1").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 = session.create_user_message("first", Vec::new());
let first_id = first.id.clone(); 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 session
.append_persisted_message("chat-1", None, ChatMessage::assistant("answer-1")) .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-1"))
.unwrap(); .unwrap();
let second = session.create_user_message("second", Vec::new()); let second = session.create_user_message("second", Vec::new());
session session
.append_persisted_message("chat-1", None, second.clone()) .append_persisted_message("chat-1", Some(&topic_id), second.clone())
.unwrap(); .unwrap();
session session
.append_persisted_message("chat-1", None, ChatMessage::assistant("answer-2")) .append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-2"))
.unwrap(); .unwrap();
let session_id = session.persistent_session_id("chat-1"); let preserved_messages = session.get_history(&topic_id).unwrap().clone();
let snapshot_end_seq = store
.get_session(&session_id)
.unwrap()
.unwrap()
.message_count;
let preserved_messages = session.get_history("chat-1").unwrap().clone();
store store
.compact_active_history( .replace_topic_history(&session_id, &topic_id, &preserved_messages)
&session_id,
snapshot_end_seq,
&[],
&ChatMessage::system("[Compressed History]\n\nsummary"),
&preserved_messages,
)
.unwrap(); .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(&topic_id, &first_id));
assert!(!session.is_latest_user_message("chat-1", &second.id)); assert!(session.is_latest_user_message(&topic_id, &second.id));
assert!(session.matches_current_user_turn("chat-1", &second)); assert!(session.matches_current_user_turn(&topic_id, &second));
} }
async fn start_mock_openai_server() -> String { async fn start_mock_openai_server() -> String {
@ -2081,7 +2049,7 @@ mod tests {
.unwrap(); .unwrap();
session.ensure_persistent_session("chat-1").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(); let history = session.get_history("chat-1").unwrap();
// 新设计：系统提示词不再持久化到历史记录，而是每次请求时动态注入 // 新设计：系统提示词不再持久化到历史记录，而是每次请求时动态注入
@ -2122,11 +2090,14 @@ mod tests {
.unwrap(); .unwrap();
session.ensure_persistent_session("chat-1").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 { for turn in 0..100 {
session 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(); .unwrap();
} }
@ -2135,7 +2106,7 @@ mod tests {
.unwrap(); .unwrap();
// 新设计：系统提示词不再持久化到历史记录 // 新设计：系统提示词不再持久化到历史记录
let history = session.get_history("chat-1").unwrap(); let history = session.get_history(&topic_id).unwrap();
let user_messages = history let user_messages = history
.iter() .iter()
.filter(|message| message.role == "user") .filter(|message| message.role == "user")
@ -2154,7 +2125,7 @@ mod tests {
session session
.ensure_agent_prompt_before_user_message("chat-1") .ensure_agent_prompt_before_user_message("chat-1")
.unwrap(); .unwrap();
let history = session.get_history("chat-1").unwrap(); let history = session.get_history(&topic_id).unwrap();
let user_messages = history let user_messages = history
.iter() .iter()
.filter(|message| message.role == "user") .filter(|message| message.role == "user")
@ -2196,11 +2167,14 @@ mod tests {
.unwrap(); .unwrap();
session.ensure_persistent_session("chat-1").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 { for turn in 0..100 {
session 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(); .unwrap();
} }
@ -2209,7 +2183,7 @@ mod tests {
.unwrap(); .unwrap();
// 新设计：系统提示词不再持久化到历史记录 // 新设计：系统提示词不再持久化到历史记录
let history = session.get_history("chat-1").unwrap(); let history = session.get_history(&topic_id).unwrap();
let user_messages = history let user_messages = history
.iter() .iter()
.filter(|message| message.role == "user") .filter(|message| message.role == "user")

154
src/gateway/session_history.rs
View File

@ -17,9 +17,12 @@ fn preview_text(content: &str, max_chars: usize) -> String {
pub(crate) struct SessionHistory { pub(crate) struct SessionHistory {
channel_name: String, channel_name: String,
chat_histories: HashMap<String, Vec<ChatMessage>>, /// 按 topic_id 键化的内存历史缓存。
chat_topic_ids: HashMap<String, String>, // 每个 chat 的当前 topic /// 不同 topic 的历史独立存储，互不干扰，支持多话题并发执行。
history_topic_ids: HashMap<String, String>, // 每个 chat 的历史所对应的话题 topic_histories: HashMap<String, Vec<ChatMessage>>,
/// UI 状态：每个 chat 当前活跃的 topic（按 chat_id 键）。
chat_topic_ids: HashMap<String, String>,
/// 正在压缩中的 topic_id 集合
compression_in_flight: HashSet<String>, compression_in_flight: HashSet<String>,
/// 按 topic_id 的串行化锁。 /// 按 topic_id 的串行化锁。
/// 同一 topic 的消息处理（agent loop + 压缩）必须串行执行， /// 同一 topic 的消息处理（agent loop + 压缩）必须串行执行，
@ -38,9 +41,8 @@ impl SessionHistory {
) -> Self { ) -> Self {
Self { Self {
channel_name: channel_name.into(), channel_name: channel_name.into(),
chat_histories: HashMap::new(), topic_histories: HashMap::new(),
chat_topic_ids: HashMap::new(), chat_topic_ids: HashMap::new(),
history_topic_ids: HashMap::new(),
compression_in_flight: HashSet::new(), compression_in_flight: HashSet::new(),
topic_serial_locks: HashMap::new(), topic_serial_locks: HashMap::new(),
conversations, conversations,
@ -71,38 +73,36 @@ impl SessionHistory {
.map_err(|err| AgentError::Other(format!("session persistence error: {}", err))) .map_err(|err| AgentError::Other(format!("session persistence error: {}", err)))
} }
/// 确保指定 topic 的历史已加载到内存。
/// 按 topic_id 键化查找，如果已存在则直接返回，否则从 DB 加载。
pub(crate) fn ensure_chat_loaded( pub(crate) fn ensure_chat_loaded(
&mut self, &mut self,
chat_id: &str, chat_id: &str,
topic_id: Option<&str>, topic_id: Option<&str>,
) -> Result<(), AgentError> { ) -> 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(()); return Ok(());
} }
// 如果提供了 topic_id，按 topic 加载；否则按 session 加载 let sid = self.persistent_session_id(chat_id);
let mut history = if let Some(tid) = topic_id { let mut history = self
let sid = self.persistent_session_id(chat_id); .conversations
self.conversations .load_messages_for_topic(tid, Some(&sid))
.load_messages_for_topic(tid, Some(&sid)) .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?;
.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)))?
};
// 清理 DB 加载的历史中可能存在的不完整 tool_call 序列
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history); let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history);
if removed > 0 { if removed > 0 {
tracing::warn!( tracing::warn!(
chat_id = %chat_id, topic_id = %tid,
removed_count = removed, removed_count = removed,
"Sanitized incomplete tool_call sequences on history load" "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(()) Ok(())
} }
@ -114,58 +114,56 @@ impl SessionHistory {
Ok(()) Ok(())
} }
pub(crate) fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec<ChatMessage> { pub(crate) fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec<ChatMessage> {
self.chat_histories.entry(chat_id.to_string()).or_default() self.topic_histories.entry(topic_id.to_string()).or_default()
} }
pub(crate) fn get_history(&self, chat_id: &str) -> Option<&Vec<ChatMessage>> { pub(crate) fn get_history(&self, topic_id: &str) -> Option<&Vec<ChatMessage>> {
self.chat_histories.get(chat_id) self.topic_histories.get(topic_id)
} }
pub(crate) fn set_history(&mut self, chat_id: &str, history: Vec<ChatMessage>) { pub(crate) fn set_history(&mut self, topic_id: &str, history: Vec<ChatMessage>) {
self.chat_histories.insert(chat_id.to_string(), history); self.topic_histories.insert(topic_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());
}
} }
/// 获取指定 chat 的历史所对应的话题 /// 设置指定 chat 的当前 topic（UI 状态）
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
pub(crate) fn set_chat_topic(&mut self, chat_id: &str, topic_id: String) { 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); 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> { pub(crate) fn chat_topic(&self, chat_id: &str) -> Option<&str> {
self.chat_topic_ids.get(chat_id).map(|s| s.as_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) { pub(crate) fn clear_chat_topic(&mut self, chat_id: &str) {
self.chat_topic_ids.remove(chat_id); self.chat_topic_ids.remove(chat_id);
} }
pub(crate) fn add_message(&mut self, chat_id: &str, message: ChatMessage) { pub(crate) fn add_message(&mut self, topic_id: &str, message: ChatMessage) {
self.get_or_create_history(chat_id).push(message); self.get_or_create_history(topic_id).push(message);
} }
pub(crate) fn remove_history(&mut self, chat_id: &str) { pub(crate) fn remove_history(&mut self, topic_id: &str) {
self.chat_histories.remove(chat_id); self.topic_histories.remove(topic_id);
self.compression_in_flight.remove(chat_id); self.compression_in_flight.remove(topic_id);
self.history_topic_ids.remove(chat_id);
} }
pub(crate) fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { /// 清空指定 chat/topic 的内存历史和 DB 消息。
if let Some(history) = self.chat_histories.get_mut(chat_id) { /// 内存按 topic_id 清，DB 按 session_id 清（保留原行为以兼容无 topic 场景）。
let len = history.len(); pub(crate) fn clear_chat_history(
history.clear(); &mut self,
#[cfg(debug_assertions)] chat_id: &str,
tracing::debug!(chat_id = %chat_id, previous_len = len, "Chat history cleared"); 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 self.conversations
@ -175,7 +173,7 @@ impl SessionHistory {
pub(crate) fn append_persisted_messages<I>( pub(crate) fn append_persisted_messages<I>(
&mut self, &mut self,
chat_id: &str, topic_id: &str,
messages: I, messages: I,
) -> Result<(), AgentError> ) -> Result<(), AgentError>
where where
@ -186,13 +184,11 @@ impl SessionHistory {
return Ok(()); return Ok(());
} }
// 在追加新消息前，先清理内存历史中的不完整 tool_call 序列 if let Some(history) = self.topic_histories.get_mut(topic_id) {
// 这防止脏数据（如取消时产生的孤立 assistant(tool_calls)）在内存历史中累积
if let Some(history) = self.chat_histories.get_mut(chat_id) {
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(history); let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(history);
if removed > 0 { if removed > 0 {
tracing::warn!( tracing::warn!(
chat_id = %chat_id, topic_id = %topic_id,
removed_count = removed, removed_count = removed,
"Sanitized in-memory history before appending persisted messages" "Sanitized in-memory history before appending persisted messages"
); );
@ -200,7 +196,7 @@ impl SessionHistory {
} }
for message in messages { for message in messages {
self.add_message(chat_id, message); self.add_message(topic_id, message);
} }
Ok(()) Ok(())
} }
@ -219,13 +215,13 @@ impl SessionHistory {
Ok(()) Ok(())
} }
pub(crate) fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> { pub(crate) fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> {
self.get_history(chat_id) self.get_history(topic_id)
.and_then(|history| history.iter().rev().find(|message| message.role == "user")) .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 { pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool {
self.latest_user_message(chat_id) self.latest_user_message(topic_id)
.map(|current| { .map(|current| {
current.id == message.id current.id == message.id
|| (current.content == message.content || (current.content == message.content
@ -237,14 +233,14 @@ impl SessionHistory {
pub(crate) fn stale_result_diagnostics( pub(crate) fn stale_result_diagnostics(
&self, &self,
chat_id: &str, topic_id: &str,
) -> (Option<&str>, Option<String>, bool, usize) { ) -> (Option<&str>, Option<String>, 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_id = latest_user.map(|message| message.id.as_str());
let latest_user_preview = latest_user.map(|message| preview_text(&message.content, 80)); 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 let history_len = self
.get_history(chat_id) .get_history(topic_id)
.map(|history| history.len()) .map(|history| history.len())
.unwrap_or(0); .unwrap_or(0);
@ -256,31 +252,29 @@ impl SessionHistory {
) )
} }
/// 清空所有内存历史（主要用于测试全局重置）。
/// 不遍历清 DB，生产环境如需清 DB 应由调用方显式调用。
pub(crate) fn clear_all_history(&mut self) -> Result<(), AgentError> { pub(crate) fn clear_all_history(&mut self) -> Result<(), AgentError> {
let chat_ids: Vec<String> = self.chat_histories.keys().cloned().collect(); let total: usize = self.topic_histories.values().map(|h| h.len()).sum();
let total: usize = self.chat_histories.values().map(|h| h.len()).sum(); self.topic_histories.clear();
self.chat_histories.clear();
self.compression_in_flight.clear(); self.compression_in_flight.clear();
#[cfg(debug_assertions)] #[cfg(debug_assertions)]
tracing::debug!(previous_total = total, "All chat histories cleared"); tracing::debug!(previous_total = total, "All topic 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))
})?;
}
Ok(()) 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 let history = self
.conversations .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)))?; .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(()) Ok(())
} }

113
src/storage/mod.rs
View File

@ -875,6 +875,74 @@ impl SessionStore {
Ok(()) 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> { pub fn mark_agent_prompt_reinjected(&self, session_id: &str) -> Result<(), StorageError> {
let now = current_timestamp(); let now = current_timestamp();
let conn = self.pool.get()?; let conn = self.pool.get()?;
@ -1667,6 +1735,51 @@ fn insert_message_with_seq(
Ok(()) 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 { fn clone_message_for_compaction(message: &ChatMessage, timestamp: i64) -> ChatMessage {
ChatMessage { ChatMessage {
id: uuid::Uuid::new_v4().to_string(), id: uuid::Uuid::new_v4().to_string(),

20
src/storage/ports.rs
View File

@ -63,6 +63,17 @@ pub trait ConversationRepository: Send + Sync + 'static {
session_id: &str, session_id: &str,
messages: &[ChatMessage], messages: &[ChatMessage],
) -> Result<(), StorageError>; ) -> 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 { pub trait PromptInjectionRepository: Send + Sync + 'static {
@ -252,6 +263,15 @@ impl ConversationRepository for super::SessionStore {
) -> Result<(), StorageError> { ) -> Result<(), StorageError> {
super::SessionStore::replace_active_history(self, session_id, messages) 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 { impl PromptInjectionRepository for super::SessionStore {