From a629486ad3c9eb894e63b9ba1477c6c3ed73f9de Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 18 Aug 2026 07:51:31 +0800 Subject: [PATCH] =?UTF-8?q?perf(chat):=20=E9=95=BF=E8=AF=9D=E9=A2=98?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E5=88=86=E9=A1=B5=E2=80=94=E2=80=94messages?= =?UTF-8?q?=20=E6=8C=89=20seq=20keyset=20=E5=A2=9E=E9=87=8F=E5=8A=A0?= =?UTF-8?q?=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端新增 load_messages_for_topic_page（seq < cursor + limit 分页，走 (session_id, seq) 索引替代 OFFSET 深翻页），ChatMessage 增加 seq 游标；历史批次消息带 topic_id 下发，前端以 seq+topic_id 双重判定批次归属，规避切话题瞬间在途旧批次污染。 前端触顶增量加载：批次缓存在 pendingHistoryRef，收到 topic_history_end 一次性去重 prepend，scrollTop 按新增高度补偿锚定原头部消息；不足一屏自动补页，loading 超时 10s 自愈；流式输出中加载历史不清空流式累加器，避免已流出文本丢失。 --- src/agent/agent_loop.rs | 1 + src/bus/message.rs | 10 ++ src/command/adapters/websocket.rs | 8 + src/command/handlers/load_chat_messages.rs | 61 +++++++ src/command/mod.rs | 6 + src/gateway/ws.rs | 175 ++++++++++++++++++--- src/protocol/mod.rs | 21 +++ src/protocol/ws_adapter.rs | 9 ++ src/storage/mod.rs | 65 ++++++++ src/storage/row_mapping.rs | 3 +- src/storage/tests.rs | 58 +++++++ tests/test_request_format.rs | 2 + web/src/App.tsx | 7 + web/src/components/Chat/ChatContainer.tsx | 12 ++ web/src/components/Chat/MessageList.tsx | 82 +++++++++- web/src/hooks/chat/useMessages.ts | 161 ++++++++++++++++++- web/src/hooks/useChat.ts | 14 ++ web/src/types/protocol.ts | 26 +++ 18 files changed, 698 insertions(+), 23 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index a4a9429..ad743a4 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -400,6 +400,7 @@ fn filter_images_by_age_and_count_inner( tool_duration_ms: message.tool_duration_ms, tool_calls: message.tool_calls.clone(), usage: message.usage.clone(), + seq: None, }); } diff --git a/src/bus/message.rs b/src/bus/message.rs index 6443417..bbe01fc 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -53,6 +53,10 @@ pub struct ChatMessage { pub content: String, pub media_refs: Vec, // Paths to media files for context pub timestamp: i64, + /// 会话内单调递增序号（DB 分配）。仅历史加载路径填充； + /// 实时推送路径不填，前端以此区分"历史批次"与"实时消息"。 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub seq: Option, #[serde(skip_serializing_if = "Option::is_none")] pub system_context: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -122,6 +126,7 @@ impl ChatMessage { tool_state: None, tool_calls: None, usage: None, + seq: None, } } @@ -140,6 +145,7 @@ impl ChatMessage { tool_state: None, tool_calls: None, usage: None, + seq: None, } } @@ -158,6 +164,7 @@ impl ChatMessage { tool_state: None, tool_calls: None, usage: None, + seq: None, } } @@ -188,6 +195,7 @@ impl ChatMessage { tool_state: None, tool_calls: Some(tool_calls), usage: None, + seq: None, } } @@ -223,6 +231,7 @@ impl ChatMessage { tool_state: None, tool_calls: None, usage: None, + seq: None, } } @@ -259,6 +268,7 @@ impl ChatMessage { tool_state: Some(tool_state), tool_calls: None, usage: None, + seq: None, } } diff --git a/src/command/adapters/websocket.rs b/src/command/adapters/websocket.rs index 676cc3a..e8c36a3 100644 --- a/src/command/adapters/websocket.rs +++ b/src/command/adapters/websocket.rs @@ -85,6 +85,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, MessageKind::Notification => { // 根据元数据判断具体类型 @@ -120,6 +121,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, } } else if let Some(topics_json) = response.metadata.get("topics") { @@ -145,6 +147,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, } } else if let Some(session_id) = response.metadata.get("session_id") { @@ -193,6 +196,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, } } else if let Some(sessions_json) = response.metadata.get("sessions") { @@ -218,6 +222,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, } } else if let Some(topics_json) = response.metadata.get("topics") { @@ -243,6 +248,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, } } else { @@ -257,6 +263,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, } } } @@ -276,6 +283,7 @@ impl OutputAdapter for WebSocketOutputAdapter { timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: None, user_message_id: None, + seq: None, }, }; outbounds.push(outbound); diff --git a/src/command/handlers/load_chat_messages.rs b/src/command/handlers/load_chat_messages.rs index 226b222..028637e 100644 --- a/src/command/handlers/load_chat_messages.rs +++ b/src/command/handlers/load_chat_messages.rs @@ -63,3 +63,64 @@ impl CommandHandler for LoadChatMessagesCommandHandler { } } } + +/// 分页加载话题更早历史消息。校验参数后由 ws.rs 读取 metadata 执行加载。 +pub struct LoadOlderMessagesCommandHandler; + +impl LoadOlderMessagesCommandHandler { + pub fn new() -> Self { + Self + } +} + +impl Default for LoadOlderMessagesCommandHandler { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl CommandHandler for LoadOlderMessagesCommandHandler { + fn can_handle(&self, cmd: &Command) -> bool { + matches!(cmd, Command::LoadOlderMessages { .. }) + } + + fn metadata(&self) -> Option { + Some(CommandMetadata { + name: "load_older_messages", + description: "分页加载话题更早的历史消息", + usage: "/load_older_messages ", + }) + } + + async fn handle( + &self, + cmd: Command, + ctx: CommandContext, + ) -> Result { + match cmd { + Command::LoadOlderMessages { + topic_id, + before_seq, + } => { + if topic_id.is_empty() { + return Err(CommandError::new( + "INVALID_ARGUMENT", + "topic_id must not be empty".to_string(), + )); + } + if before_seq < 0 { + return Err(CommandError::new( + "INVALID_ARGUMENT", + "before_seq must be non-negative".to_string(), + )); + } + + Ok(CommandResponse::success(ctx.request_id) + .with_metadata("load_older_topic_id", &topic_id) + .with_metadata("load_older_before_seq", before_seq.to_string())) + } + _ => unreachable!(), + } + } +} diff --git a/src/command/mod.rs b/src/command/mod.rs index 6861a10..ec8f10d 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -49,6 +49,11 @@ pub enum Command { ListSchedulerJobs, /// 加载指定 channel + chat_id 的对话消息 LoadChatMessages { channel: String, chat_id: String }, + /// 分页加载话题更早的历史消息（用户向上滚动触发） + LoadOlderMessages { + topic_id: String, + before_seq: i64, + }, /// 删除指定话题 DeleteTopic { topic_id: String }, /// 重命名指定话题 @@ -95,6 +100,7 @@ impl Command { Command::LoadTaskMessages { .. } => "load_task_messages", Command::ListSchedulerJobs => "list_scheduler_jobs", Command::LoadChatMessages { .. } => "load_chat_messages", + Command::LoadOlderMessages { .. } => "load_older_messages", Command::DeleteTopic { .. } => "delete_topic", Command::RenameTopic { .. } => "rename_topic", Command::StopExecution => "stop_execution", diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index 3ee2d60..f2e7d6b 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -588,6 +588,10 @@ async fn handle_inbound( router.register(Box::new(MemoryCrudCommandHandler::new(store.clone()))); // 注册 load_chat_messages 处理器 router.register(Box::new(LoadChatMessagesCommandHandler::new())); + // 注册 load_older_messages 处理器（历史分页） + router.register(Box::new( + crate::command::handlers::load_chat_messages::LoadOlderMessagesCommandHandler::new(), + )); // 注册 stop_execution 处理器 router.register(Box::new(StopExecutionCommandHandler::new( state.cancel_manager.clone(), @@ -797,6 +801,31 @@ async fn handle_inbound( } } + // 分页加载话题更早历史（前端向上滚动触发） + if let Some(topic_id) = response.metadata.get("load_older_topic_id") { + let before_seq = response + .metadata + .get("load_older_before_seq") + .and_then(|v| v.parse::().ok()) + .unwrap_or(i64::MAX); + if let Err(e) = send_older_messages( + &store, + current_session_id, + topic_id, + before_seq, + sender, + ) + .await + { + tracing::warn!( + error = %e, + topic_id = %topic_id, + before_seq, + "Failed to send older messages" + ); + } + } + if current_topic_id.is_none() && let Some(topics_json) = response.metadata.get("topics") { @@ -855,6 +884,10 @@ fn resolve_ws_sender_id(sender_id: Option<&str>, runtime_session_id: &str) -> St } /// 加载并发送话题历史消息 +/// 话题历史初始页大小：切换话题时仅加载最新 N 条， +/// 更早消息由前端滚动触发 load_older_messages 增量加载。 +const TOPIC_HISTORY_PAGE_SIZE: usize = 200; + async fn send_topic_history( store: &Arc, session_id: &str, @@ -862,26 +895,40 @@ async fn send_topic_history( sender: &mpsc::Sender, task_repository: &Arc, ) -> Result<(), Box> { - // 加载话题消息，按 session_id 过滤，避免混入子智能体消息。 + // 分页加载最新一页，避免长话题全量传输/解析阻塞首屏。 // SQLite 同步加载 + running 占位对账移入 blocking 线程池，避免阻塞 async worker。 let store_bg = store.clone(); let topic_id_bg = topic_id.to_string(); let session_id_bg = session_id.to_string(); - let messages = tokio::task::spawn_blocking(move || { - let mut messages = - store_bg.load_messages_for_topic_full(&topic_id_bg, Some(&session_id_bg))?; + let (messages, has_more) = tokio::task::spawn_blocking(move || { + let (mut messages, has_more) = store_bg.load_messages_for_topic_page( + &topic_id_bg, + Some(&session_id_bg), + None, + TOPIC_HISTORY_PAGE_SIZE, + )?; reconcile_running_in_messages(&mut messages, &store_bg, &topic_id_bg); - Ok::<_, crate::storage::StorageError>(messages) + Ok::<_, crate::storage::StorageError>((messages, has_more)) }) .await .map_err(|e| format!("Topic history load task failed: {}", e))??; - tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history"); + tracing::info!(topic_id = %topic_id, message_count = messages.len(), has_more, "Sending topic history (paged)"); - // 收集已有 tool_result 的 tool_call_id 集合，用于判断任务是否已有结果 + // 收集已加载页内的 tool_call/tool_result id 集合，用于判断任务是否已有结果。 + // 分页后集合仅覆盖已加载窗口：窗口外的任务不补发 TaskStarted + // （其 tool_call 气泡同样不在窗口内，补发会造成前端凭空出现孤立卡片）。 + let mut tool_call_ids: std::collections::HashSet = std::collections::HashSet::new(); let mut tool_call_ids_with_results: std::collections::HashSet = std::collections::HashSet::new(); for msg in &messages { + if msg.role == "assistant" + && let Some(ref tool_calls) = msg.tool_calls + { + for tc in tool_calls { + tool_call_ids.insert(tc.id.clone()); + } + } if msg.role == "tool" && let Some(ref tcid) = msg.tool_call_id { @@ -892,10 +939,15 @@ async fn send_topic_history( // 将消息转换为 WsOutbound 并发送。 // 转换过程对每条媒体引用做同步文件读取 + base64 编码（CPU/IO 密集）， // 整体移入 blocking 线程池一次性产出，避免阻塞 async worker。 + // （闭包 move 捕获 messages，先提取分页游标） + // topic_id 随批次下发：前端以 seq+topic_id 双重判定历史批次归属， + // 规避切话题瞬间在途旧批次污染新话题列表。 + let oldest_seq = messages.first().and_then(|m| m.seq); + let topic_id_for_convert = topic_id.to_string(); let outbound_batches: Vec> = tokio::task::spawn_blocking(move || { messages .iter() - .map(chat_message_to_ws_outbound) + .map(|m| chat_message_to_ws_outbound(m, Some(&topic_id_for_convert))) .collect::>() }) .await @@ -904,6 +956,15 @@ async fn send_topic_history( let _ = sender.send(outbound).await; } + // 批次结束标记：前端据此初始化分页游标与 has_more 状态 + let _ = sender + .send(WsOutbound::TopicHistoryEnd { + topic_id: topic_id.to_string(), + has_more, + oldest_seq, + }) + .await; + // 查询该话题下所有子智能体任务，补发 TaskStarted 事件 // 解决页面刷新后 navigateToTaskId 丢失的问题 let tasks = match task_repository.list_tasks_for_topic(topic_id).await { @@ -916,8 +977,17 @@ async fn send_topic_history( for task in tasks { // 判断是否需要补发 TaskStarted： - // - 如果该任务的 tool_call_id 已有对应的 tool_result，前端会显示结果，不需要补发 - // - 否则（Running 状态或已完成但结果未进入历史），补发 TaskStarted 以便前端显示"查看实时进度" + // - 任务的 tool_call 不在已加载窗口内 → 不补发（前端无对应气泡，补发会产生孤立卡片） + // - tool_call 在窗口内且已有 tool_result → 前端会显示结果，不补发 + // - tool_call 在窗口内但无 tool_result（Running 或结果未入历史）→ 补发以显示"查看实时进度" + let task_call_in_window = task + .tool_call_id + .as_ref() + .map(|tcid| tool_call_ids.contains(tcid)) + .unwrap_or(false); + if !task_call_in_window { + continue; + } let has_tool_result = task .tool_call_id .as_ref() @@ -953,6 +1023,63 @@ async fn send_topic_history( Ok(()) } +/// 分页加载并发送话题更早的历史消息（load_older_messages 命令）。 +/// 与初始页不同：不补发 TaskStarted（运行中任务必在最新页）， +/// 仅发送消息批次 + TopicHistoryEnd 游标标记。 +async fn send_older_messages( + store: &Arc, + session_id: &str, + topic_id: &str, + before_seq: i64, + sender: &mpsc::Sender, +) -> Result<(), Box> { + let store_bg = store.clone(); + let topic_id_bg = topic_id.to_string(); + let session_id_bg = session_id.to_string(); + let (messages, has_more) = tokio::task::spawn_blocking(move || { + store_bg.load_messages_for_topic_page( + &topic_id_bg, + Some(&session_id_bg), + Some(before_seq), + TOPIC_HISTORY_PAGE_SIZE, + ) + }) + .await + .map_err(|e| format!("Older messages load task failed: {}", e))??; + + tracing::debug!( + topic_id = %topic_id, + before_seq, + message_count = messages.len(), + has_more, + "Sending older messages (paged)" + ); + + let oldest_seq = messages.first().and_then(|m| m.seq); + let topic_id_for_convert = topic_id.to_string(); + let outbound_batches: Vec> = tokio::task::spawn_blocking(move || { + messages + .iter() + .map(|m| chat_message_to_ws_outbound(m, Some(&topic_id_for_convert))) + .collect::>() + }) + .await + .map_err(|e| format!("Older messages convert task failed: {}", e))?; + for outbound in outbound_batches.into_iter().flatten() { + let _ = sender.send(outbound).await; + } + + let _ = sender + .send(WsOutbound::TopicHistoryEnd { + topic_id: topic_id.to_string(), + has_more, + oldest_seq, + }) + .await; + + Ok(()) +} + /// 发送前对账消息列表中的 task "running" 占位。 /// /// DB messages 表中的 task tool_result 行固化在 spawn 时刻的 running 状态， @@ -1036,7 +1163,9 @@ async fn send_task_messages( messages .iter() .map(|msg| { - let mut outbounds = chat_message_to_ws_outbound(msg); + // 任务会话消息不属于主话题历史：topic_id 传 None， + // 前端按 topic_id 过滤历史批次时不会误收。 + let mut outbounds = chat_message_to_ws_outbound(msg, None); if let Some(ref task_id) = subagent_task_id_bg { for ob in &mut outbounds { set_subagent_task_id(ob, task_id); @@ -1138,8 +1267,12 @@ fn extract_parent_task_id(task: &crate::tools::task::types::TaskSession) -> Opti } /// 将 ChatMessage 转换为 WsOutbound 列表 -fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec { +fn chat_message_to_ws_outbound( + msg: &crate::bus::ChatMessage, + topic_id: Option<&str>, +) -> Vec { use crate::bus::message::ToolMessageState; + let topic_id_out = topic_id.map(str::to_string); // Helper function to strip media_refs_json from content fn strip_media_refs_json(content: &str) -> String { @@ -1210,10 +1343,11 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec role: msg.role.clone(), attachments: Vec::new(), subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), timestamp: Some(msg.timestamp / 1000), reasoning_content: msg.reasoning_content.clone(), user_message_id: None, + seq: msg.seq, }); } // AssistantResponse 已携带 reasoning 时，ToolCall 不再重复 @@ -1231,10 +1365,11 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec content: format!("{}\nargs: {}", tool_call.name, tool_call.arguments), role: msg.role.clone(), subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), timestamp: Some(msg.timestamp / 1000), reasoning_content: tc_reasoning.clone(), user_message_id: None, + seq: msg.seq, }); } outbound @@ -1246,10 +1381,11 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec role: msg.role.clone(), attachments: Vec::new(), subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), timestamp: Some(msg.timestamp / 1000), reasoning_content: msg.reasoning_content.clone(), user_message_id: None, + seq: msg.seq, }] } } @@ -1269,9 +1405,10 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec content: msg.content.clone(), role: msg.role.clone(), subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), duration_ms: msg.tool_duration_ms, timestamp: Some(msg.timestamp / 1000), + seq: msg.seq, }], ToolMessageState::PendingUserAction => vec![WsOutbound::ToolPending { id: msg @@ -1284,8 +1421,9 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec role: msg.role.clone(), resume_hint: "完成外部操作后，直接发一条继续消息即可。".to_string(), subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), timestamp: Some(msg.timestamp / 1000), + seq: msg.seq, }], } } @@ -1295,10 +1433,11 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec role: msg.role.clone(), attachments, subagent_task_id: None, - topic_id: None, + topic_id: topic_id_out.clone(), timestamp: Some(msg.timestamp / 1000), reasoning_content: None, user_message_id: None, + seq: msg.seq, }], _ => Vec::new(), } diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index 844c080..6031ab9 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -176,6 +176,9 @@ pub enum WsOutbound { reasoning_content: Option, #[serde(default, skip_serializing_if = "Option::is_none")] user_message_id: Option, + /// 历史分页游标：仅历史加载批次填充（实时推送为 None） + #[serde(default, skip_serializing_if = "Option::is_none")] + seq: Option, }, #[serde(rename = "tool_call")] ToolCall { @@ -195,6 +198,9 @@ pub enum WsOutbound { reasoning_content: Option, #[serde(default, skip_serializing_if = "Option::is_none")] user_message_id: Option, + /// 历史分页游标：仅历史加载批次填充（实时推送为 None） + #[serde(default, skip_serializing_if = "Option::is_none")] + seq: Option, }, #[serde(rename = "tool_result")] ToolResult { @@ -211,6 +217,9 @@ pub enum WsOutbound { duration_ms: Option, #[serde(default, skip_serializing_if = "Option::is_none")] timestamp: Option, + /// 历史分页游标：仅历史加载批次填充（实时推送为 None） + #[serde(default, skip_serializing_if = "Option::is_none")] + seq: Option, }, #[serde(rename = "tool_pending")] ToolPending { @@ -226,6 +235,18 @@ pub enum WsOutbound { topic_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] timestamp: Option, + /// 历史分页游标：仅历史加载批次填充（实时推送为 None） + #[serde(default, skip_serializing_if = "Option::is_none")] + seq: Option, + }, + /// 话题历史批次结束标记：随每批历史消息末尾发送， + /// 前端据此更新分页游标（oldest_seq）与 has_more 状态 + #[serde(rename = "topic_history_end")] + TopicHistoryEnd { + topic_id: String, + has_more: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + oldest_seq: Option, }, #[serde(rename = "error")] Error { diff --git a/src/protocol/ws_adapter.rs b/src/protocol/ws_adapter.rs index 56c2560..7188a70 100644 --- a/src/protocol/ws_adapter.rs +++ b/src/protocol/ws_adapter.rs @@ -28,6 +28,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec Vec Vec Vec vec![WsOutbound::ToolPending { id: message @@ -97,6 +101,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec Vec::new(), @@ -130,6 +135,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: message.reasoning_content.clone(), user_message_id: message.metadata.get("user_message_id").cloned(), + seq: None, }] } OutboundEventKind::ToolCall => vec![WsOutbound::ToolCall { @@ -150,6 +156,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve timestamp: Some(crate::protocol::now_timestamp()), reasoning_content: message.reasoning_content.clone(), user_message_id: message.metadata.get("user_message_id").cloned(), + seq: None, }], OutboundEventKind::ToolResult => vec![WsOutbound::ToolResult { id: message @@ -167,6 +174,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve .get("tool_duration_ms") .and_then(|v| v.parse().ok()), timestamp: Some(crate::protocol::now_timestamp()), + seq: None, }], OutboundEventKind::ToolPending => vec![WsOutbound::ToolPending { id: message @@ -181,6 +189,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve subagent_task_id: message.metadata.get("subagent_task_id").cloned(), topic_id: message.metadata.get("topic_id").cloned(), timestamp: Some(crate::protocol::now_timestamp()), + seq: None, }], OutboundEventKind::ErrorNotification => vec![WsOutbound::Error { code: "AGENT_ERROR".to_string(), diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 4c98659..f19beb5 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -1835,6 +1835,68 @@ impl SessionStore { } } + /// 按话题分页加载消息（keyset 分页，从最新端向前取页）。 + /// + /// 切换话题时只加载最新 `limit` 条，更早消息由用户向上滚动时按 + /// `before_seq` 游标增量加载，避免长话题全量传输/解析阻塞首屏。 + /// 过滤条件与 [`Self::load_messages_for_topic_full`] 一致 + /// （排除 history_compaction 摘要行）。返回正序消息 + 是否还有更早页。 + pub fn load_messages_for_topic_page( + &self, + topic_id: &str, + session_id: Option<&str>, + before_seq: Option, + limit: usize, + ) -> Result<(Vec, bool), StorageError> { + let conn = self.pool.get()?; + // 多取 1 条仅用于判断 has_more，截断后返回 limit 条 + let fetch = limit + 1; + let mut messages: Vec = if let Some(sid) = session_id { + let mut stmt = conn.prepare(&format!( + " + SELECT {MESSAGE_LOAD_COLUMNS} + FROM messages + WHERE topic_id = ?1 AND session_id = ?2 + AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%') + AND (?3 IS NULL OR seq < ?3) + ORDER BY seq DESC + LIMIT {fetch} + ", + ))?; + let rows = stmt.query_map(params![topic_id, sid, before_seq], map_chat_message_row)?; + let mut out = Vec::new(); + for row in rows { + out.push(row?); + } + out + } else { + let mut stmt = conn.prepare(&format!( + " + SELECT {MESSAGE_LOAD_COLUMNS} + FROM messages + WHERE topic_id = ?1 + AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%') + AND (?2 IS NULL OR seq < ?2) + ORDER BY seq DESC + LIMIT {fetch} + ", + ))?; + let rows = stmt.query_map(params![topic_id, before_seq], map_chat_message_row)?; + let mut out = Vec::new(); + for row in rows { + out.push(row?); + } + out + }; + let has_more = messages.len() > limit; + if has_more { + messages.truncate(limit); + } + // DESC 取页后反转为正序，与全量加载的顺序语义一致 + messages.reverse(); + Ok((messages, has_more)) + } + /// 定向查询指定话题的第一条 user 消息内容。 /// /// 数据库侧 `LIMIT 1`，避免为取单条消息全量加载并反序列化整个话题历史 @@ -2446,6 +2508,7 @@ fn clone_message_for_compaction(message: &ChatMessage, timestamp: i64) -> ChatMe tool_calls: message.tool_calls.clone(), // 压缩克隆不保留 usage：压缩产生的是合成消息，不代表真实 LLM 调用 usage: None, + seq: None, } } @@ -2504,6 +2567,7 @@ fn load_messages_between( tool_duration_ms: row.get::<_, Option>(10)?.map(|v| v as u64), tool_calls, usage: map_usage_row(row, 11, 12, 13, 14, 15)?, + seq: row.get(16)?, }) }, )?; @@ -2566,6 +2630,7 @@ fn load_messages_after( tool_duration_ms: row.get::<_, Option>(10)?.map(|v| v as u64), tool_calls, usage: map_usage_row(row, 11, 12, 13, 14, 15)?, + seq: row.get(16)?, }) })?; diff --git a/src/storage/row_mapping.rs b/src/storage/row_mapping.rs index 0a2db5b..dc6b71c 100644 --- a/src/storage/row_mapping.rs +++ b/src/storage/row_mapping.rs @@ -17,7 +17,7 @@ use super::{ /// 消息加载查询的共享列清单（列序与 map_chat_message_row / map_usage_row 的下标一一对应）。 /// 新增 usage 列时只需改这里 + map_usage_row，无需逐条 SELECT 手工对齐。 -pub(super) const MESSAGE_LOAD_COLUMNS: &str = "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, cached_tokens"; +pub(super) const MESSAGE_LOAD_COLUMNS: &str = "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, cached_tokens, seq"; /// 从指定列索引读取 token usage 五元组（含 context_window_tokens、cached_tokens）。 pub(super) fn map_usage_row( @@ -209,6 +209,7 @@ pub(super) fn map_chat_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result< tool_duration_ms: row.get::<_, Option>(10)?.map(|v| v as u64), tool_calls, usage: map_usage_row(row, 11, 12, 13, 14, 15)?, + seq: row.get(16)?, }) } diff --git a/src/storage/tests.rs b/src/storage/tests.rs index e2fe56d..5fd8bc4 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -718,6 +718,64 @@ fn test_get_topic_message_count_uses_count_query() { ); } +/// 回归防护：所有按 topic_id 过滤的查询必须命中索引（idx_messages_topic_seq +/// 或 idx_messages_session_*），不允许退化为全表扫描。 +/// 消息表是最大的表，长会话场景下全表扫描是数据层主要退化点。 +#[test] +fn test_topic_id_queries_use_index() { + let store = SessionStore::in_memory().unwrap(); + + // 索引必须存在（partial index：topic_id IS NOT NULL） + let conn = store.pool.get().unwrap(); + let index_sql: String = conn + .query_row( + "SELECT sql FROM sqlite_master WHERE type='index' AND name='idx_messages_topic_seq'", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(index_sql.contains("topic_id")); + + // 覆盖所有 topic_id 查询形态（与 mod.rs 中实际 SQL 一致） + let queries: &[&str] = &[ + // load_messages_for_topic（带/不带 session_id 分支） + "SELECT id FROM messages WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0 ORDER BY seq ASC", + "SELECT id FROM messages WHERE topic_id = ?1 AND is_compacted = 0 ORDER BY seq ASC", + // load_messages_for_topic_full + "SELECT id FROM messages WHERE topic_id = ?1 AND session_id = ?2 ORDER BY seq ASC", + "SELECT id FROM messages WHERE topic_id = ?1 ORDER BY seq ASC", + // count_user_messages_for_topic / get_topic_message_count + "SELECT COUNT(*) FROM messages WHERE topic_id = ?1", + "SELECT COUNT(*) FROM messages WHERE topic_id = ?1 AND role = 'user' AND is_compacted = 0", + "SELECT COUNT(*) FROM messages WHERE topic_id = ?1 AND role = 'user'", + // 按会话+话题删除 + "DELETE FROM messages WHERE session_id = ?1 AND topic_id = ?2", + // 话题聚合（IN 列表） + "SELECT topic_id, COUNT(*) FROM messages WHERE topic_id IN (?1, ?2) AND role = 'assistant' GROUP BY topic_id", + ]; + + for sql in queries { + let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap(); + // EXPLAIN 不执行语句，但 rusqlite 要求参数计数匹配；按占位符数量传哑参数 + let param_count = sql.matches('?').count(); + let mut rows = stmt + .query(rusqlite::params_from_iter(vec!["x"; param_count])) + .unwrap(); + let mut plan = String::new(); + while let Some(row) = rows.next().unwrap() { + let detail: String = row.get(3).unwrap(); + plan.push_str(&detail); + plan.push_str(" | "); + } + // partial index（WHERE topic_id IS NOT NULL）可被 topic_id = ? / IN + // 等值查询使用；复合条件查询允许规划器选择 session 前缀索引。 + assert!( + plan.contains("idx_messages_topic_seq") || plan.contains("idx_messages_session"), + "query degrades to full scan: {sql}\nplan: {plan}" + ); + } +} + #[test] fn test_repair_session_id_prefix_pollution() { let store = SessionStore::in_memory().unwrap(); diff --git a/tests/test_request_format.rs b/tests/test_request_format.rs index eac4cb8..e54ed93 100644 --- a/tests/test_request_format.rs +++ b/tests/test_request_format.rs @@ -129,6 +129,7 @@ fn test_tool_call_outbound_serialization() { timestamp: None, reasoning_content: None, user_message_id: None, + seq: None, }; let json = serde_json::to_string(&msg).unwrap(); @@ -164,6 +165,7 @@ fn test_tool_result_outbound_serialization() { duration_ms: None, topic_id: None, timestamp: None, + seq: None, }; let json = serde_json::to_string(&msg).unwrap(); diff --git a/web/src/App.tsx b/web/src/App.tsx index eec70b2..0fb0975 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -65,6 +65,10 @@ function App() { messages, isLoading, isReadOnly, + // 历史分页 + hasMoreOlder, + loadingOlder, + loadOlderMessages, // 子智能体视图 subAgentView, subAgentStack, @@ -1060,6 +1064,9 @@ function App() { sessionId={sessionId} settingsClosedTick={settingsClosedTick} onOpenSettings={openExpertsSettings} + hasMoreOlder={!subAgentView && !schedulerView ? hasMoreOlder : false} + loadingOlder={!subAgentView && !schedulerView ? loadingOlder : false} + onLoadOlder={!subAgentView && !schedulerView ? loadOlderMessages : undefined} /> diff --git a/web/src/components/Chat/ChatContainer.tsx b/web/src/components/Chat/ChatContainer.tsx index 0e21df6..f420104 100644 --- a/web/src/components/Chat/ChatContainer.tsx +++ b/web/src/components/Chat/ChatContainer.tsx @@ -27,6 +27,12 @@ interface ChatContainerProps { settingsClosedTick?: number; /** 当前话题 ID，用于切换话题时清空输入框草稿 */ topicId?: string | null; + /** 历史分页：是否还有更早的消息可加载 */ + hasMoreOlder?: boolean; + /** 历史分页：是否正在加载更早一页 */ + loadingOlder?: boolean; + /** 触顶时请求加载更早的历史消息 */ + onLoadOlder?: () => void; } // memo：props 除 messages 外全部稳定（useCallback/原始值）， @@ -46,6 +52,9 @@ export const ChatContainer = memo(function ChatContainer({ onOpenSettings, settingsClosedTick, topicId, + hasMoreOlder, + loadingOlder, + onLoadOlder, }: ChatContainerProps) { const [selectedExpert, setSelectedExpert] = useState<{ name: string; @@ -127,6 +136,9 @@ export const ChatContainer = memo(function ChatContainer({ viewKey={viewKey} highlightedMessageId={highlightedMessageId} effectiveModel={effectiveModel} + hasMoreOlder={hasMoreOlder} + loadingOlder={loadingOlder} + onLoadOlder={onLoadOlder} /> )} diff --git a/web/src/components/Chat/MessageList.tsx b/web/src/components/Chat/MessageList.tsx index 8572a53..dde55f5 100644 --- a/web/src/components/Chat/MessageList.tsx +++ b/web/src/components/Chat/MessageList.tsx @@ -2,7 +2,7 @@ import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo, mem import { useVirtualizer } from '@tanstack/react-virtual'; import { MessageBubble } from './MessageBubble'; import type { ChatMessage } from '../../types/protocol'; -import { Sparkles, ArrowDown, ArrowUp } from 'lucide-react'; +import { Sparkles, ArrowDown, ArrowUp, Loader2 } from 'lucide-react'; interface MessageListProps { messages: ChatMessage[]; @@ -14,6 +14,12 @@ interface MessageListProps { highlightedMessageId?: string | null; /** 主代理当前生效模型（透传给 MessageBubble 做 Task 卡片差异显示） */ effectiveModel?: { provider: string; model: string } | null; + /** 历史分页：是否还有更早的消息可加载 */ + hasMoreOlder?: boolean; + /** 历史分页：是否正在加载更早一页 */ + loadingOlder?: boolean; + /** 触顶时请求加载更早的历史消息 */ + onLoadOlder?: () => void; } // memo：props 与 ChatContainer 同源（messages 变化时才需重渲染）， @@ -25,6 +31,9 @@ export const MessageList = memo(function MessageList({ viewKey, highlightedMessageId, effectiveModel, + hasMoreOlder = false, + loadingOlder = false, + onLoadOlder, }: MessageListProps) { const containerRef = useRef(null); const isAtBottomRef = useRef(true); @@ -34,6 +43,10 @@ export const MessageList = memo(function MessageList({ const viewKeyRef = useRef(viewKey); viewKeyRef.current = viewKey; + // 历史分页：onLoadOlder 经 ref 转发，保持 handleScroll 空依赖稳定 + const onLoadOlderRef = useRef(onLoadOlder); + onLoadOlderRef.current = onLoadOlder; + // 追踪上次的消息条数，用于计算真正新增的消息数（而非 messages 引用变化次数）。 // 流式输出时每个 delta 都会产生新的 messages 数组引用，但消息条数不变， // 不应计入 newMessageCount。 @@ -116,6 +129,14 @@ export const MessageList = memo(function MessageList({ const el = containerRef.current; if (!el) return; + // 触顶加载更早历史：滚顶动画运行中不触发（动画会持续覆写 scrollTop， + // 与 prepend 锚定补偿互相打架）；动画落定在 scrollTop=0 时会发出最后一次 + // scroll 事件，此刻 rafRef 已清零，可正常触发。重复触发由 hook 内 + // loading/hasMore 守卫吸收。 + if (el.scrollTop <= 0 && scrollTopRafRef.current === 0) { + onLoadOlderRef.current?.(); + } + const distanceFromBottom = el.scrollHeight - el.scrollTop - el.clientHeight; const nearBottom = distanceFromBottom < 120; @@ -141,18 +162,45 @@ export const MessageList = memo(function MessageList({ }, []); // ---- auto-scroll: handle view switches and message updates ---- + // 同时承担历史分页 prepend 的滚动锚定：头部插入更早消息后， + // 按新增高度补偿 scrollTop，让用户视野停在原头部消息上。 + + // 头部 prepend 检测：记录上次渲染的头部消息 id 与滚动内容总高 + const prevFirstIdRef = useRef(null); + const prevScrollHeightRef = useRef(0); useLayoutEffect(() => { + const el = containerRef.current; const prevKey = prevViewKeyRef.current; const viewChanged = prevKey !== viewKey; prevViewKeyRef.current = viewKey; + const firstId = messages[0]?.id ?? null; + const prevFirstId = prevFirstIdRef.current; + prevFirstIdRef.current = firstId; + if (messages.length === 0) { isAtBottomRef.current = true; prevMessageCountRef.current = 0; + if (el) prevScrollHeightRef.current = el.scrollHeight; return; } + // 同视图内头部变化 = 历史消息 prepend：锚定原头部消息的视觉位置。 + // 不触发自动滚底，也不计入 newMessageCount（历史消息不是"新消息"）。 + const isHeadPrepend = !viewChanged && el !== null && prevFirstId !== null && firstId !== prevFirstId; + if (isHeadPrepend) { + const prevH = prevScrollHeightRef.current; + const newH = el!.scrollHeight; + if (newH > prevH) { + el!.scrollTop += newH - prevH; + } + prevScrollHeightRef.current = newH; + prevMessageCountRef.current = messages.length; + return; + } + if (el) prevScrollHeightRef.current = el.scrollHeight; + if (viewChanged) { // View switched (e.g. breadcrumb navigation): restore saved scroll position stopScrollTopAnimation(); @@ -190,6 +238,18 @@ export const MessageList = memo(function MessageList({ } }, [messages, viewKey, virtualizer, stopScrollTopAnimation]); + // ---- 历史分页：内容不足一屏时自动补页 ---- + // 首屏/补页后若无滚动条（scrollTop 恒为 0，永远不触发触顶事件）， + // 只要还有更早的消息就继续请求，直到出现滚动条或加载完毕。 + useEffect(() => { + if (!hasMoreOlder || loadingOlder) return; + const el = containerRef.current; + if (!el || messages.length === 0) return; + if (el.scrollHeight <= el.clientHeight) { + onLoadOlderRef.current?.(); + } + }, [messages, hasMoreOlder, loadingOlder]); + // ---- mount: always scroll to bottom if messages already loaded ---- useEffect(() => { @@ -303,7 +363,12 @@ export const MessageList = memo(function MessageList({ return (
-
+
{/* 虚拟化容器：总高度撑开滚动条，子项绝对定位 */}
{virtualItems.map((vi) => { @@ -337,6 +402,19 @@ export const MessageList = memo(function MessageList({
+ {/* 历史分页加载指示 */} + {loadingOlder && ( +
+ + 加载历史消息… +
+ )} + {/* 浮动导航按钮 — 底部居中并排 */} {showScrollToBottom && (
diff --git a/web/src/hooks/chat/useMessages.ts b/web/src/hooks/chat/useMessages.ts index b0268f6..61a86f4 100644 --- a/web/src/hooks/chat/useMessages.ts +++ b/web/src/hooks/chat/useMessages.ts @@ -16,6 +16,7 @@ import type { ToolCall, ToolResult, ToolPending, + TopicHistoryEnd, ExecutionCompleted, WsError, TaskStarted, @@ -51,6 +52,14 @@ export interface UseMessagesReturn { handleStop: () => Command; /** 处理主视图的消息类 case（task_started, stream_*, tool_*, execution_*, error），返回是否已处理 */ handleMainViewMessage: (message: WsOutbound) => boolean; + /** 历史分页：是否还有更早的消息可加载 */ + hasMoreOlder: boolean; + /** 历史分页：当前最早已加载消息的 seq 游标 */ + oldestSeq: number | null; + /** 历史分页：是否正在加载更早一页 */ + loadingOlder: boolean; + /** 触顶时请求加载更早一页（返回待发送命令，由调用方发送） */ + requestLoadOlder: () => Command | null; } export function useMessages(options: UseMessagesOptions): UseMessagesReturn { @@ -223,22 +232,162 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { if (selectedTopicRef.current) markTopicProcessing(selectedTopicRef.current); }, [selectedTopicRef, markTopicProcessing]); + // ---- 历史分页（load_older_messages） ---- + // 历史批次消息（带 seq）先缓存在 ref，收到 topic_history_end 时一次性 + // 去重后 prepend 到列表头部：避免逐条 setState 且保证批次内顺序稳定。 + const pendingHistoryRef = useRef([]); + const [olderHistory, setOlderHistory] = useState<{ + hasMore: boolean; + oldestSeq: number | null; + loading: boolean; + }>({ hasMore: false, oldestSeq: null, loading: false }); + + const oldestSeqRef = useRef(null); + oldestSeqRef.current = olderHistory.oldestSeq; + const clearMessages = useCallback(() => { clearStreaming(); setMessages([]); + // 重置历史分页状态与未 flush 的批次缓存 + pendingHistoryRef.current = []; + setOlderHistory({ hasMore: false, oldestSeq: null, loading: false }); }, [clearStreaming]); + const requestLoadOlder = useCallback((): Command | null => { + const topicId = selectedTopicRef.current; + const before = oldestSeqRef.current; + if (!topicId || before === null || olderHistory.loading || !olderHistory.hasMore) { + return null; + } + setOlderHistory((prev) => (prev.loading ? prev : { ...prev, loading: true })); + // 超时自愈：topic_history_end 因断连/异常永不到达时解除 loading 锁， + // 避免历史分页永久卡死（幂等：end 先到则 loading 已为 false，无副作用） + setTimeout(() => { + setOlderHistory((prev) => (prev.loading ? { ...prev, loading: false } : prev)); + }, 10000); + return { type: 'load_older_messages', topic_id: topicId, before_seq: before }; + }, [olderHistory.loading, olderHistory.hasMore, selectedTopicRef]); + + /** topic_history_end 到达：flush 历史批次 + 更新分页游标 */ + const handleTopicHistoryEnd = useCallback((msg: TopicHistoryEnd) => { + if (msg.topic_id !== selectedTopicRef.current) return; + const batch = pendingHistoryRef.current; + pendingHistoryRef.current = []; + if (batch.length > 0) { + setMessages((prev) => { + const existing = new Set(prev.map((m) => m.id)); + const fresh = batch.filter((m) => !existing.has(m.id)); + if (fresh.length === 0) return prev; + return [...fresh, ...prev]; + }); + } + setOlderHistory({ + hasMore: msg.has_more, + oldestSeq: msg.oldest_seq ?? null, + loading: false, + }); + }, [selectedTopicRef]); + + /** 历史批次消息（带 seq）转 ChatMessage 缓存；返回 true 表示已处理 */ + const tryCollectHistoryMessage = useCallback( + (message: WsOutbound): boolean => { + const seq = (message as { seq?: number }).seq; + if (seq === undefined) return false; + // 历史批次必带 topic_id（后端历史路径填充）。不匹配（切话题瞬间在途的 + // 旧批次）或不带（任务会话消息）的交回常规分支：前者被 per-case 的 + // topic_id 检查丢弃，后者维持原有的实时消息处理路径。 + const batchTopicId = (message as { topic_id?: string }).topic_id; + if (batchTopicId !== selectedTopicRef.current) return false; + let converted: ChatMessage | null = null; + const m = message as + | AssistantResponse + | ToolCall + | ToolResult + | ToolPending; + switch (m.type) { + case 'assistant_response': + converted = { + id: m.id, + role: m.role === 'user' || m.role === 'tool' ? m.role : 'assistant', + content: m.content, + timestamp: m.timestamp ?? Math.floor(Date.now() / 1000), + seq, + type: 'message', + attachments: m.attachments, + reasoningContent: m.reasoning_content, + }; + break; + case 'tool_call': + converted = { + id: m.id, + role: 'tool', + content: m.content, + timestamp: m.timestamp ?? Math.floor(Date.now() / 1000), + seq, + type: 'tool_call', + toolName: m.tool_name, + toolCallId: m.tool_call_id, + arguments: m.arguments, + reasoningContent: m.reasoning_content, + }; + break; + case 'tool_result': + converted = { + id: m.id, + role: 'tool', + content: m.content, + timestamp: m.timestamp ?? Math.floor(Date.now() / 1000), + seq, + type: 'tool_result', + toolName: m.tool_name, + toolCallId: m.tool_call_id, + durationMs: m.duration_ms, + }; + break; + case 'tool_pending': + converted = { + id: m.id, + role: 'tool', + content: `${m.content}\n\n${m.resume_hint}`, + timestamp: m.timestamp ?? Math.floor(Date.now() / 1000), + seq, + type: 'tool_pending', + toolName: m.tool_name, + toolCallId: m.tool_call_id, + }; + break; + default: + return false; + } + pendingHistoryRef.current.push(converted); + return true; + }, + [], + ); + const handleStop = useCallback((): Command => { return { type: 'stop_execution' }; }, []); const handleMainViewMessage = useCallback( (message: WsOutbound): boolean => { - // 非流式消息到达前，先把 pending 的流式内容落盘，避免被后续消息覆盖或丢失 - if (message.type !== 'stream_delta') { + // 非流式消息到达前，先把 pending 的流式内容落盘，避免被后续消息覆盖或丢失。 + // 例外：历史分页批次（带 seq）与 topic_history_end 是历史数据回放， + // 与活动流无关——若在流式输出进行中触顶加载历史，误清累加器会导致 + // 后续 delta 从零累积并覆写壳内容，造成已流出文本丢失。 + if ( + message.type !== 'stream_delta' && + message.type !== 'topic_history_end' && + (message as { seq?: number }).seq === undefined + ) { finishStreaming(); } switch (message.type) { + case 'topic_history_end': { + handleTopicHistoryEnd(message as TopicHistoryEnd); + return true; + } + case 'task_started': { const msg = message as TaskStarted; // 只 backfill 当前话题的 task tool_call，避免跨话题串扰 @@ -359,6 +508,7 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { } case 'assistant_response': { + if (tryCollectHistoryMessage(message)) return true; const msg = message as AssistantResponse; if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true; const role = msg.role === 'user' || msg.role === 'tool' ? msg.role : 'assistant'; @@ -390,6 +540,7 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { } case 'tool_call': { + if (tryCollectHistoryMessage(message)) return true; const msg = message as ToolCall; if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true; setMessages((prev) => [ @@ -412,6 +563,7 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { } case 'tool_result': { + if (tryCollectHistoryMessage(message)) return true; const msg = message as ToolResult; if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true; setMessages((prev) => [ @@ -432,6 +584,7 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { } case 'tool_pending': { + if (tryCollectHistoryMessage(message)) return true; const msg = message as ToolPending; if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true; setMessages((prev) => [ @@ -511,5 +664,9 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { finishStreaming, handleStop, handleMainViewMessage, + hasMoreOlder: olderHistory.hasMore, + oldestSeq: olderHistory.oldestSeq, + loadingOlder: olderHistory.loading, + requestLoadOlder, }; } diff --git a/web/src/hooks/useChat.ts b/web/src/hooks/useChat.ts index 0dcae29..9967af6 100644 --- a/web/src/hooks/useChat.ts +++ b/web/src/hooks/useChat.ts @@ -44,6 +44,11 @@ interface UseChatReturn { messages: ChatMessage[]; isLoading: boolean; + // 历史分页 + hasMoreOlder: boolean; + loadingOlder: boolean; + loadOlderMessages: () => void; + // 通道状态 channels: Channel[]; selectedChannel: string; @@ -352,6 +357,12 @@ export function useChat(): UseChatReturn { [sessions.selectedSessionId], ); + // ---- 历史分页：触顶时请求更早一页（hook 内部含 loading/hasMore/游标守卫） ---- + const loadOlderMessages = useCallback(() => { + const cmd = messages.requestLoadOlder(); + if (cmd) conn.sendCommand(cmd); + }, [messages.requestLoadOlder, conn.sendCommand]); + // ---- 委托方法 ---- const requestSessionList = useCallback((): Command => { return sessions.requestSessionList(sideData.selectedChannel); @@ -388,6 +399,9 @@ export function useChat(): UseChatReturn { setSelectedTopic: topics.setSelectedTopic, messages: resolvedMessages, isLoading: messages.isLoading, + hasMoreOlder: messages.hasMoreOlder, + loadingOlder: messages.loadingOlder, + loadOlderMessages, isReadOnly, isWritable: sideData.isWritable, channels: sideData.channels, diff --git a/web/src/types/protocol.ts b/web/src/types/protocol.ts index 3adea31..14d35bc 100644 --- a/web/src/types/protocol.ts +++ b/web/src/types/protocol.ts @@ -47,6 +47,8 @@ export interface AssistantResponse { timestamp?: number; reasoning_content?: string; user_message_id?: string; + /** 历史分页游标：仅历史加载批次携带（实时推送无此字段） */ + seq?: number; } export interface ToolCall { @@ -62,6 +64,8 @@ export interface ToolCall { timestamp?: number; reasoning_content?: string; user_message_id?: string; + /** 历史分页游标：仅历史加载批次携带（实时推送无此字段） */ + seq?: number; } export interface ToolResult { @@ -75,6 +79,8 @@ export interface ToolResult { topic_id?: string; duration_ms?: number; timestamp?: number; + /** 历史分页游标：仅历史加载批次携带（实时推送无此字段） */ + seq?: number; } export interface ToolPending { @@ -88,6 +94,16 @@ export interface ToolPending { subagent_task_id?: string; topic_id?: string; timestamp?: number; + /** 历史分页游标：仅历史加载批次携带（实时推送无此字段） */ + seq?: number; +} + +/** 话题历史批次结束标记：更新分页游标与 has_more 状态 */ +export interface TopicHistoryEnd { + type: 'topic_history_end'; + topic_id: string; + has_more: boolean; + oldest_seq?: number; } export interface WsError { @@ -315,6 +331,7 @@ export type WsOutbound = | ToolCall | ToolResult | ToolPending + | TopicHistoryEnd | WsError | TaskStarted | StreamDelta @@ -411,6 +428,12 @@ export interface LoadChatMessagesCommand { chat_id: string; } +export interface LoadOlderMessagesCommand { + type: 'load_older_messages'; + topic_id: string; + before_seq: number; +} + export interface DeleteTopicCommand { type: 'delete_topic'; topic_id: string; @@ -472,6 +495,7 @@ export type Command = | LoadTaskMessagesCommand | ListSchedulerJobsCommand | LoadChatMessagesCommand + | LoadOlderMessagesCommand | DeleteTopicCommand | RenameTopicCommand | StopExecutionCommand @@ -491,6 +515,8 @@ export interface ChatMessage { role: 'user' | 'assistant' | 'tool'; content: string; timestamp: number; + /** 历史分页游标（仅历史批次消息携带，用于触顶加载更早页） */ + seq?: number; type?: 'message' | 'tool_call' | 'tool_result' | 'tool_pending' | 'merged_tool'; toolName?: string; toolCallId?: string;