diff --git a/Cargo.lock b/Cargo.lock index 7dfddb9..253fb7e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1635,7 +1635,7 @@ dependencies = [ [[package]] name = "picobot" -version = "0.2.0" +version = "0.3.0" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 2a5699d..c9a9901 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "picobot" -version = "0.2.0" +version = "0.3.0" edition = "2024" [lints.rust] diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 661a389..094b53e 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -2,6 +2,70 @@ 本文件记录 Picobot 各版本的显著变更,遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/) 风格。 +## [0.3.0] - 2026-08-04 + +较 [0.2.0] 的 14 个已提交 commit + 本次版本一并提交的锁屏冻结架构修复(12 文件),聚焦 **Agent 执行与显示层解耦**、**工程化基线**、**并发持久化稳定性** 与 **安全加固** 四大方向。架构修复经五轮对抗性审查验证。 + +### 新增功能 + +#### 话题重命名 +- 新增 `RenameTopic` 命令与 `TopicRenamed` 协议消息,复用存储层已有的 `update_topic_title` 方法。后端响应携带刷新后的完整 topic 列表,前端零额外往返同步侧边栏。 +- 前端 `TopicList` 侧边栏增加内联编辑入口:悬停显示铅笔图标,Enter 提交 / Esc 取消 / blur 取消,通过 `onMouseDown preventDefault` 防止按钮点击时 input 提前失焦。 + +#### 前端自动构建集成 +- 新增 `build.rs`,`cargo build` 时自动执行 `npm install + npm run build`,消除前后端构建割裂。支持 `SKIP_FRONTEND_BUILD` 环境变量跳过。 + +#### 工程化基线 +- `rustfmt.toml` 固化 `max_width=100` / 4 空格缩进;`Cargo.toml` 配置 `[lints.rust]` 与 `[lints.clippy]` 渐进式规则。 +- `.github/workflows/ci.yml`:Rust(fmt + clippy + test)+ 前端(eslint + tsc + test + prettier format:check)双平台 CI。 +- `Makefile` 新增 `check` / `fmt` / `fix` 目标,clippy 对齐 `--all-targets --all-features`。 +- 前端 eslint flat config + prettier 配置,对 47 个前端文件统一格式化并在 CI 强制 `format:check`。 + +#### 默认迭代上限调整 +- `max_tool_iterations` 默认值调整为 1000,匹配长任务子代理的实际需求。 + +### 架构修复 + +#### Agent 执行与显示层解耦(锁屏冻结根因修复) +根因:浏览器锁屏导致 WebSocket 半死,`ws_sender.send().await` 永久阻塞,级联阻塞 dispatcher → MessageBus → Agent Loop,后端停止执行直到解锁。 + +基于第一性原理建立"执行-显示解耦"原则:agent 执行只依赖 SQLite 持久化,实时广播是可丢弃的最佳努力通道。 + +- `MessageBus::publish_outbound` 由 `send().await` 改为 `try_send()`:bus 满时丢弃消息并告警,agent 不再被显示层阻塞。新增 `BusError::Dropped` 变体。 +- WebSocket writer task 包裹 `tokio::time::timeout(30s)`:半死连接超时即关闭。使用每连接独立的 `CancellationToken`(非共享的 CliChannel 级 token),避免一个连接超时关闭所有连接。writer 退出时 cancel 通知主 loop 退出,确保 `unregister_connection` 执行。 +- `CliChannel::send` 由 `send().await` 改为 `try_send()`:dispatcher 单线程顺序处理,原阻塞式发送在 writer 卡住时会阻塞所有连接 37s,改后立即返回。 +- 全仓 13 处 `publish_outbound` 调用统一区分 `Dropped`(warn,预期背压)/ `Closed`(error,异常)日志级别,消除监控噪音。scheduler 的 3 处 `?` 改为 warn,避免 `Dropped` 触发 misfire 重试风暴。 +- 前端 WebSocket 添加 25s 客户端 ping + 指数退避重连(3s→6s→12s→24s→60s 封顶,上限 999 次)。重连后区分"重连恢复"与"首次/切换通道":前者保留断连前 messages 并刷新 topic 列表,后者清空数据避免污染。 + +### 安全加固 + +#### SSRF 重定向绕过修复 +- `web_fetch` 工具禁用 HTTP 重定向(`redirect::Policy::none()`):原 `validate_url` 只校验初始 URL 的 host,跟随 302 跳转可重定向到 `169.254.169.254`(云元数据)或 `127.0.0.1` 等内网地址,绕过 `is_private_host` 的 SSRF 防护。与 `http_request` 工具保持一致。 + +#### Safety Guard 正则收紧 +- `format` 正则收紧为 `\bformat\s+.*[a-z]:`,要求出现盘符才拦截,避免误伤 `dart format`、`buf format`、`pytest --format` 等合法命令。 +- 按平台分组注入规则,Unix 不再注入 Windows 专用规则;`Remove-Item` 正则改为小写,与 `guard_command` 的大小写处理一致。 + +### 稳定性修复 + +#### SQLite 并发写入死锁 +- 7 个写事务从 `BEGIN DEFERRED` 改为 `BEGIN IMMEDIATE`:在事务开始即获取写锁,消除多 sub-agent 并发写入时的死锁路径。`busy_timeout` 从 5s 提升至 30s。 +- 根因:`BEGIN DEFERRED` 下多个事务可同时读 `MAX(seq)` 不持写锁,提交时互相阻塞,5s timeout 耗尽后返回 `SQLITE_BUSY`。`BEGIN IMMEDIATE` 强制写者串行排队,顺带消除 `MAX(seq)+1` 竞态导致的 UNIQUE 约束冲突。 + +#### 测试与代码质量 +- 修复 `anthropic` 错误链嵌套测试为真正的 `#[source]` 链(原测试是无效的,inner 变量被 `let _` 抑制)。 +- 补充 `anthropic` provider 15 个纯函数单测(原 396 行零测试),覆盖 data URL 解析、图片过滤、字段过滤、响应反序列化、错误链格式化。 +- 修复 3 个失败测试:`agent_md_template` 写入模板内容 / `StreamingAccumulator` BTreeMap 保序 / `source_order` 断言补全。 +- 清理未使用的 dead code 函数与方法。 +- 新增 `ARCHITECTURE.md` 架构文档,聚焦数据流与 7 个关键设计决策。 + +#### CI 覆盖范围扩大 +- `cargo build --lib` 改为 `cargo build`,确保 `main.rs` 二进制入口被编译验证。 +- `cargo test --lib` 改为 `cargo test`,纳入 `tests/` 目录集成测试。`test_request_format.rs` 的 8 个序列化测试此前从未在 CI 中运行。 + +#### 跨平台构建修复 +- 修正依赖分类错误:`rmcp` / `schemars` / `http` / `tower-http` / `rust-embed` 均为跨平台 crate,错放在 `[target.'cfg(windows)'.dependencies]` 下导致 Linux 构建失败,移出 target 段。 + ## [0.2.0] - 2026-07-31 较 [0.1.2] 的 47 个 commit 迭代,聚焦 **能力策略**、**模型独立配置**、**话题级并发隔离** 与 **Agent Loop 性能优化** 四大方向。 diff --git a/src/bus/mod.rs b/src/bus/mod.rs index de0d507..ef7e737 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -52,14 +52,26 @@ impl MessageBus { Some(msg) } - /// Publish a message to the outbound queue + /// Publish a message to the outbound queue. + /// + /// Uses `try_send` (non-blocking): if the queue is full, the message is + /// dropped immediately with a warning. This ensures the agent loop is never + /// blocked by slow or disconnected display consumers. Persistent state is + /// unaffected — messages are stored in SQLite independently. pub async fn publish_outbound(&self, msg: OutboundMessage) -> Result<(), BusError> { #[cfg(debug_assertions)] tracing::debug!(channel = %msg.channel, chat_id = %msg.chat_id, content_len = %msg.content.len(), "Bus: publishing outbound message"); - self.outbound_tx - .send(msg) - .await - .map_err(|_| BusError::Closed) + match self.outbound_tx.try_send(msg) { + Ok(()) => Ok(()), + Err(tokio::sync::mpsc::error::TrySendError::Full(msg)) => { + tracing::warn!( + channel = %msg.channel, + "Outbound bus full, dropping message" + ); + Err(BusError::Dropped) + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => Err(BusError::Closed), + } } /// Consume an outbound message from the outbound queue. @@ -76,12 +88,14 @@ impl MessageBus { #[derive(Debug)] pub enum BusError { Closed, + Dropped, } impl std::fmt::Display for BusError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { BusError::Closed => write!(f, "Bus channel closed"), + BusError::Dropped => write!(f, "Bus full, message dropped"), } } } diff --git a/src/channels/cli.rs b/src/channels/cli.rs index b103f41..1c18ede 100644 --- a/src/channels/cli.rs +++ b/src/channels/cli.rs @@ -100,12 +100,15 @@ impl Channel for CliChannel { ))); }; + // 使用 try_send 避免阻塞 dispatcher——dispatcher 是单线程顺序处理, + // 若 writer task 卡在 ws_sender.send() 上,send().await 会阻塞, + // 导致所有连接的实时消息被阻塞。try_send 满时立即返回错误, + // dispatcher 记录后继续处理下一条消息。 for outbound in ws_outbound_from_outbound_message(&msg) { connection .sender - .send(outbound) - .await - .map_err(|_| ChannelError::SendError("CLI websocket sender closed".to_string()))?; + .try_send(outbound) + .map_err(|_| ChannelError::SendError("CLI websocket sender closed or full".to_string()))?; } Ok(()) diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 53d8ec2..1156e20 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -229,7 +229,14 @@ impl InboundProcessor { )) .await { - tracing::error!(error = %error, "Failed to publish command response"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, "Failed to publish command response"); + } + } } } } else if let Some(error) = response.error { @@ -245,7 +252,14 @@ impl InboundProcessor { )) .await { - tracing::error!(error = %e, "Failed to publish error response"); + match e { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %e, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %e, "Failed to publish error response"); + } + } } } return Ok(()); @@ -305,7 +319,14 @@ impl InboundProcessor { .insert("topic_id".to_string(), topic_id.clone()); } if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!(error = %error, "Failed to publish outbound"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, "Failed to publish outbound"); + } + } } } @@ -407,7 +428,14 @@ impl InboundProcessor { )) .await { - tracing::error!(error = %publish_error, "Failed to publish execution error outbound"); + match publish_error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %publish_error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %publish_error, "Failed to publish execution error outbound"); + } + } } } } @@ -433,7 +461,14 @@ impl InboundProcessor { )) .await { - tracing::error!(error = %error, "Failed to publish execution_completed"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, "Failed to publish execution_completed"); + } + } } Ok(()) diff --git a/src/gateway/session.rs b/src/gateway/session.rs index 1dfd678..d442186 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -94,7 +94,14 @@ impl EmittedMessageHandler for BusToolCallEmitter { &message, ) { if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live outbound tool call"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live outbound tool call"); + } + } } } } @@ -113,7 +120,14 @@ impl EmittedMessageHandler for BusToolCallEmitter { &message, ) { if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live outbound tool call"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live outbound tool call"); + } + } } } @@ -154,7 +168,14 @@ impl EmittedMessageHandler for BusToolCallEmitter { }; if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!(error = %error, channel = %self.channel_name, "Failed to publish stream delta"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, "Failed to publish stream delta"); + } + } } } diff --git a/src/gateway/session_message_sender.rs b/src/gateway/session_message_sender.rs index 5bcaf2c..75640ee 100644 --- a/src/gateway/session_message_sender.rs +++ b/src/gateway/session_message_sender.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use async_trait::async_trait; -use crate::bus::{MessageBus, OutboundMessage}; +use crate::bus::{BusError, MessageBus, OutboundMessage}; use crate::tools::{SessionMessageSender, SessionSendOutcome, SessionSendRequest, ToolContext}; pub(crate) struct BusSessionMessageSender { @@ -55,15 +55,28 @@ impl SessionMessageSender for BusSessionMessageSender { if attachment_count > 0 { outbound.media = request.attachments.clone(); } - self.bus.publish_outbound(outbound).await?; - published_messages += 1; - tracing::info!( - channel = %channel_name, - chat_id = %chat_id, - content_len = content_len, - attachment_count = attachment_count, - "Published session text message to outbound bus" - ); + match self.bus.publish_outbound(outbound).await { + Ok(()) => { + published_messages += 1; + tracing::info!( + channel = %channel_name, + chat_id = %chat_id, + content_len = content_len, + attachment_count = attachment_count, + "Published session text message to outbound bus" + ); + } + Err(BusError::Dropped) => { + tracing::warn!( + channel = %channel_name, + chat_id = %chat_id, + "Outbound bus full, dropping session text message" + ); + } + Err(BusError::Closed) => { + return Err(anyhow::anyhow!("Outbound bus closed")); + } + } } else { for attachment in request.attachments { let media_path = attachment.path.clone(); @@ -77,15 +90,28 @@ impl SessionMessageSender for BusSessionMessageSender { metadata.clone(), ); outbound.media = vec![attachment]; - self.bus.publish_outbound(outbound).await?; - published_messages += 1; - tracing::info!( - channel = %channel_name, - chat_id = %chat_id, - media_type = %media_type, - media_path = %media_path, - "Published session attachment to outbound bus" - ); + match self.bus.publish_outbound(outbound).await { + Ok(()) => { + published_messages += 1; + tracing::info!( + channel = %channel_name, + chat_id = %chat_id, + media_type = %media_type, + media_path = %media_path, + "Published session attachment to outbound bus" + ); + } + Err(BusError::Dropped) => { + tracing::warn!( + channel = %channel_name, + chat_id = %chat_id, + "Outbound bus full, dropping session attachment" + ); + } + Err(BusError::Closed) => { + return Err(anyhow::anyhow!("Outbound bus closed")); + } + } } } diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index a5ddcd3..960040d 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -40,6 +40,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; const WS_CHANNEL_NAME: &str = "websocket"; @@ -221,13 +222,34 @@ async fn handle_socket(ws: WebSocket, state: Arc) { let mut receiver = receiver; let session_id_for_sender = runtime_session_id.clone(); + // 每连接独立的关闭信号:writer 超时/错误时 cancel,通知主 loop 退出 + // 不能用共享的 shutdown_token——那是 CliChannel 级别的,cancel 会关闭所有连接 + let writer_closed = CancellationToken::new(); + let writer_closed_clone = writer_closed.clone(); tokio::spawn(async move { while let Some(msg) = receiver.recv().await { if let Ok(text) = serialize_outbound(&msg) { - if ws_sender.send(WsMessage::Text(text.into())).await.is_err() { - #[cfg(debug_assertions)] - tracing::debug!(session_id = %session_id_for_sender, "WebSocket send error"); - break; + let send_result = tokio::time::timeout( + std::time::Duration::from_secs(30), + ws_sender.send(WsMessage::Text(text.into())), + ) + .await; + match send_result { + Ok(Ok(())) => {} + Ok(Err(_)) => { + #[cfg(debug_assertions)] + tracing::debug!(session_id = %session_id_for_sender, "WebSocket send error"); + writer_closed_clone.cancel(); + break; + } + Err(_) => { + tracing::warn!( + session_id = %session_id_for_sender, + "WebSocket send timed out after 30s, closing connection" + ); + writer_closed_clone.cancel(); + break; + } } } } @@ -235,11 +257,16 @@ async fn handle_socket(ws: WebSocket, state: Arc) { loop { tokio::select! { - // 监听 shutdown 信号(来自 CliChannel::stop()) + // 监听全局 shutdown 信号(来自 CliChannel::stop()) _ = shutdown_token.cancelled() => { tracing::info!(session_id = %current_session_id, "WebSocket shutdown signal received, closing connection"); break; } + // 监听 writer 退出信号(writer 超时或错误,仅关闭当前连接) + _ = writer_closed.cancelled() => { + tracing::info!(session_id = %current_session_id, "WebSocket writer closed, shutting down connection"); + break; + } // 监听 WebSocket 消息 msg = ws_receiver.next() => { let Some(msg) = msg else { diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index 5c46f03..f8255aa 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -298,7 +298,11 @@ impl Scheduler { match job.kind { SchedulerJobKind::OutboundMessage => { let message = build_outbound_message(job)?; - self.bus.publish_outbound(message).await?; + // publish_outbound 失败(bus 满或关闭)不视为 job 失败: + // 通知丢弃是预期的背压行为,标记 job 失败会触发 misfire 重试风暴 + if let Err(e) = self.bus.publish_outbound(message).await { + tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler job"); + } } SchedulerJobKind::InternalEvent => { execute_internal_event(self.maintenance_executor.as_ref(), job).await?; @@ -311,7 +315,9 @@ impl Scheduler { ) .await?; for message in outbound_messages { - self.bus.publish_outbound(message).await?; + if let Err(e) = self.bus.publish_outbound(message).await { + tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler agent task"); + } } } SchedulerJobKind::SilentAgentTask => { @@ -407,7 +413,8 @@ impl Scheduler { "silent_agent_task".to_string(), ); - self.bus + if let Err(e) = self + .bus .publish_outbound(OutboundMessage::error_notification( channel, chat_id, @@ -421,7 +428,10 @@ impl Scheduler { metadata, )) .await - .map_err(|error| anyhow::anyhow!(error.to_string())) + { + tracing::warn!(error = %e, job_id = %job.id, "Dropping silent agent task failure notification"); + } + Ok(()) } } diff --git a/src/tools/shell_session.rs b/src/tools/shell_session.rs index a495aee..98419a3 100644 --- a/src/tools/shell_session.rs +++ b/src/tools/shell_session.rs @@ -140,20 +140,17 @@ impl ShellSessionManager { // Wait for new output or process exit let deadline = Instant::now() + Duration::from_millis(OUTPUT_WAIT_MS); - loop { - tokio::select! { - status = session.child.wait() => { - // Process exited — collect final output - let stdout = session.stdout_buf.lock().await.clone(); - let stderr = session.stderr_buf.lock().await.clone(); - let code = status.ok().and_then(|s| s.code()); - drop(sessions); - return Ok(Self::format_output(&stdout, &stderr, code)); - } - _ = tokio::time::sleep_until(deadline) => { - // Timeout — return current output - break; - } + tokio::select! { + status = session.child.wait() => { + // Process exited — collect final output + let stdout = session.stdout_buf.lock().await.clone(); + let stderr = session.stderr_buf.lock().await.clone(); + let code = status.ok().and_then(|s| s.code()); + drop(sessions); + return Ok(Self::format_output(&stdout, &stderr, code)); + } + _ = tokio::time::sleep_until(deadline) => { + // Timeout — fall through to return current output } } diff --git a/src/tools/task/runtime.rs b/src/tools/task/runtime.rs index d64a2de..3acf74c 100644 --- a/src/tools/task/runtime.rs +++ b/src/tools/task/runtime.rs @@ -128,12 +128,14 @@ impl EmittedMessageHandler for SubAgentEmitter { &message, ) { if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!( - error = %error, - channel = %self.channel_name, - chat_id = %self.chat_id, - "Failed to publish live sub-agent tool call" - ); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live sub-agent tool call"); + } + } } } } @@ -152,12 +154,14 @@ impl EmittedMessageHandler for SubAgentEmitter { &message, ) { if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!( - error = %error, - channel = %self.channel_name, - chat_id = %self.chat_id, - "Failed to publish live sub-agent tool call" - ); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, chat_id = %self.chat_id, "Failed to publish live sub-agent tool call"); + } + } } } @@ -196,7 +200,14 @@ impl EmittedMessageHandler for SubAgentEmitter { }; if let Err(error) = self.bus.publish_outbound(outbound).await { - tracing::error!(error = %error, channel = %self.channel_name, "Failed to publish sub-agent stream delta"); + match error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %error, channel = %self.channel_name, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %error, channel = %self.channel_name, "Failed to publish sub-agent stream delta"); + } + } } } diff --git a/web/src/hooks/chat/useSessions.ts b/web/src/hooks/chat/useSessions.ts index a41e31c..e7b69dc 100644 --- a/web/src/hooks/chat/useSessions.ts +++ b/web/src/hooks/chat/useSessions.ts @@ -1,4 +1,4 @@ -import { useState, useCallback, useMemo, type Dispatch, type SetStateAction } from 'react'; +import { useState, useCallback, useMemo, useEffect, useRef, type Dispatch, type SetStateAction, type MutableRefObject } from 'react'; import type { SessionSummary, Command } from '../../types/protocol'; export interface UseSessionsReturn { @@ -6,6 +6,7 @@ export interface UseSessionsReturn { setSessions: Dispatch>; selectedSessionId: string | null; setSelectedSessionId: Dispatch>; + selectedSessionIdRef: MutableRefObject; session: SessionSummary | null; sessionId: string | null; chatId: string; @@ -22,6 +23,11 @@ export function useSessions(options?: UseSessionsOptions): UseSessionsReturn { const [sessions, setSessions] = useState([]); const [selectedSessionId, setSelectedSessionId] = useState(null); + const selectedSessionIdRef = useRef(null); + useEffect(() => { + selectedSessionIdRef.current = selectedSessionId; + }, [selectedSessionId]); + const selectedSession = useMemo( () => sessions.find((s) => s.session_id === selectedSessionId) ?? null, [sessions, selectedSessionId], @@ -50,6 +56,7 @@ export function useSessions(options?: UseSessionsOptions): UseSessionsReturn { setSessions, selectedSessionId, setSelectedSessionId, + selectedSessionIdRef, session: selectedSession, sessionId, chatId, diff --git a/web/src/hooks/useChat.ts b/web/src/hooks/useChat.ts index ef166f5..d96cb35 100644 --- a/web/src/hooks/useChat.ts +++ b/web/src/hooks/useChat.ts @@ -159,22 +159,42 @@ export function useChat(): UseChatReturn { conn.setConnectionId(message.session_id); return; - case 'session_list': - // 清空旧数据(切换通道时避免数据污染) - topics.setTopics([]); - topics.setSelectedTopic(null); - messages.setMessages([]); + case 'session_list': { + // 重连恢复:重连前已有选中 topic,且原 session 仍存在于新列表中 + // 保留断连前的 messages(用户仍可查看之前的对话),仅刷新 session/topic 列表 + // 注意:不发 load_chat_messages——历史消息不带 topic_id(ChatMessage 结构无此字段), + // 前端无法按 topic 过滤,会导致跨 topic 消息污染 + 与残留消息重复 + // 断连期间的新消息在 SQLite 中,需后端新增按 topic 加载的命令才能恢复(既有限制) + const prevSid = sessions.selectedSessionIdRef.current; + const prevSessionExists = + prevSid !== null && message.sessions.some((s) => s.session_id === prevSid); + const isReconnect = + topics.selectedTopicRef.current !== null && prevSessionExists; sessions.setSessions(message.sessions); - // 自动选中:优先保持当前选中,否则选第一个 - sessions.setSelectedSessionId((prev) => - prev && message.sessions.some((s) => s.session_id === prev) - ? prev - : message.sessions.length > 0 - ? message.sessions[0].session_id - : null, - ); - messages.setIsLoading(false); + if (isReconnect) { + // 原 session 仍在,保持选中 + sessions.setSelectedSessionId(prevSid); + // 刷新 topic 列表(断连期间可能新建了 topic) + const topicCmd = topics.requestTopicList(prevSid!); + if (topicCmd) conn.sendCommand(topicCmd); + // 重置 loading 状态(断连时可能卡在 loading) + messages.setIsLoading(false); + } else { + // 首次连接、切换通道、或原 session 已被删除:清空旧数据避免污染 + topics.setTopics([]); + topics.setSelectedTopic(null); + messages.setMessages([]); + sessions.setSelectedSessionId((prev) => + prev && message.sessions.some((s) => s.session_id === prev) + ? prev + : message.sessions.length > 0 + ? message.sessions[0].session_id + : null, + ); + messages.setIsLoading(false); + } return; + } case 'session_created': case 'session_loaded': diff --git a/web/src/hooks/useWebSocket.ts b/web/src/hooks/useWebSocket.ts index 5ba2a63..c2b19ff 100644 --- a/web/src/hooks/useWebSocket.ts +++ b/web/src/hooks/useWebSocket.ts @@ -18,6 +18,11 @@ interface UseWebSocketReturn { disconnect: () => void; } +// 客户端 ping 间隔(ms):小于典型代理/NAT 60s 空闲超时 +const PING_INTERVAL_MS = 25000; +// 指数退避重连上限(ms) +const MAX_RECONNECT_DELAY_MS = 60000; + export function useWebSocket({ url, onMessage, @@ -25,13 +30,21 @@ export function useWebSocket({ onDisconnect, onError, reconnectInterval = 3000, - maxReconnectAttempts = 5, + maxReconnectAttempts = 999, }: UseWebSocketOptions): UseWebSocketReturn { const [status, setStatus] = useState('disconnected'); const wsRef = useRef(null); const reconnectAttemptsRef = useRef(0); const reconnectTimerRef = useRef | null>(null); const isManualDisconnectRef = useRef(false); + const pingIntervalRef = useRef | null>(null); + + const clearPing = useCallback(() => { + if (pingIntervalRef.current) { + clearInterval(pingIntervalRef.current); + pingIntervalRef.current = null; + } + }, []); const connect = useCallback(() => { if (wsRef.current?.readyState === WebSocket.OPEN) { @@ -48,6 +61,18 @@ export function useWebSocket({ ws.onopen = () => { setStatus('connected'); reconnectAttemptsRef.current = 0; + // 启动客户端 ping:定期写数据保持连接活性,写失败会触发 onclose 检测半死连接 + clearPing(); + pingIntervalRef.current = setInterval(() => { + if (wsRef.current?.readyState === WebSocket.OPEN) { + try { + wsRef.current.send(JSON.stringify({ type: 'ping' })); + } catch { + // 极端时序下 send 可能抛 InvalidStateError,忽略即可 + // onclose 会随后触发,进入重连流程 + } + } + }, PING_INTERVAL_MS); onConnect?.(); }; @@ -67,21 +92,24 @@ export function useWebSocket({ ws.onclose = () => { setStatus('disconnected'); + clearPing(); onDisconnect?.(); - // Auto reconnect if not manually disconnected + // 指数退避自动重连:3s → 6s → 12s → 24s → 60s 封顶 if (!isManualDisconnectRef.current && reconnectAttemptsRef.current < maxReconnectAttempts) { reconnectAttemptsRef.current += 1; + const attempts = reconnectAttemptsRef.current; + const delay = Math.min(reconnectInterval * Math.pow(2, attempts - 1), MAX_RECONNECT_DELAY_MS); reconnectTimerRef.current = setTimeout(() => { connect(); - }, reconnectInterval); + }, delay); } }; } catch (error) { setStatus('error'); console.error('WebSocket connection error:', error); } - }, [url, onMessage, onConnect, onDisconnect, onError, reconnectInterval, maxReconnectAttempts]); + }, [url, onMessage, onConnect, onDisconnect, onError, reconnectInterval, maxReconnectAttempts, clearPing]); const disconnect = useCallback(() => { isManualDisconnectRef.current = true; @@ -91,13 +119,15 @@ export function useWebSocket({ reconnectTimerRef.current = null; } + clearPing(); + if (wsRef.current) { wsRef.current.close(); wsRef.current = null; } setStatus('disconnected'); - }, []); + }, [clearPing]); const sendMessage = useCallback((message: WsInbound): boolean => { if (wsRef.current?.readyState === WebSocket.OPEN) {