4baa8e7a6b
- 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 时更新内存，防止延迟消息污染新话题历史
72 lines
2.4 KiB
Rust
72 lines
2.4 KiB
Rust
use std::sync::Arc;
|
||
|
||
use tokio::sync::Mutex;
|
||
|
||
use crate::agent::AgentError;
|
||
|
||
use super::session::Session;
|
||
|
||
/// Run two-segment history compression synchronously.
|
||
///
|
||
/// Unlike the previous background approach (tokio::spawn), this holds the
|
||
/// session lock during the LLM calls (2–5 seconds). Since the agent loop
|
||
/// 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<Mutex<Session>>,
|
||
chat_id: impl Into<String>,
|
||
topic_id: impl Into<String>,
|
||
) -> 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, Some(&topic_id))?;
|
||
|
||
let history = session_guard.get_or_create_history(&topic_id).clone();
|
||
let compressor = session_guard.compressor().clone();
|
||
|
||
if !compressor.should_compress(&history) {
|
||
return Ok(());
|
||
}
|
||
|
||
let store = session_guard.store();
|
||
let session_id = session_guard.persistent_session_id(&chat_id);
|
||
let provider_config = session_guard.provider_config().clone();
|
||
|
||
tracing::info!(
|
||
chat_id = %chat_id,
|
||
topic_id = %topic_id,
|
||
msg_count = history.len(),
|
||
"Starting synchronous two-segment compression"
|
||
);
|
||
|
||
// Synchronous compression — holds lock during LLM calls.
|
||
// compress_two_segment guarantees the result contains no tool_calls,
|
||
// so there is no risk of orphaned tool call sequences.
|
||
let compressed = compressor
|
||
.compress_two_segment(&history, &provider_config)
|
||
.await?;
|
||
|
||
// Replace only this topic's history in DB (not the entire session).
|
||
// This avoids clobbering other topics' messages during compaction.
|
||
store
|
||
.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_topic_history(&chat_id, &topic_id)?;
|
||
|
||
Ok(())
|
||
}
|