perf(chat): 长话题历史分页——messages 按 seq keyset 增量加载
后端新增 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 自愈；流式输出中加载历史不清空流式累加器，避免已流出文本丢失。
This commit is contained in:
oudecheng
parent
0159227828
commit
a629486ad3
@ -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,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@ -53,6 +53,10 @@ pub struct ChatMessage {
|
||||
pub content: String,
|
||||
pub media_refs: Vec<String>, // Paths to media files for context
|
||||
pub timestamp: i64,
|
||||
/// 会话内单调递增序号（DB 分配）。仅历史加载路径填充；
|
||||
/// 实时推送路径不填，前端以此区分"历史批次"与"实时消息"。
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub seq: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub system_context: Option<String>,
|
||||
#[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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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<CommandMetadata> {
|
||||
Some(CommandMetadata {
|
||||
name: "load_older_messages",
|
||||
description: "分页加载话题更早的历史消息",
|
||||
usage: "/load_older_messages <topic_id> <before_seq>",
|
||||
})
|
||||
}
|
||||
|
||||
async fn handle(
|
||||
&self,
|
||||
cmd: Command,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
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!(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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::<i64>().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<crate::storage::SessionStore>,
|
||||
session_id: &str,
|
||||
@ -862,26 +895,40 @@ async fn send_topic_history(
|
||||
sender: &mpsc::Sender<WsOutbound>,
|
||||
task_repository: &Arc<dyn TaskRepository>,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
// 加载话题消息，按 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<String> = std::collections::HashSet::new();
|
||||
let mut tool_call_ids_with_results: std::collections::HashSet<String> =
|
||||
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<Vec<WsOutbound>> = 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::<Vec<_>>()
|
||||
})
|
||||
.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<crate::storage::SessionStore>,
|
||||
session_id: &str,
|
||||
topic_id: &str,
|
||||
before_seq: i64,
|
||||
sender: &mpsc::Sender<WsOutbound>,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
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<Vec<WsOutbound>> = tokio::task::spawn_blocking(move || {
|
||||
messages
|
||||
.iter()
|
||||
.map(|m| chat_message_to_ws_outbound(m, Some(&topic_id_for_convert)))
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.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<WsOutbound> {
|
||||
fn chat_message_to_ws_outbound(
|
||||
msg: &crate::bus::ChatMessage,
|
||||
topic_id: Option<&str>,
|
||||
) -> Vec<WsOutbound> {
|
||||
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<WsOutbound>
|
||||
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<WsOutbound>
|
||||
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<WsOutbound>
|
||||
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<WsOutbound>
|
||||
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<WsOutbound>
|
||||
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<WsOutbound>
|
||||
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(),
|
||||
}
|
||||
|
||||
@ -176,6 +176,9 @@ pub enum WsOutbound {
|
||||
reasoning_content: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
user_message_id: Option<String>,
|
||||
/// 历史分页游标：仅历史加载批次填充（实时推送为 None）
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
seq: Option<i64>,
|
||||
},
|
||||
#[serde(rename = "tool_call")]
|
||||
ToolCall {
|
||||
@ -195,6 +198,9 @@ pub enum WsOutbound {
|
||||
reasoning_content: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
user_message_id: Option<String>,
|
||||
/// 历史分页游标：仅历史加载批次填充（实时推送为 None）
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
seq: Option<i64>,
|
||||
},
|
||||
#[serde(rename = "tool_result")]
|
||||
ToolResult {
|
||||
@ -211,6 +217,9 @@ pub enum WsOutbound {
|
||||
duration_ms: Option<u64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
timestamp: Option<i64>,
|
||||
/// 历史分页游标：仅历史加载批次填充（实时推送为 None）
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
seq: Option<i64>,
|
||||
},
|
||||
#[serde(rename = "tool_pending")]
|
||||
ToolPending {
|
||||
@ -226,6 +235,18 @@ pub enum WsOutbound {
|
||||
topic_id: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
timestamp: Option<i64>,
|
||||
/// 历史分页游标：仅历史加载批次填充（实时推送为 None）
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
seq: Option<i64>,
|
||||
},
|
||||
/// 话题历史批次结束标记：随每批历史消息末尾发送，
|
||||
/// 前端据此更新分页游标（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<i64>,
|
||||
},
|
||||
#[serde(rename = "error")]
|
||||
Error {
|
||||
|
||||
@ -28,6 +28,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec<WsOutb
|
||||
timestamp: None,
|
||||
reasoning_content: message.reasoning_content.clone(),
|
||||
user_message_id: None,
|
||||
seq: None,
|
||||
});
|
||||
}
|
||||
|
||||
@ -49,6 +50,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec<WsOutb
|
||||
timestamp: None,
|
||||
reasoning_content: tc_reasoning.clone(),
|
||||
user_message_id: None,
|
||||
seq: None,
|
||||
}));
|
||||
outbound
|
||||
} else {
|
||||
@ -62,6 +64,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec<WsOutb
|
||||
timestamp: None,
|
||||
reasoning_content: message.reasoning_content.clone(),
|
||||
user_message_id: None,
|
||||
seq: None,
|
||||
}]
|
||||
}
|
||||
}
|
||||
@ -83,6 +86,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec<WsOutb
|
||||
topic_id: None,
|
||||
duration_ms: None,
|
||||
timestamp: None,
|
||||
seq: None,
|
||||
}],
|
||||
ToolMessageState::PendingUserAction => vec![WsOutbound::ToolPending {
|
||||
id: message
|
||||
@ -97,6 +101,7 @@ pub(crate) fn ws_outbound_from_chat_message(message: &ChatMessage) -> Vec<WsOutb
|
||||
subagent_task_id: None,
|
||||
topic_id: None,
|
||||
timestamp: None,
|
||||
seq: None,
|
||||
}],
|
||||
},
|
||||
_ => 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(),
|
||||
|
||||
@ -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<i64>,
|
||||
limit: usize,
|
||||
) -> Result<(Vec<ChatMessage>, bool), StorageError> {
|
||||
let conn = self.pool.get()?;
|
||||
// 多取 1 条仅用于判断 has_more，截断后返回 limit 条
|
||||
let fetch = limit + 1;
|
||||
let mut messages: Vec<ChatMessage> = 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<i64>>(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<i64>>(10)?.map(|v| v as u64),
|
||||
tool_calls,
|
||||
usage: map_usage_row(row, 11, 12, 13, 14, 15)?,
|
||||
seq: row.get(16)?,
|
||||
})
|
||||
})?;
|
||||
|
||||
|
||||
@ -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<i64>>(10)?.map(|v| v as u64),
|
||||
tool_calls,
|
||||
usage: map_usage_row(row, 11, 12, 13, 14, 15)?,
|
||||
seq: row.get(16)?,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@ -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}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
|
||||
@ -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<HTMLDivElement>(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<string | null>(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 (
|
||||
<div className="relative h-full">
|
||||
<div ref={containerRef} onScroll={handleScroll} className="h-full overflow-y-auto p-6">
|
||||
<div
|
||||
ref={containerRef}
|
||||
onScroll={handleScroll}
|
||||
className="h-full overflow-y-auto p-6"
|
||||
style={{ overflowAnchor: 'none' }}
|
||||
>
|
||||
{/* 虚拟化容器：总高度撑开滚动条，子项绝对定位 */}
|
||||
<div style={{ height: `${totalSize}px`, position: 'relative' }}>
|
||||
{virtualItems.map((vi) => {
|
||||
@ -337,6 +402,19 @@ export const MessageList = memo(function MessageList({
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* 历史分页加载指示 */}
|
||||
{loadingOlder && (
|
||||
<div
|
||||
className="absolute top-3 left-1/2 z-10 flex -translate-x-1/2 items-center gap-1.5
|
||||
rounded-full border border-[var(--border-color)] bg-[var(--bg-tertiary)]/90
|
||||
px-3 py-1.5 shadow-sm backdrop-blur-md"
|
||||
aria-live="polite"
|
||||
>
|
||||
<Loader2 className="h-3.5 w-3.5 animate-spin text-[var(--accent-cyan)]" />
|
||||
<span className="text-xs text-[var(--text-secondary)]">加载历史消息…</span>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* 浮动导航按钮 — 底部居中并排 */}
|
||||
{showScrollToBottom && (
|
||||
<div className="absolute bottom-4 left-1/2 -translate-x-1/2 z-10 flex items-center gap-2">
|
||||
|
||||
@ -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<ChatMessage[]>([]);
|
||||
const [olderHistory, setOlderHistory] = useState<{
|
||||
hasMore: boolean;
|
||||
oldestSeq: number | null;
|
||||
loading: boolean;
|
||||
}>({ hasMore: false, oldestSeq: null, loading: false });
|
||||
|
||||
const oldestSeqRef = useRef<number | null>(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,
|
||||
};
|
||||
}
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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;
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user