diff --git a/src/command/handlers/get_current.rs b/src/command/handlers/get_current.rs index 9fa0d05..a9ae551 100644 --- a/src/command/handlers/get_current.rs +++ b/src/command/handlers/get_current.rs @@ -79,7 +79,7 @@ async fn handle_get_current_session( // 直读 DB 按话题加载消息（不依赖 SessionManager 内存，重启后也能正确获取历史） let messages = handler .store - .load_messages_for_topic(topic_id, Some(&topic.session_id)) + .load_messages_for_topic_full(topic_id, Some(&topic.session_id)) .map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?; let actual_message_count = messages.len(); diff --git a/src/command/handlers/save_topic.rs b/src/command/handlers/save_topic.rs index cefb9cd..6e0636d 100644 --- a/src/command/handlers/save_topic.rs +++ b/src/command/handlers/save_topic.rs @@ -254,7 +254,7 @@ async fn handle_save_topic( let messages = handler .store - .load_messages_for_topic(topic_id, Some(&topic_record.session_id)) + .load_messages_for_topic_full(topic_id, Some(&topic_record.session_id)) .map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?; tracing::debug!( diff --git a/src/gateway/compaction.rs b/src/gateway/compaction.rs index 39c2542..4444dce 100644 --- a/src/gateway/compaction.rs +++ b/src/gateway/compaction.rs @@ -52,17 +52,17 @@ pub(crate) async fn schedule_background_history_compaction( .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. + // 保留原始消息（标记 is_compacted=1）+ 插入压缩摘要，不删除原消息， + // 从而让前端仍能展示完整原始对话，LLM 只看压缩后的精简历史。 store - .replace_topic_history(&session_id, &topic_id, &compressed) - .map_err(|e| AgentError::Other(format!("replace_topic_history error: {}", e)))?; + .compact_topic_history(&session_id, &topic_id, &compressed) + .map_err(|e| AgentError::Other(format!("compact_topic_history error: {}", e)))?; tracing::info!( chat_id = %chat_id, topic_id = %topic_id, compressed_msg_count = compressed.len(), - "Two-segment compression committed" + "Two-segment compression committed (original messages retained)" ); session_guard.reload_topic_history(&chat_id, &topic_id)?; diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 1156e20..27ec216 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -362,7 +362,7 @@ impl InboundProcessor { tokio::spawn(async move { // 从 DB 查询该 topic 的第一条用户消息作为描述生成的依据 let first_user_message = store_clone - .load_messages_for_topic(&topic_id_clone, None) + .load_messages_for_topic_full(&topic_id_clone, None) .ok() .and_then(|msgs| { msgs.into_iter().find(|m| m.role == "user") diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index d755ea0..c9fa19e 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -786,7 +786,7 @@ async fn send_topic_history( task_repository: &Arc, ) -> Result<(), Box> { // 加载话题消息，按 session_id 过滤，避免混入子智能体消息 - let messages = store.load_messages_for_topic(topic_id, Some(session_id))?; + let messages = store.load_messages_for_topic_full(topic_id, Some(session_id))?; tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history"); diff --git a/src/storage/migrations.rs b/src/storage/migrations.rs index 3776351..217f04c 100644 --- a/src/storage/migrations.rs +++ b/src/storage/migrations.rs @@ -71,6 +71,16 @@ pub(super) fn ensure_messages_schema(conn: &Connection) -> Result<(), StorageErr )?; } + // is_compacted: 1 表示该消息是被压缩消费掉的原始消息（前端可见、LLM 不可见）。 + // 压缩摘要消息 is_compacted=0（LLM 可见），通过 system_context='history_compaction*' + // 在前端查询中被排除。保留的原消息（system_guards / 最新 user）is_compacted=0，不重复。 + if !has_column(conn, "messages", "is_compacted")? { + add_column_if_missing( + conn, + "ALTER TABLE messages ADD COLUMN is_compacted INTEGER NOT NULL DEFAULT 0", + )?; + } + // 创建 topic_id 索引（如果不存在） conn.execute( "CREATE INDEX IF NOT EXISTS idx_messages_topic_seq ON messages(topic_id, seq) WHERE topic_id IS NOT NULL", diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 68800da..9379114 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -958,6 +958,107 @@ impl SessionStore { Ok(()) } + /// 压缩该 topic 的历史：保留原始消息（标记 is_compacted=1，前端可见、LLM 不可见）， + /// 并插入压缩摘要消息（is_compacted=0，LLM 可见，前端通过 system_context 过滤排除）。 + /// + /// 与 `replace_topic_history` 的区别：不删除原消息，仅打标记，从而让前端仍能展示 + /// 完整原始对话，同时 LLM 只看到压缩后的精简历史。 + /// + /// `new_messages` 是 `compress_two_segment` 的输出，包含： + /// - 保留原样的消息（system_guards / 最新 user，保留原 ID） + /// - 压缩摘要消息（system_context = history_compaction_*，新 ID） + pub fn compact_topic_history( + &self, + session_id: &str, + topic_id: &str, + new_messages: &[ChatMessage], + ) -> Result<(), StorageError> { + let mut conn = self.pool.get()?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let now = current_timestamp(); + + // 分离摘要消息与保留消息（保留消息携带原 ID，摘要消息是新构造的） + let (summaries, preserved): (Vec<&ChatMessage>, Vec<&ChatMessage>) = new_messages + .iter() + .partition(|m| { + m.system_context + .as_deref() + .map_or(false, |sc| sc.starts_with("history_compaction")) + }); + + // 先删除该 topic 下已有的旧压缩摘要（system_context LIKE 'history_compaction%'）。 + // 旧摘要已被新摘要替代，保留它们只会累积垃圾行（前端和 LLM 都看不到，但占存储）。 + tx.execute( + "DELETE FROM messages \ + WHERE topic_id = ?1 AND session_id = ?2 \ + AND system_context LIKE 'history_compaction%'", + params![topic_id, session_id], + )?; + + // 将该 topic 中未被保留的原消息标记为 is_compacted=1（仅更新尚未标记的行，避免重复写）。 + // 保留消息（system_guards / 最新 user）保持 is_compacted=0，不重复插入。 + let preserved_ids: Vec = + preserved.iter().map(|m| m.id.clone()).collect(); + if preserved_ids.is_empty() { + tx.execute( + "UPDATE messages SET is_compacted = 1 \ + WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0", + params![topic_id, session_id], + )?; + } else { + let placeholders = (0..preserved_ids.len()) + .map(|_| "?") + .collect::>() + .join(","); + let sql = format!( + "UPDATE messages SET is_compacted = 1 \ + WHERE topic_id = ? AND session_id = ? AND is_compacted = 0 \ + AND id NOT IN ({})", + placeholders + ); + let mut params_vec: Vec = vec![topic_id.to_string(), session_id.to_string()]; + params_vec.extend(preserved_ids.iter().cloned()); + tx.execute(&sql, rusqlite::params_from_iter(params_vec))?; + } + + // 插入压缩摘要消息（is_compacted=0，由列默认值保证） + 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 summaries.iter().enumerate() { + let seq = start_seq + i as i64; + insert_message_with_topic_seq(&tx, session_id, topic_id, seq, message)?; + } + + // 更新 topic / session 计数（基于该 topic 全部消息，含被压缩的原始消息） + let topic_count: i64 = tx.query_row( + "SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND topic_id = ?2", + params![session_id, topic_id], + |row| row.get(0), + )?; + tx.execute( + "UPDATE topics SET message_count = ?2, last_active_at = ?3, updated_at = ?3 WHERE id = ?1", + params![topic_id, topic_count, now], + )?; + 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> { let now = current_timestamp(); let conn = self.pool.get()?; @@ -1556,6 +1657,8 @@ impl SessionStore { load_messages_after(&conn, session_id, 0) } + /// LLM 视角：只返回 is_compacted = 0 的消息（压缩摘要 + 未被压缩的新消息）。 + /// 被压缩消费掉的原始消息（is_compacted = 1）对 LLM 不可见，以节省 context。 pub fn load_messages_for_topic( &self, topic_id: &str, @@ -1563,12 +1666,56 @@ impl SessionStore { ) -> Result, StorageError> { let conn = self.pool.get()?; + if let Some(sid) = session_id { + let mut stmt = conn.prepare( + " + SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens + FROM messages + WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0 + ORDER BY seq ASC + ", + )?; + let rows = stmt.query_map(params![topic_id, sid], map_chat_message_row)?; + let mut messages = Vec::new(); + for row in rows { + messages.push(row?); + } + Ok(messages) + } else { + let mut stmt = conn.prepare( + " + SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens + FROM messages + WHERE topic_id = ?1 AND is_compacted = 0 + ORDER BY seq ASC + ", + )?; + let rows = stmt.query_map(params![topic_id], map_chat_message_row)?; + let mut messages = Vec::new(); + for row in rows { + messages.push(row?); + } + Ok(messages) + } + } + + /// UI 视角：返回原始消息（含被压缩消费的 is_compacted=1 消息）+ 未压缩新消息， + /// 排除压缩摘要消息（system_context LIKE 'history_compaction%'）。 + /// 用于前端历史展示、/current、/save-topic、topic 描述生成等场景。 + pub fn load_messages_for_topic_full( + &self, + topic_id: &str, + session_id: Option<&str>, + ) -> Result, StorageError> { + let conn = self.pool.get()?; + if let Some(sid) = session_id { let mut stmt = conn.prepare( " SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens FROM messages WHERE topic_id = ?1 AND session_id = ?2 + AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%') ORDER BY seq ASC ", )?; @@ -1584,6 +1731,7 @@ impl SessionStore { SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens FROM messages WHERE topic_id = ?1 + AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%') ORDER BY seq ASC ", )?; diff --git a/src/storage/ports.rs b/src/storage/ports.rs index 84ee6df..e5cd658 100644 --- a/src/storage/ports.rs +++ b/src/storage/ports.rs @@ -28,6 +28,14 @@ pub trait ConversationRepository: Send + Sync + 'static { session_id: Option<&str>, ) -> Result, StorageError>; + /// UI 视角：返回原始消息（含被压缩消费的 is_compacted=1 消息）+ 未压缩新消息， + /// 排除压缩摘要消息（system_context LIKE 'history_compaction%'）。 + fn load_messages_for_topic_full( + &self, + topic_id: &str, + session_id: Option<&str>, + ) -> Result, StorageError>; + fn append_message(&self, session_id: &str, message: &ChatMessage) -> Result<(), StorageError>; fn append_message_with_topic( @@ -74,6 +82,15 @@ pub trait ConversationRepository: Send + Sync + 'static { topic_id: &str, messages: &[ChatMessage], ) -> Result<(), StorageError>; + + /// 压缩该 topic 的历史：保留原始消息（标记 is_compacted=1）+ 插入压缩摘要。 + /// 不删除原消息，让前端仍能展示完整原始对话，LLM 只看压缩后的精简历史。 + fn compact_topic_history( + &self, + session_id: &str, + topic_id: &str, + new_messages: &[ChatMessage], + ) -> Result<(), StorageError>; } pub trait PromptInjectionRepository: Send + Sync + 'static { @@ -272,6 +289,23 @@ impl ConversationRepository for super::SessionStore { ) -> Result<(), StorageError> { super::SessionStore::replace_topic_history(self, session_id, topic_id, messages) } + + fn load_messages_for_topic_full( + &self, + topic_id: &str, + session_id: Option<&str>, + ) -> Result, StorageError> { + super::SessionStore::load_messages_for_topic_full(self, topic_id, session_id) + } + + fn compact_topic_history( + &self, + session_id: &str, + topic_id: &str, + new_messages: &[ChatMessage], + ) -> Result<(), StorageError> { + super::SessionStore::compact_topic_history(self, session_id, topic_id, new_messages) + } } impl PromptInjectionRepository for super::SessionStore {