diff --git a/src/gateway/compaction.rs b/src/gateway/compaction.rs index 4444dce..15b4e5d 100644 --- a/src/gateway/compaction.rs +++ b/src/gateway/compaction.rs @@ -6,13 +6,20 @@ use crate::agent::AgentError; use super::session::Session; -/// Run two-segment history compression synchronously. +/// Run two-segment history compression. /// -/// 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. +/// The session lock is held only for the brief data-gathering and +/// history-reload phases. The expensive LLM call (2–5 seconds) and the DB +/// write happen **without** holding the session lock, so other commands +/// (e.g. `create_session`, `list_topics`) are not blocked during compression. +/// +/// Concurrency safety: +/// - **Same topic**: the caller (`prepare_and_execute_message`) holds the +/// per-topic serial lock (`_serial_guard`) for the entire duration of +/// execution + compaction, so no other message for this topic can modify +/// the in-memory or DB history between phases. +/// - **Different topic**: fully unblocked — the session lock is free during +/// the LLM call. /// /// 按 topic_id 隔离:压缩只处理指定 topic 的历史,DB 替换也只影响该 topic。 pub(crate) async fn schedule_background_history_compaction( @@ -23,35 +30,41 @@ pub(crate) async fn schedule_background_history_compaction( 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))?; + // Phase 1: brief session lock to gather compaction inputs. + let (history, compressor, store, session_id, provider_config) = { + 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(); + let history = session_guard.get_or_create_history(&topic_id).clone(); + let compressor = session_guard.compressor().clone(); + let store = session_guard.store(); + let session_id = session_guard.persistent_session_id(&chat_id); + let provider_config = session_guard.provider_config().clone(); + (history, compressor, store, session_id, provider_config) + }; + // session lock released here 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" + "Starting two-segment compression (session lock released during LLM call)" ); - // Synchronous compression — holds lock during LLM calls. + // Phase 2: LLM compression WITHOUT holding the session lock. // 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?; + // Phase 3: DB write — store is Arc, no + // session lock needed. // 保留原始消息(标记 is_compacted=1)+ 插入压缩摘要,不删除原消息, // 从而让前端仍能展示完整原始对话,LLM 只看压缩后的精简历史。 store @@ -65,7 +78,11 @@ pub(crate) async fn schedule_background_history_compaction( "Two-segment compression committed (original messages retained)" ); - session_guard.reload_topic_history(&chat_id, &topic_id)?; + // Phase 4: re-acquire session lock to refresh in-memory history. + { + let mut session_guard = session.lock().await; + session_guard.reload_topic_history(&chat_id, &topic_id)?; + } Ok(()) }