ooodc/PicoBot
ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

perf: 第二批性能修复——blocking 线程池隔离、HTTP 客户端复用、前端 memo 与流式节流

Browse Source 
- 同步阻塞操作（附件处理、历史加载、scheduler/memory_search 的 SQLite 调用）
  移入 spawn_blocking，避免占用 async worker
- LLM Provider reqwest::Client 按超时配置缓存复用，减少 TLS/连接开销
- agent loop：图片过滤加廉价预判避免全量深拷贝；请求克隆改借用；工具定义 Arc 化
- 定向 COUNT/LIMIT 1 查询替代全量加载计数（wait_coordinator、task session 重建）
- 前端：面板/侧栏/聊天组件 memo 化；merged_tool 对象按值复用缓存；
  流式 delta rAF 节流批量 flush；useMemo 缓存分组排序结果
This commit is contained in:
oudecheng 2026-08-18 06:55:21 +08:00
parent e7deac6950
commit 0159227828
30 changed files with 581 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/agent/agent_loop.rs
View File

@ -258,6 +258,43 @@ fn filter_images_by_age_and_count(
// 消息列表顺序：[old, ..., new]，所以末尾是最新的
let msg_count = messages.len();
// 廉价预判：完整复刻下方保留逻辑，仅判断是否"确有图片会被移除"。
// 有图片但无需过滤时（常见情形）直接返回借用，避免每轮 LLM 迭代
// 对整个历史做全量深拷贝。
let mut dry_run_kept = 0usize;
for (idx, message) in messages.iter().enumerate().rev() {
let age_from_end = msg_count.saturating_sub(idx).saturating_sub(1);
let exceeds_age_limit = max_age_rounds > 0 && age_from_end >= max_age_rounds;
let image_count_in_msg = message
.media_refs
.iter()
.filter(|p| supported_image_mime_type(p).is_some())
.count();
if image_count_in_msg == 0 {
continue;
}
if exceeds_age_limit {
// 该消息的图片全部会被过滤
return filter_images_by_age_and_count_inner(messages, max_age_rounds, max_images);
}
let can_keep = std::cmp::min(image_count_in_msg, max_images.saturating_sub(dry_run_kept));
if can_keep < image_count_in_msg {
// 超出数量上限的图片会被过滤
return filter_images_by_age_and_count_inner(messages, max_age_rounds, max_images);
}
dry_run_kept += can_keep;
}
Cow::Borrowed(messages)
}
/// `filter_images_by_age_and_count` 的重建实现：仅在预判确认有图片需要过滤时调用。
fn filter_images_by_age_and_count_inner(
messages: &[ChatMessage],
max_age_rounds: usize,
max_images: usize,
) -> Cow<'_, [ChatMessage]> {
let msg_count = messages.len();
// 先从后向前遍历，计算每条消息应该保留多少张图片
// 使用 Vec<usize> 存储每条消息应该保留的图片数量
let mut images_to_keep_per_msg: Vec<usize> = vec![0; msg_count];
@ -1132,7 +1169,8 @@ impl AgentLoop {
let tools = if tool_defs.is_empty() {
None
} else {
Some(tool_defs)
// Arc 共享：process() 内多轮迭代只读复用，避免每轮深拷贝全部工具定义
Some(std::sync::Arc::new(tool_defs))
};
// 工具 token 估算在循环外算一次（tool_defs 在 process() 期间不变），
// 避免每轮 serde_json::to_string 全量序列化工具定义。
@ -1247,7 +1285,7 @@ impl AgentLoop {
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
return Ok(cancel);
}
result = self.provider.chat_with_streaming(request.clone(), stream_callback.clone()) => {
result = self.provider.chat_with_streaming(&request, stream_callback.clone()) => {
llm_result = result;
}
}
@ -1255,7 +1293,7 @@ impl AgentLoop {
} else {
llm_result = self
.provider
.chat_with_streaming(request.clone(), stream_callback)
.chat_with_streaming(&request, stream_callback)
.await;
}
@ -1636,7 +1674,7 @@ impl AgentLoop {
&self,
messages: &[ChatMessage],
system_prompt_context: Option<&SystemPromptContext>,
tools: Option<Vec<crate::domain::tools::Tool>>,
tools: Option<std::sync::Arc<Vec<crate::domain::tools::Tool>>>,
tools_tokens: usize,
) -> ChatCompletionRequest {
let filtered_messages = filter_images_by_age_and_count(
@ -1827,12 +1865,12 @@ impl AgentLoop {
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
return cancel;
}
result = self.provider.chat(request.clone()) => {
result = self.provider.chat(&request) => {
final_result = result;
}
}
} else {
final_result = self.provider.chat(request.clone()).await;
final_result = self.provider.chat(&request).await;
}
match final_result {
@ -3320,7 +3358,7 @@ mod tests {
impl LLMProvider for MockProvider {
async fn chat(
&self,
_request: ChatCompletionRequest,
_request: &ChatCompletionRequest,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>> {
let mut responses = self.responses.lock().unwrap();
if responses.is_empty() {

4
src/agent/context_compressor.rs
View File

@ -458,7 +458,7 @@ OLDER SEGMENT (events from earlier in the session):
};
let response = provider
.chat(request)
.chat(&request)
.await
.map_err(|e| AgentError::LlmError(e.to_string()))?;
Ok(response.content)
@ -540,7 +540,7 @@ OLDER SEGMENT (events from earlier in the session):
};
let response = provider
.chat(request)
.chat(&request)
.await
.map_err(|e| AgentError::LlmError(e.to_string()))?;
Ok(response.content)

8
src/command/handlers/list_sessions.rs
View File

@ -73,11 +73,9 @@ async fn handle_list_sessions(
let is_current = topic.id == current_topic_id;
let marker = if is_current { " *" } else { "" };
// 使用辅助方法获取消息数量
let msg_count = handler
.store
.get_topic_message_count(&topic.id)
.unwrap_or(0);
// 使用 topics 表预计算列，避免循环内 N+1 次 COUNT 查询
// （message_count 与 list_topics 的展示口径一致）
let msg_count = topic.message_count;
lines.push(format!(
"{}. {}{} ({})",

12
src/command/handlers/load_task_messages.rs
View File

@ -141,15 +141,15 @@ fn reconstruct_task_from_db(
store: &SessionStore,
task_id: &str,
) -> Result<Option<TaskSession>, CommandError> {
let sessions = store
.find_sessions_by_id_suffix(&format!(":{}", task_id))
let record = store
.find_first_session_by_id_suffix(&format!(":{}", task_id))
.map_err(|e| CommandError::new("DB_ERROR", e.to_string()))?;
if sessions.is_empty() {
return Ok(None);
}
let record = match record {
Some(record) => record,
None => return Ok(None),
};
let record = &sessions[0];
let session_id = record.id.clone();
// Extract parent_session_id from session_id: "sub:{parent}:task:{uuid}"

20
src/command/handlers/save_session.rs
View File

@ -33,7 +33,7 @@ pub async fn save_session_to_file(
store: &SessionStore,
task_repository: Option<&dyn TaskRepository>,
system_prompt_provider: &dyn SystemPromptProvider,
) -> Result<PathBuf, String> {
) -> Result<(PathBuf, usize), String> {
// 获取会话记录
let record = store
.get_session(session_id)
@ -83,7 +83,8 @@ pub async fn save_session_to_file(
// 写入文件
std::fs::write(&output_path, markdown).map_err(|e| format!("Failed to write file: {}", e))?;
Ok(output_path)
// 返回已加载的消息数，调用方无需为此再次全量加载消息
Ok((output_path, messages.len()))
}
/// 保存会话命令处理器
@ -187,8 +188,8 @@ async fn handle_save_session(
}
}
// 调用公共函数
let output_path = save_session_to_file(
// 调用公共函数（返回路径与已加载消息数，避免二次全量加载只为计数）
let (output_path, message_count) = save_session_to_file(
session_id,
filepath,
include_all,
@ -200,15 +201,6 @@ async fn handle_save_session(
.await
.map_err(|e| CommandError::new("SAVE_ERROR", e))?;
// 根据 include_all 获取消息数量
let message_count = if include_all {
handler.store.load_all_messages(session_id)
} else {
handler.store.load_messages(session_id)
}
.map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?
.len();
Ok(CommandResponse::success(ctx.request_id)
.with_message(
MessageKind::Notification,
@ -716,7 +708,7 @@ impl InChatCommandHandler for SaveSessionInChatHandler {
// 返回成功或失败消息
match result {
Ok(output_path) => {
Ok((output_path, _message_count)) => {
let msg = format!(
"Session saved to: {}",
output_path.display().to_string().replace('\\', "/")

4
src/gateway/memory_maintenance.rs
View File

@ -192,7 +192,7 @@ impl MemoryMaintenanceService {
.chain(std::iter::once(None))
.enumerate()
{
let response = match provider.chat(request.clone()).await {
let response = match provider.chat(&request).await {
Ok(success) => success,
Err(err) => {
let error_text = err.to_string();
@ -313,7 +313,7 @@ impl MemoryMaintenanceService {
.chain(std::iter::once(None))
.enumerate()
{
match provider.chat(request.clone()).await {
match provider.chat(&request).await {
Ok(success) => {
response = Some(success);
break;

5
src/gateway/wait_coordinator.rs
View File

@ -117,11 +117,10 @@ impl WaitCoordinator for SessionWaitCoordinator {
};
// 3.5. 记录等待前的用户消息数量（用于 wakeup 后提取新注入的消息）
// 直接从 SQLite 读取，不持有任何锁
// 直接从 SQLite 读取（定向 COUNT，避免全量加载消息体），不持有任何锁
let user_msg_count_before = self
.store
.load_messages_for_topic(&self.topic_id, None)
.map(|msgs| msgs.iter().filter(|m| m.role == "user").count())
.count_user_messages_for_topic(&self.topic_id)
.unwrap_or(0);
// 4. 释放 serial_lock（取出 guard 并 drop）

178
src/gateway/ws.rs
View File

@ -186,23 +186,43 @@ async fn handle_socket(ws: WebSocket, state: Arc<GatewayState>) {
let store = state.session_manager.store();
// 1. 查询 websocket 和 cli 两个通道的 Sessions（兼容旧版本 cli 通道创建的会话）
let mut websocket_sessions = store.list_sessions("websocket", false).unwrap_or_default();
let cli_channel_sessions = store.list_sessions("cli", false).unwrap_or_default();
websocket_sessions.extend(cli_channel_sessions);
websocket_sessions.sort_by_key(|s| -(s.last_active_at));
// SQLite 同步查询移入 blocking 线程池，避免连接建立时阻塞 async worker。
let store_for_init = store.clone();
let cli_sessions_for_init = cli_sessions.clone();
let (mut websocket_sessions, initial_result) = tokio::task::spawn_blocking(move || {
let mut websocket_sessions = store_for_init
.list_sessions("websocket", false)
.unwrap_or_default();
let cli_channel_sessions = store_for_init
.list_sessions("cli", false)
.unwrap_or_default();
websocket_sessions.extend(cli_channel_sessions);
websocket_sessions.sort_by_key(|s| -(s.last_active_at));
// 2. 如果没有，自动创建一个默认 Session
let initial_record = if websocket_sessions.is_empty() {
match cli_sessions.create_with_channel("websocket", Some("默认会话")) {
Ok(record) => record,
Err(e) => {
tracing::error!(error = %e, "Failed to create initial WebSocket session");
return;
// 2. 如果没有，自动创建一个默认 Session
let initial_record = if websocket_sessions.is_empty() {
match cli_sessions_for_init.create_with_channel("websocket", Some("默认会话")) {
Ok(record) => Some(record),
Err(e) => {
tracing::error!(error = %e, "Failed to create initial WebSocket session");
None
}
}
}
} else {
// 使用最新的 Session
websocket_sessions[0].clone()
} else {
// 使用最新的 Session
Some(websocket_sessions[0].clone())
};
(websocket_sessions, initial_record)
})
.await
.unwrap_or_else(|e| {
tracing::error!(error = %e, "WebSocket session init task failed");
(Vec::new(), None)
});
let initial_record = match initial_result {
Some(record) => record,
None => return,
};
let runtime_session_id = uuid::Uuid::new_v4().to_string();
@ -414,7 +434,14 @@ async fn handle_inbound(
.await;
// Process attachments: save base64 content to local files and build MediaItems with correct paths
let media = process_attachments_with_base64(attachments)?;
// base64 解码 + 同步写盘是 CPU/IO 密集操作（单条消息最大约 67MB），
// 移入 blocking 线程池，避免长时间霸占 async worker。
let media =
tokio::task::spawn_blocking(move || process_attachments_with_base64(attachments))
.await
.map_err(|e| {
AgentError::Other(format!("Attachment processing task failed: {}", e))
})??;
state
.bus
@ -720,24 +747,31 @@ async fn handle_inbound(
let _ = sender.send(WsOutbound::MemoryList { memories }).await;
}
// 记忆 CRUD 后自动刷新列表
if response.metadata.get("memory_updated").map(|v| v.as_str()) == Some("true")
&& let Ok(records) =
store.list_memories_for_scope("user", crate::storage::GLOBAL_SCOPE_KEY)
{
let memories: Vec<crate::protocol::MemorySummary> = records
.into_iter()
.filter(|m| m.namespace != "_meta")
.map(|m| crate::protocol::MemorySummary {
id: m.id,
namespace: m.namespace,
memory_key: m.memory_key,
content: m.content,
created_at: m.created_at,
updated_at: m.updated_at,
})
.collect();
let _ = sender.send(WsOutbound::MemoryList { memories }).await;
// 记忆 CRUD 后自动刷新列表（SQLite 同步查询移入 blocking 线程池）
if response.metadata.get("memory_updated").map(|v| v.as_str()) == Some("true") {
let store_bg = store.clone();
let records = tokio::task::spawn_blocking(move || {
store_bg.list_memories_for_scope("user", crate::storage::GLOBAL_SCOPE_KEY)
})
.await;
if let Err(e) = &records {
tracing::warn!(error = %e, "Memory list task failed");
}
if let Ok(Ok(records)) = records {
let memories: Vec<crate::protocol::MemorySummary> = records
.into_iter()
.filter(|m| m.namespace != "_meta")
.map(|m| crate::protocol::MemorySummary {
id: m.id,
namespace: m.namespace,
memory_key: m.memory_key,
content: m.content,
created_at: m.created_at,
updated_at: m.updated_at,
})
.collect();
let _ = sender.send(WsOutbound::MemoryList { memories }).await;
}
}
// 处理加载聊天消息请求
@ -828,14 +862,19 @@ async fn send_topic_history(
sender: &mpsc::Sender<WsOutbound>,
task_repository: &Arc<dyn TaskRepository>,
) -> Result<(), Box<dyn std::error::Error>> {
// 加载话题消息，按 session_id 过滤，避免混入子智能体消息
let mut messages = store.load_messages_for_topic_full(topic_id, Some(session_id))?;
// 对账 running 占位：DB 中的 task tool_result 永远保持 spawn 时的 running 状态
// （实时完成信号只更新前端内存与 pending_subagents 表），若不替换，
// 刷新/切话题后前端卡片会永远显示"运行中"。与 Session::reconcile_running_placeholders
// 同语义，仅改发送副本，不落库。
reconcile_running_in_messages(&mut messages, store, topic_id);
// 加载话题消息，按 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))?;
reconcile_running_in_messages(&mut messages, &store_bg, &topic_id_bg);
Ok::<_, crate::storage::StorageError>(messages)
})
.await
.map_err(|e| format!("Topic history load task failed: {}", e))??;
tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history");
@ -850,11 +889,19 @@ async fn send_topic_history(
}
}
// 将消息转换为 WsOutbound 并发送
for msg in messages {
for outbound in chat_message_to_ws_outbound(&msg) {
let _ = sender.send(outbound).await;
}
// 将消息转换为 WsOutbound 并发送。
// 转换过程对每条媒体引用做同步文件读取 + base64 编码（CPU/IO 密集），
// 整体移入 blocking 线程池一次性产出，避免阻塞 async worker。
let outbound_batches: Vec<Vec<WsOutbound>> = tokio::task::spawn_blocking(move || {
messages
.iter()
.map(chat_message_to_ws_outbound)
.collect::<Vec<_>>()
})
.await
.map_err(|e| format!("Topic history convert task failed: {}", e))?;
for outbound in outbound_batches.into_iter().flatten() {
let _ = sender.send(outbound).await;
}
// 查询该话题下所有子智能体任务，补发 TaskStarted 事件
@ -963,7 +1010,12 @@ async fn send_task_messages(
subagent_task_id: Option<String>,
task_repository: Option<&Arc<dyn TaskRepository>>,
) -> Result<(), Box<dyn std::error::Error>> {
let messages = store.load_messages(session_id)?;
// SQLite 同步加载移入 blocking 线程池，避免阻塞 async worker。
let store_bg = store.clone();
let session_id_bg = session_id.to_string();
let messages = tokio::task::spawn_blocking(move || store_bg.load_messages(&session_id_bg))
.await
.map_err(|e| format!("Task messages load task failed: {}", e))??;
tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending task messages");
@ -978,16 +1030,26 @@ async fn send_task_messages(
}
}
for msg in messages {
let mut outbounds = chat_message_to_ws_outbound(&msg);
if let Some(ref task_id) = subagent_task_id {
for ob in &mut outbounds {
set_subagent_task_id(ob, task_id);
}
}
for outbound in outbounds {
let _ = sender.send(outbound).await;
}
// 转换（含媒体文件同步读取 + base64 编码）移入 blocking 线程池一次性产出
let subagent_task_id_bg = subagent_task_id.clone();
let outbound_batches: Vec<Vec<WsOutbound>> = tokio::task::spawn_blocking(move || {
messages
.iter()
.map(|msg| {
let mut outbounds = chat_message_to_ws_outbound(msg);
if let Some(ref task_id) = subagent_task_id_bg {
for ob in &mut outbounds {
set_subagent_task_id(ob, task_id);
}
}
outbounds
})
.collect::<Vec<_>>()
})
.await
.map_err(|e| format!("Task messages convert task failed: {}", e))?;
for outbound in outbound_batches.into_iter().flatten() {
let _ = sender.send(outbound).await;
}
// 补发子任务（孙智能体）的 TaskStarted 事件

12
src/providers/anthropic.rs
View File

@ -3,7 +3,6 @@ use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use super::traits::Usage;
use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, Tool, ToolCall};
@ -144,10 +143,9 @@ impl AnthropicProvider {
max_tokens: Option<u32>,
model_extra: HashMap<String, serde_json::Value>,
) -> Self {
let client = Client::builder()
.timeout(Duration::from_secs(llm_timeout_secs))
.build()
.unwrap_or_else(|_| Client::new());
// 复用按超时配置的共享 client（TLS 上下文 + 连接池），
// 避免每条消息重建 Provider 时重复构造。
let client = crate::providers::shared_llm_http_client(llm_timeout_secs);
// 兼容带末尾斜杠的 base_url，避免 format!("{}/v1/messages", base_url) 产生双斜杠
let base_url = base_url.trim_end_matches('/').to_string();
@ -258,7 +256,7 @@ impl LLMProvider for AnthropicProvider {
#[tracing::instrument(skip(self, request), fields(provider = %self.name, model = %self.model_id))]
async fn chat(
&self,
request: ChatCompletionRequest,
request: &ChatCompletionRequest,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>> {
let url = format!("{}/v1/messages", self.base_url);
let max_tokens = request.max_tokens.or(self.max_tokens).unwrap_or(8192);
@ -271,7 +269,7 @@ impl LLMProvider for AnthropicProvider {
"Anthropic: sending chat completion request"
);
let tools = request.tools.map(|tools| {
let tools = request.tools.as_ref().map(|tools| {
tools
.iter()
.map(|t: &Tool| AnthropicTool {

35
src/providers/mod.rs
View File

@ -12,6 +12,41 @@ pub use traits::{
StreamCallback, StreamDelta, Usage,
};
/// 共享 HTTP client 缓存：按超时配置复用 reqwest::Client。
///
/// reqwest::Client 持有 TLS 上下文与连接池，构造成本高；每条消息重建
/// Provider 会导致每次 LLM 请求都无法复用 keep-alive 连接。
/// api_key / base_url / extra_headers 均为 per-request 应用（不影响 client
/// 构造），因此缓存键只需超时值，不同 provider/模型可安全共享同一 client。
use std::collections::HashMap;
use std::sync::LazyLock;
use std::sync::Mutex;
static SHARED_HTTP_CLIENTS: LazyLock<Mutex<HashMap<u64, reqwest::Client>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
/// 缓存条目上限：超时配置的取值种类极少（来自模型配置的 llm_timeout_secs），
/// 超过上限说明配置在频繁变动，直接清空重建（代价仅为缓存失效）。
const SHARED_HTTP_CLIENT_CACHE_CAP: usize = 16;
pub(crate) fn shared_llm_http_client(timeout_secs: u64) -> reqwest::Client {
let mut cache = SHARED_HTTP_CLIENTS
.lock()
.unwrap_or_else(|e| e.into_inner());
if let Some(client) = cache.get(&timeout_secs) {
return client.clone();
}
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(timeout_secs))
.build()
.unwrap_or_else(|_| reqwest::Client::new());
if cache.len() >= SHARED_HTTP_CLIENT_CACHE_CAP {
cache.clear();
}
cache.insert(timeout_secs, client.clone());
client
}
pub fn create_provider(
config: ProviderRuntimeConfig,
) -> Result<Box<dyn LLMProvider>, ProviderError> {

21
src/providers/openai.rs
View File

@ -5,7 +5,6 @@ use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::time::Duration;
use super::traits::{StreamCallback, StreamDelta, Usage};
use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, ToolCall};
@ -278,10 +277,9 @@ impl OpenAIProvider {
max_tokens: Option<u32>,
model_extra: HashMap<String, serde_json::Value>,
) -> Self {
let client = Client::builder()
.timeout(Duration::from_secs(llm_timeout_secs))
.build()
.unwrap_or_else(|_| Client::new());
// 复用按超时配置的共享 client（TLS 上下文 + 连接池），
// 避免每条消息重建 Provider 时重复构造。
let client = crate::providers::shared_llm_http_client(llm_timeout_secs);
// 兼容带末尾斜杠的 base_url（如 https://opencode.ai/zen/go/v1/），
// 否则 format!("{}/chat/completions", base_url) 会产生双斜杠导致 404
@ -1103,12 +1101,12 @@ impl OpenAIUsage {
impl LLMProvider for OpenAIProvider {
async fn chat(
&self,
request: ChatCompletionRequest,
request: &ChatCompletionRequest,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>> {
// 检查是否启用流式输出
if self.is_streaming_enabled() {
// 优先尝试流式输出（无回调）
match self.chat_streaming_internal(&request, None).await {
match self.chat_streaming_internal(request, None).await {
Ok(response) => return Ok(response),
Err(e) => {
tracing::debug!(
@ -1127,7 +1125,7 @@ impl LLMProvider for OpenAIProvider {
// 非流式回退实现
let url = format!("{}/chat/completions", self.base_url);
let body = self.build_request_body(&request);
let body = self.build_request_body(request);
// Debug: Log LLM request summary (only in debug builds)
#[cfg(debug_assertions)]
@ -1266,14 +1264,11 @@ impl LLMProvider for OpenAIProvider {
async fn chat_with_streaming(
&self,
request: ChatCompletionRequest,
request: &ChatCompletionRequest,
callback: StreamCallback,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>> {
if self.is_streaming_enabled() {
match self
.chat_streaming_internal(&request, Some(&callback))
.await
{
match self.chat_streaming_internal(request, Some(&callback)).await {
Ok(response) => return Ok(response),
Err(e) => {
tracing::debug!(

9
src/providers/traits.rs
View File

@ -117,7 +117,10 @@ pub struct ChatCompletionRequest {
pub messages: Vec<Message>,
pub temperature: Option<f32>,
pub max_tokens: Option<u32>,
pub tools: Option<Vec<Tool>>,
/// 工具定义在单次 process() 内跨多轮 LLM 迭代只读复用，
/// 用 Arc 共享避免每轮深拷贝（含完整 JSON schema，可达数十 KB）。
/// serde 对 Arc 透明序列化，线上请求格式不变。
pub tools: Option<Arc<Vec<Tool>>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -158,7 +161,7 @@ pub type StreamCallback = Arc<dyn Fn(StreamDelta) + Send + Sync>;
pub trait LLMProvider: Send + Sync {
async fn chat(
&self,
request: ChatCompletionRequest,
request: &ChatCompletionRequest,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>>;
/// 带流式回调的 chat：每收到一个 SSE delta 就调用 callback。
@ -166,7 +169,7 @@ pub trait LLMProvider: Send + Sync {
/// 默认实现忽略 callback，直接调用 chat()。
async fn chat_with_streaming(
&self,
request: ChatCompletionRequest,
request: &ChatCompletionRequest,
_callback: StreamCallback,
) -> Result<ChatCompletionResponse, Box<dyn std::error::Error + Send + Sync>> {
self.chat(request).await

146
src/scheduler/mod.rs
View File

@ -107,7 +107,7 @@ impl Scheduler {
return;
}
if let Err(error) = self.sync_config_jobs() {
if let Err(error) = self.sync_config_jobs().await {
tracing::error!(error = %error, "Failed to sync scheduler config jobs");
}
@ -137,25 +137,41 @@ impl Scheduler {
}
}
fn sync_config_jobs(&self) -> anyhow::Result<()> {
/// 同步配置中的 job 定义到 DB。整体移入 spawn_blocking：
/// 方法内是同步 SQLite 调用（含锁等待），直接在 async worker 上执行
/// 会阻塞同 worker 的其他任务。
async fn sync_config_jobs(&self) -> anyhow::Result<()> {
let now = Utc::now();
for job in self.config.effective_jobs(&crate::config::TimeConfig {
timezone: self.timezone.name().to_string(),
}) {
let runtime =
RuntimeJob::from_config(&job, now, self.config.misfire_policy, self.timezone)?;
let mut upsert = runtime.to_upsert();
if let Some(existing) = self.jobs.get_scheduler_job(&runtime.id)? {
preserve_persisted_runtime(&mut upsert, &existing);
let config = self.config.clone();
let jobs = self.jobs.clone();
let timezone = self.timezone;
let misfire_policy = config.misfire_policy;
tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
for job in config.effective_jobs(&crate::config::TimeConfig {
timezone: timezone.name().to_string(),
}) {
let runtime = RuntimeJob::from_config(&job, now, misfire_policy, timezone)?;
let mut upsert = runtime.to_upsert();
if let Some(existing) = jobs.get_scheduler_job(&runtime.id)? {
preserve_persisted_runtime(&mut upsert, &existing);
}
jobs.upsert_scheduler_job(&upsert)?;
}
self.jobs.upsert_scheduler_job(&upsert)?;
}
Ok(())
Ok(())
})
.await
.map_err(|e| anyhow::anyhow!("scheduler config sync task failed: {e}"))?
}
async fn recover_interrupted_jobs(&self) -> anyhow::Result<usize> {
let now = Utc::now();
let running_jobs = self.jobs.list_running_scheduler_jobs()?;
let jobs_repo = self.jobs.clone();
let running_jobs =
tokio::task::spawn_blocking(move || jobs_repo.list_running_scheduler_jobs())
.await
.map_err(|e| {
anyhow::anyhow!("scheduler interrupted jobs load task failed: {e}")
})??;
let mut recovered_count = 0;
for record in running_jobs {
@ -192,17 +208,19 @@ impl Scheduler {
other => other,
};
self.jobs.update_scheduler_job_runtime(
&record.id,
Self::persist_job_runtime_off_worker(
self.jobs.clone(),
record.id.clone(),
SchedulerJobState::Scheduled,
Some(SchedulerJobStatus::Error),
Some(&error_msg),
Some(error_msg),
record.run_count,
record.last_fired_at,
next_fire_at,
record.paused_at,
record.completed_at,
)?;
)
.await?;
recovered_count += 1;
tracing::info!(
@ -226,7 +244,10 @@ impl Scheduler {
async fn process_tick(&self) -> anyhow::Result<()> {
let now = Utc::now();
let jobs = self.jobs.list_scheduler_jobs(true)?;
let jobs_repo = self.jobs.clone();
let jobs = tokio::task::spawn_blocking(move || jobs_repo.list_scheduler_jobs(true))
.await
.map_err(|e| anyhow::anyhow!("scheduler jobs load task failed: {e}"))??;
for record in jobs {
let Some(job) =
@ -236,17 +257,19 @@ impl Scheduler {
};
if record.next_fire_at.is_none() && job.next_fire_at.is_some() {
self.jobs.update_scheduler_job_runtime(
&job.id,
Self::persist_job_runtime_off_worker(
self.jobs.clone(),
job.id.clone(),
job.state.clone(),
job.last_status.clone(),
job.last_error.as_deref(),
job.last_error.clone(),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
)?;
)
.await?;
}
if !job.is_due(now) {
@ -265,17 +288,19 @@ impl Scheduler {
continue;
};
self.jobs.update_scheduler_job_runtime(
&job.id,
Self::persist_job_runtime_off_worker(
self.jobs.clone(),
job.id.clone(),
SchedulerJobState::Running,
job.last_status.clone(),
job.last_error.as_deref(),
job.last_error.clone(),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
)?;
)
.await?;
// 执行与事后状态写入移入后台任务：tick 循环只做派发，
// 长耗时任务（agent_task 可能长达数分钟）不再串行阻塞其他 job 的触发。
@ -314,17 +339,20 @@ impl Scheduler {
timezone,
) {
Ok(()) => {
if let Err(error) = jobs_repo.update_scheduler_job_runtime(
&job.id,
if let Err(error) = Scheduler::persist_job_runtime_off_worker(
jobs_repo.clone(),
job.id.clone(),
job.state.clone(),
status,
job.last_error.as_deref(),
job.last_error.clone(),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
) {
)
.await
{
tracing::error!(
job_id = %job.id,
error = %error,
@ -339,17 +367,20 @@ impl Scheduler {
error = %error,
"Failed to compute post-execution scheduler state, resetting to Scheduled"
);
if let Err(update_error) = jobs_repo.update_scheduler_job_runtime(
&job.id,
if let Err(update_error) = Scheduler::persist_job_runtime_off_worker(
jobs_repo,
job.id.clone(),
SchedulerJobState::Scheduled,
Some(SchedulerJobStatus::Error),
Some(&error.to_string()),
Some(error.to_string()),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
) {
)
.await
{
tracing::error!(
job_id = %job.id,
error = %update_error,
@ -379,6 +410,39 @@ impl Scheduler {
.await
}
/// 将 job 运行时状态写入移入 spawn_blocking：
/// update_scheduler_job_runtime 是同步 SQLite 调用，DB 锁竞争时
/// 的等待时间不可控，直接 await 点外执行会阻塞 tokio async worker。
async fn persist_job_runtime_off_worker(
jobs: Arc<dyn SchedulerJobRepository>,
job_id: String,
state: SchedulerJobState,
last_status: Option<SchedulerJobStatus>,
last_error: Option<String>,
run_count: i64,
last_fired_at: Option<i64>,
next_fire_at: Option<i64>,
paused_at: Option<i64>,
completed_at: Option<i64>,
) -> anyhow::Result<()> {
tokio::task::spawn_blocking(move || {
jobs.update_scheduler_job_runtime(
&job_id,
state,
last_status,
last_error.as_deref(),
run_count,
last_fired_at,
next_fire_at,
paused_at,
completed_at,
)
})
.await
.map_err(|e| anyhow::anyhow!("scheduler job state persist task failed: {e}"))?
.map_err(anyhow::Error::from)
}
/// job 执行主体：不依赖 &self，便于移入 tokio::spawn 的后台任务。
async fn execute_job_inner(
bus: &Arc<MessageBus>,
@ -1601,8 +1665,8 @@ mod tests {
assert_eq!(saved.state, SchedulerJobState::Scheduled);
}
#[test]
fn sync_config_jobs_persists_builtin_memory_maintenance_job() {
#[tokio::test]
async fn sync_config_jobs_persists_builtin_memory_maintenance_job() {
let store = Arc::new(SessionStore::in_memory().unwrap());
let (agent_task_executor, maintenance_service) = test_scheduler_services();
@ -1615,7 +1679,7 @@ mod tests {
maintenance_service,
);
scheduler.sync_config_jobs().unwrap();
scheduler.sync_config_jobs().await.unwrap();
let saved = store
.get_scheduler_job(BUILTIN_MEMORY_MAINTENANCE_JOB_ID)
@ -1647,8 +1711,8 @@ mod tests {
assert!(saved.next_fire_at.is_some());
}
#[test]
fn sync_config_jobs_preserves_persisted_next_fire_at_for_matching_jobs() {
#[tokio::test]
async fn sync_config_jobs_preserves_persisted_next_fire_at_for_matching_jobs() {
let store = Arc::new(SessionStore::in_memory().unwrap());
let persisted_next_fire_at = 1_700_000_300_000;
let config_job = SchedulerJobConfig {
@ -1735,7 +1799,7 @@ mod tests {
maintenance_service,
);
scheduler.sync_config_jobs().unwrap();
scheduler.sync_config_jobs().await.unwrap();
let saved = store.get_scheduler_job("agent.heartbeat").unwrap().unwrap();

46
src/storage/mod.rs
View File

@ -388,6 +388,37 @@ impl SessionStore {
Ok(sessions)
}
/// 按 id 后缀查找最近活跃的一条 session（`LIMIT 1` 定向查询）。
///
/// 调用方只需要"最新一条"时使用本方法，避免 LIKE 匹配多行时
/// 全部反序列化后仅取 `[0]`。
pub fn find_first_session_by_id_suffix(
&self,
suffix: &str,
) -> Result<Option<SessionRecord>, StorageError> {
let conn = self.pool.get()?;
let pattern = format!("%{}", suffix);
conn.query_row(
"
SELECT id, title, channel_name, chat_id, summary,
created_at, updated_at, last_active_at,
archived_at, deleted_at, message_count,
user_turn_count, agent_prompt_reinjection_count
FROM sessions
WHERE id LIKE ?1 AND deleted_at IS NULL
ORDER BY last_active_at DESC
LIMIT 1
",
params![pattern],
map_session_record,
)
.map(Some)
.or_else(|e| match e {
rusqlite::Error::QueryReturnedNoRows => Ok(None),
other => Err(other.into()),
})
}
pub fn list_sessions(
&self,
channel_name: &str,
@ -1846,6 +1877,21 @@ impl SessionStore {
Ok(count as usize)
}
/// 按 topic 统计 user 角色消息数（定向 `COUNT(*)`）。
///
/// 与 `get_topic_message_count` 同理：数据库侧计数，
/// 避免为得到数量而全量加载消息体（content、tool_calls_json 等大字段）。
/// 过滤条件与 `load_messages_for_topic(_, None)` 一致（排除已压缩消息）。
pub fn count_user_messages_for_topic(&self, topic_id: &str) -> Result<usize, StorageError> {
let conn = self.pool.get()?;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM messages WHERE topic_id = ?1 AND role = 'user' AND is_compacted = 0",
params![topic_id],
|row| row.get(0),
)?;
Ok(count as usize)
}
pub fn load_all_messages(&self, session_id: &str) -> Result<Vec<ChatMessage>, StorageError> {
let conn = self.pool.get()?;
load_messages_after(&conn, session_id, 0)

46
src/tools/memory_search.rs
View File

@ -90,9 +90,15 @@ impl Tool for MemorySearchTool {
let payload = match action {
"list" => {
let limit = extract_u64(&args, "limit").unwrap_or(10) as usize;
let memories = self
.memories
.list_memories("user", &scope_key, namespace, limit)?;
// 同步 SQLite 查询移入 spawn_blocking，避免阻塞 tokio worker
let memories_repo = self.memories.clone();
let scope = scope_key.clone();
let ns = namespace.map(str::to_string);
let memories = tokio::task::spawn_blocking(move || {
memories_repo.list_memories("user", &scope, ns.as_deref(), limit)
})
.await
.map_err(|e| anyhow::anyhow!("memory list task failed: {e}"))??;
json!({
"count": memories.len(),
"memories": memories.into_iter().map(memory_to_json).collect::<Vec<_>>()
@ -138,9 +144,22 @@ impl Tool for MemorySearchTool {
return Ok(error_result("Missing required parameter: queries"));
}
let limit = extract_u64(&args, "limit").unwrap_or(10) as usize;
let memories = self
.memories
.search_memories_any("user", &scope_key, &queries, namespace, limit)?;
let memories_repo = self.memories.clone();
let scope = scope_key.clone();
let ns = namespace.map(str::to_string);
let query_terms = queries.clone();
// 同步 SQLite 多关键词查询（LIKE 扫描）移入 spawn_blocking
let memories = tokio::task::spawn_blocking(move || {
memories_repo.search_memories_any(
"user",
&scope,
&query_terms,
ns.as_deref(),
limit,
)
})
.await
.map_err(|e| anyhow::anyhow!("memory search task failed: {e}"))??;
json!({
"queries": queries,
"count": memories.len(),
@ -157,10 +176,17 @@ impl Tool for MemorySearchTool {
None => return Ok(error_result("Missing required parameter: key")),
};
match self
.memories
.get_memory("user", &scope_key, namespace, key)?
{
let memories_repo = self.memories.clone();
let scope = scope_key.clone();
let ns = namespace.to_string();
let memory_key = key.to_string();
// 同步 SQLite 查询移入 spawn_blocking
let memory = tokio::task::spawn_blocking(move || {
memories_repo.get_memory("user", &scope, &ns, &memory_key)
})
.await
.map_err(|e| anyhow::anyhow!("memory get task failed: {e}"))??;
match memory {
Some(memory) => memory_to_json(memory),
None => {
return Ok(error_result(&format!(

2
src/topic_description.rs
View File

@ -15,7 +15,7 @@ pub async fn generate_topic_description(
tools: None,
};
let response = provider.chat(request).await?;
let response = provider.chat(&request).await?;
let description = response.content.trim().to_string();
if description.is_empty() {

4
tests/test_integration.rs
View File

@ -65,7 +65,7 @@ async fn test_openai_simple_completion() {
let config = load_config().expect("Please configure tests/test.env with valid API keys");
let provider = create_provider(to_runtime_config(config)).expect("Failed to create provider");
let response = provider.chat(create_request("Say 'ok'")).await.unwrap();
let response = provider.chat(&create_request("Say 'ok'")).await.unwrap();
assert!(!response.id.is_empty());
assert!(!response.content.is_empty());
@ -91,7 +91,7 @@ async fn test_openai_conversation() {
tools: None,
};
let response = provider.chat(request).await.unwrap();
let response = provider.chat(&request).await.unwrap();
assert!(response.content.to_lowercase().contains("alice"));
}

14
tests/test_tool_calling.rs
View File

@ -83,10 +83,10 @@ async fn test_openai_tool_call() {
messages: vec![Message::user("What is the weather in Tokyo?")],
temperature: Some(0.0),
max_tokens: Some(200),
tools: Some(vec![make_weather_tool()]),
tools: Some(std::sync::Arc::new(vec![make_weather_tool()])),
};
let response = provider.chat(request).await.unwrap();
let response = provider.chat(&request).await.unwrap();
// Should have tool calls
assert!(
@ -112,10 +112,10 @@ async fn test_openai_tool_call_with_manual_execution() {
messages: vec![Message::user("What is the weather in Tokyo?")],
temperature: Some(0.0),
max_tokens: Some(200),
tools: Some(vec![make_weather_tool()]),
tools: Some(std::sync::Arc::new(vec![make_weather_tool()])),
};
let response1 = provider.chat(request1).await.unwrap();
let response1 = provider.chat(&request1).await.unwrap();
let tool_call = response1.tool_calls.first().expect("Expected tool call");
assert_eq!(tool_call.name, "get_weather");
@ -127,10 +127,10 @@ async fn test_openai_tool_call_with_manual_execution() {
],
temperature: Some(0.0),
max_tokens: Some(200),
tools: Some(vec![make_weather_tool()]),
tools: Some(std::sync::Arc::new(vec![make_weather_tool()])),
};
let response2 = provider.chat(request2).await.unwrap();
let response2 = provider.chat(&request2).await.unwrap();
// Should have a response
assert!(!response2.content.is_empty() || !response2.tool_calls.is_empty());
@ -150,7 +150,7 @@ async fn test_openai_no_tool_when_not_provided() {
tools: None,
};
let response = provider.chat(request).await.unwrap();
let response = provider.chat(&request).await.unwrap();
// Should NOT have tool calls
assert!(response.tool_calls.is_empty());

35
web/src/App.tsx
View File

@ -645,6 +645,11 @@ function App() {
[selectedSessionId, selectSession],
);
// merged_tool 对象复用缓存：流式期间 messages 每帧换引用，但绝大多数
// tool_call/tool_result 输入未变；按值相等复用上次对象，使 MessageBubble
// 的 memo 浅比较命中，避免所有工具气泡每帧重渲染（含 ReactMarkdown 重解析）。
const mergedToolCacheRef = useRef(new Map<string, ChatMessage>());
const chatMessages = useMemo(() => {
const result: ChatMessage[] = [];
const toolCallIndex = new Map<string, number>();
@ -683,6 +688,28 @@ function App() {
}
}
// 按决定性字段值相等复用上次的 merged_tool 对象（渲染结果相同 → 引用可复用）
const cache = mergedToolCacheRef.current;
const nextCache = new Map<string, ChatMessage>();
for (let i = 0; i < result.length; i++) {
const cur = result[i];
if (cur.type !== 'merged_tool') continue;
const key = cur.toolCallId || cur.id;
const prev = cache.get(key);
if (
prev &&
prev.id === cur.id &&
prev.status === cur.status &&
prev.callContent === cur.callContent &&
prev.resultContent === cur.resultContent &&
prev.durationMs === cur.durationMs
) {
result[i] = prev;
}
nextCache.set(key, result[i]);
}
mergedToolCacheRef.current = nextCache;
// 过滤无实质内容的 merged_tool：result 到达后才显示保留；calling/pending 有 callContent 也保留
return result.filter((msg) => {
if (msg.type !== 'merged_tool') return true;
@ -691,12 +718,14 @@ function App() {
});
}, [messages]);
// 视图标识：用于 MessageList 保存/恢复每个视图的滚动位置
// 视图标识：用于 MessageList 保存/恢复每个视图的滚动位置。
// 依赖 subAgentTaskId（原始值）而非 subAgentView 对象，避免子代理流式
// 期间每帧无效重算（对象每帧换引用但 taskId 不变）。
const viewKey = useMemo(() => {
if (schedulerView) return `scheduler:${schedulerView.jobId}`;
if (subAgentView) return `subagent:${subAgentView.taskId}`;
if (subAgentTaskId) return `subagent:${subAgentTaskId}`;
return `main:${selectedTopic ?? ''}`;
}, [schedulerView, subAgentView, selectedTopic]);
}, [schedulerView, subAgentTaskId, selectedTopic]);
return (
<div className="flex h-screen overflow-hidden bg-[var(--bg-primary)] text-[var(--text-primary)]">

8
web/src/components/Chat/ChatContainer.tsx
View File

@ -1,4 +1,4 @@
import { useState, useCallback } from 'react';
import { useState, useCallback, memo } from 'react';
import { MessageList } from './MessageList';
import { MessageInput } from './MessageInput';
import { ExpertSelector } from './ExpertSelector';
@ -29,7 +29,9 @@ interface ChatContainerProps {
topicId?: string | null;
}
export function ChatContainer({
// memo：props 除 messages 外全部稳定（useCallback/原始值），
// App 因非消息类 state（侧栏折叠、主题等）重渲染时跳过整个聊天子树。
export const ChatContainer = memo(function ChatContainer({
messages,
isLoading,
isReadOnly = false,
@ -136,4 +138,4 @@ export function ChatContainer({
</div>
</div>
);
}
});

7
web/src/components/Chat/ExpertSelector.tsx
View File

@ -1,4 +1,4 @@
import { useState, useEffect, useRef, useCallback } from 'react';
import { useState, useEffect, useRef, useCallback, memo } from 'react';
import { UserCheck, ChevronDown, Loader2, Settings, Check } from 'lucide-react';
import { getSelectedExpert, selectExpert, listExperts } from '../../api/experts';
@ -24,7 +24,8 @@ interface ExpertSelectorProps {
settingsClosedTick?: number;
}
export function ExpertSelector({
// memo：props 全部稳定（useCallback/原始值），流式期间跳过重渲染
export const ExpertSelector = memo(function ExpertSelector({
sessionId,
onManageExperts,
onSelectionChange,
@ -278,4 +279,4 @@ export function ExpertSelector({
{error && <span className="text-xs text-[rgb(242,90,90)] truncate">{error}</span>}
</div>
);
}
});

8
web/src/components/Chat/MessageInput.tsx
View File

@ -10,7 +10,7 @@ import {
MusicIcon,
VideoIcon,
} from 'lucide-react';
import { useState, useRef, useEffect } from 'react';
import { useState, useRef, useEffect, memo } from 'react';
import type { Attachment } from '../../types/protocol';
const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50MB
@ -77,7 +77,9 @@ function getMediaType(mimeType: string): string {
return 'file';
}
export function MessageInput({
// memo：props 全部稳定（回调 useCallback、对象 prop 来自父组件 state），
// 流式期间父组件每帧重渲染时完全跳过输入区子树。
export const MessageInput = memo(function MessageInput({
onSend,
onStop,
disabled = false,
@ -528,4 +530,4 @@ export function MessageInput({
</div>
</div>
);
}
});

8
web/src/components/Chat/MessageList.tsx
View File

@ -1,4 +1,4 @@
import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo } from 'react';
import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo, memo } from 'react';
import { useVirtualizer } from '@tanstack/react-virtual';
import { MessageBubble } from './MessageBubble';
import type { ChatMessage } from '../../types/protocol';
@ -16,7 +16,9 @@ interface MessageListProps {
effectiveModel?: { provider: string; model: string } | null;
}
export function MessageList({
// memo：props 与 ChatContainer 同源（messages 变化时才需重渲染），
// 阻断父组件非消息类重渲染向虚拟化列表的传导。
export const MessageList = memo(function MessageList({
messages,
onNavigateToSubAgent,
showThinking = true,
@ -385,4 +387,4 @@ export function MessageList({
)}
</div>
);
}
});

7
web/src/components/Chat/ModelSelector.tsx
View File

@ -1,4 +1,4 @@
import { useState, useEffect, useRef, useCallback } from 'react';
import { useState, useEffect, useRef, useCallback, memo } from 'react';
import { Cpu, ChevronDown, Loader2, Check } from 'lucide-react';
import {
listModelOptions,
@ -19,7 +19,8 @@ interface ModelSelectorProps {
onSelectionChange?: (effective: { provider: string; model: string; overridden: boolean }) => void;
}
export function ModelSelector({
// memo：props 全部稳定（useCallback/原始值），流式期间跳过重渲染
export const ModelSelector = memo(function ModelSelector({
sessionId,
topicId,
settingsClosedTick,
@ -284,4 +285,4 @@ export function ModelSelector({
{error && <span className="text-xs text-[rgb(242,90,90)] truncate">{error}</span>}
</div>
);
}
});

42
web/src/components/Panel/MemoryPanel.tsx
View File

@ -1,4 +1,4 @@
import { useState } from 'react';
import { useState, useMemo, memo } from 'react';
import {
Brain,
User,
@ -310,7 +310,9 @@ function SectionHeader({
/* ── main component ────────────────────────────────────── */
export function MemoryPanel({
// memo：props 全部稳定（memories 仅在刷新时换引用、回调均 useCallback），
// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组/排序重算。
export const MemoryPanel = memo(function MemoryPanel({
memories,
onRefresh,
onClose,
@ -342,22 +344,26 @@ export function MemoryPanel({
});
};
const grouped = new Map<string, MemorySummary[]>();
for (const m of memories) {
const l = grouped.get(m.namespace) || [];
l.push(m);
grouped.set(m.namespace, l);
}
// memories 引用未变时跳过分组/排序重算（流式期间 App 每帧重渲染）
const { grouped, sorted } = useMemo(() => {
const grouped = new Map<string, MemorySummary[]>();
for (const m of memories) {
const l = grouped.get(m.namespace) || [];
l.push(m);
grouped.set(m.namespace, l);
}
const order = ['user', 'semantic', 'episodic', 'skill', 'environment', 'reflection', 'other'];
const sorted = Array.from(grouped.keys()).sort((a, b) => {
const ai = order.indexOf(a);
const bi = order.indexOf(b);
if (ai !== -1 && bi !== -1) return ai - bi;
if (ai !== -1) return -1;
if (bi !== -1) return 1;
return a.localeCompare(b);
});
const order = ['user', 'semantic', 'episodic', 'skill', 'environment', 'reflection', 'other'];
const sorted = Array.from(grouped.keys()).sort((a, b) => {
const ai = order.indexOf(a);
const bi = order.indexOf(b);
if (ai !== -1 && bi !== -1) return ai - bi;
if (ai !== -1) return -1;
if (bi !== -1) return 1;
return a.localeCompare(b);
});
return { grouped, sorted };
}, [memories]);
const handleCreate = (ns: string, key: string, content: string) => {
sendCommand(onCreateMemory(ns, key, content));
@ -469,4 +475,4 @@ export function MemoryPanel({
)}
</div>
);
}
});

49
web/src/components/Panel/SkillList.tsx
View File

@ -1,4 +1,4 @@
import { useState } from 'react';
import { useState, useMemo, memo } from 'react';
import {
Package,
User,
@ -100,7 +100,9 @@ function SkillCard({ skill, config }: { skill: SkillSummary; config: SourceConfi
/* ── main component ────────────────────────────────────── */
export function SkillList({ skills, onRefresh }: SkillListProps) {
// memo：props 全部稳定（skills 仅在刷新时换引用、onRefresh 为 useCallback），
// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组/排序重算。
export const SkillList = memo(function SkillList({ skills, onRefresh }: SkillListProps) {
const [collapsed, setCollapsed] = useState<Set<string>>(() => {
try {
const s = localStorage.getItem('picobot-skill-collapsed');
@ -123,22 +125,33 @@ export function SkillList({ skills, onRefresh }: SkillListProps) {
});
};
const grouped = new Map<string, SkillSummary[]>();
for (const s of skills) {
const l = grouped.get(s.source) || [];
l.push(s);
grouped.set(s.source, l);
}
// skills 引用未变时跳过分组/排序重算（流式期间 App 每帧重渲染）
const { grouped, sorted } = useMemo(() => {
const grouped = new Map<string, SkillSummary[]>();
for (const s of skills) {
const l = grouped.get(s.source) || [];
l.push(s);
grouped.set(s.source, l);
}
const order = ['user', 'useragent', 'useropenclaw', 'project', 'projectagent', 'projectopenclaw'];
const sorted = Array.from(grouped.keys()).sort((a, b) => {
const ai = order.indexOf(a);
const bi = order.indexOf(b);
if (ai !== -1 && bi !== -1) return ai - bi;
if (ai !== -1) return -1;
if (bi !== -1) return 1;
return a.localeCompare(b);
});
const order = [
'user',
'useragent',
'useropenclaw',
'project',
'projectagent',
'projectopenclaw',
];
const sorted = Array.from(grouped.keys()).sort((a, b) => {
const ai = order.indexOf(a);
const bi = order.indexOf(b);
if (ai !== -1 && bi !== -1) return ai - bi;
if (ai !== -1) return -1;
if (bi !== -1) return 1;
return a.localeCompare(b);
});
return { grouped, sorted };
}, [skills]);
return (
<div className="flex h-full flex-col">
@ -209,4 +222,4 @@ export function SkillList({ skills, onRefresh }: SkillListProps) {
)}
</div>
);
}
});

16
web/src/components/Panel/TodoPanel.tsx
View File

@ -1,4 +1,4 @@
import { useState, useCallback, useEffect, useRef } from 'react';
import { useState, useCallback, useEffect, useRef, useMemo, memo } from 'react';
import { ClipboardList, ChevronDown, RefreshCw } from 'lucide-react';
import type { TodoItemSummary, Command } from '../../types/protocol';
@ -53,7 +53,14 @@ function PulseDot() {
/* ── TodoPanel ────────────────────────────────────────── */
export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }: TodoPanelProps) {
// memo：props 全部稳定（todos 仅在 todo_list 消息到达时换引用、回调均 useCallback），
// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组重算。
export const TodoPanel = memo(function TodoPanel({
todos,
requestTodoList,
sendCommand,
onTodoClick,
}: TodoPanelProps) {
const [collapsedGroups, setCollapsedGroups] = useState<Set<string>>(
() => new Set(['completed', 'cancelled']),
);
@ -75,7 +82,8 @@ export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }:
prevTodoIdsRef.current = newIds;
}, [todos]);
const grouped = groupTodos(todos);
// todos 引用未变时跳过分组重算（流式期间 App 每帧重渲染）
const grouped = useMemo(() => groupTodos(todos), [todos]);
const inProgressCount = grouped.get('in_progress')?.length ?? 0;
const totalCount = todos.length;
@ -180,4 +188,4 @@ export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }:
</div>
</div>
);
}
});

9
web/src/components/Panel/TopicTokenStatsPanel.tsx
View File

@ -1,3 +1,4 @@
import { memo } from 'react';
import { Coins } from 'lucide-react';
import type { TopicTokenStats } from '../../types/protocol';
import {
@ -12,7 +13,11 @@ interface TopicTokenStatsPanelProps {
tokenStats?: TopicTokenStats | null;
}
export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps) {
// memo：tokenStats 引用稳定（仅 topic 刷新时换引用），
// 流式期间 App 每帧重渲染时跳过面板重渲染。
export const TopicTokenStatsPanel = memo(function TopicTokenStatsPanel({
tokenStats,
}: TopicTokenStatsPanelProps) {
// 无数据态
if (!tokenStats || tokenStats.total_tokens === 0) {
return (
@ -126,4 +131,4 @@ export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps)
)}
</div>
);
}
});

12
web/src/components/Sidebar/SchedulerJobList.tsx
View File

@ -1,3 +1,4 @@
import { memo } from 'react';
import { Clock, RefreshCw, ChevronRight, Check, X, Minus } from 'lucide-react';
import type { SchedulerJobSummary, SchedulerJobSessionLookup } from '../../types/protocol';
@ -92,7 +93,14 @@ function lastStatusIcon(lastStatus: string | undefined) {
}
}
export function SchedulerJobList({ jobs, onRefresh, onViewJob, sessionId }: SchedulerJobListProps) {
// memo：props 全部稳定（jobs 仅在刷新时换引用、回调均 useCallback），
// 阻断 App 重渲染向定时任务列表的无效传导。
export const SchedulerJobList = memo(function SchedulerJobList({
jobs,
onRefresh,
onViewJob,
sessionId,
}: SchedulerJobListProps) {
return (
<div className="flex h-full flex-col">
{/* Header */}
@ -228,4 +236,4 @@ export function SchedulerJobList({ jobs, onRefresh, onViewJob, sessionId }: Sche
</div>
</div>
);
}
});

8
web/src/components/Sidebar/TopicList.tsx
View File

@ -1,4 +1,4 @@
import { useState, useEffect, useMemo, useRef, useCallback } from 'react';
import { useState, useEffect, useMemo, useRef, useCallback, memo } from 'react';
import {
Plus,
MessageSquare,
@ -43,7 +43,9 @@ function formatTime(timestamp: number): string {
}
}
export function TopicList({
// memo：props 全部稳定（topics 仅在 topic_list 消息到达时换引用、回调均 useCallback），
// 流式期间 App 每帧重渲染时跳过话题列表重渲染与分页重算。
export const TopicList = memo(function TopicList({
sessionId,
topics,
currentTopicId,
@ -357,4 +359,4 @@ export function TopicList({
)}
</div>
);
}
});