diff --git a/aibox/aibox.py b/aibox/aibox.py index dd9fbe5..470b555 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -7133,7 +7133,32 @@ class DeepSeekChat: result_raw = None try: + # 启动后台线程持续排空 output_queue,将流式输出实时转发给 output_callback, + # 避免 SubAgent 等长时间工具执行期间前端收不到数据导致 WS 空闲断线 + def _drain_queue(q=output_queue, cb=output_callback, stop_event=self._cancel_event): + while not stop_event.is_set(): + try: + msg_type, content = q.get(timeout=0.1) + if cb and msg_type == "line": + cb("line", str(content) if content is not None else "") + except queue.Empty: + continue + + import threading as _th + drain_thread = _th.Thread(target=_drain_queue, daemon=True) + drain_thread.start() + result_raw = self.tool_registry.execute_tool(tool_name, **args) + + # 工具执行完毕,排空队列中剩余内容并等待 drain 线程退出 + while not output_queue.empty(): + try: + msg_type, content = output_queue.get_nowait() + if output_callback and msg_type == "line": + output_callback("line", str(content) if content is not None else "") + except queue.Empty: + break + # drain 线程会在下一次 get 超时后自动退出(队列已空) except Exception as e: result_raw = f"❌ 工具 {tool_name} 执行异常: {str(e)}" self.logger.error(f"工具 {tool_name} 执行异常: {e}") @@ -7141,18 +7166,8 @@ class DeepSeekChat: # 安全转换并清理结果 result = self._sanitize_tool_result(result_raw) - # 排空 output_queue,把流式行输出转发给 output_callback - has_stream_output = False - while not output_queue.empty(): - try: - msg_type, content = output_queue.get_nowait() - if output_callback and msg_type == "line": - # 只转发"line"类型;"complete"是流式结束信号,不作为行输出 - # (之前complete也被转发导致工具结果在侧栏显示两次) - output_callback("line", str(content) if content is not None else "") - has_stream_output = True - except queue.Empty: - break + # drain 线程已排空队列中的 line 输出,这里只需判断是否有流式输出 + has_stream_output = not output_queue.empty() # 普通(非流式)工具没有队列输出,直接发送结果 if not has_stream_output and output_callback and result: diff --git a/libhttp/src/HttpServer_WsSession.cpp b/libhttp/src/HttpServer_WsSession.cpp index 0b1016c..8366671 100644 --- a/libhttp/src/HttpServer_WsSession.cpp +++ b/libhttp/src/HttpServer_WsSession.cpp @@ -82,8 +82,12 @@ void HttpServer::Session::doWsRead() { auto self = shared_from_this(); ws_buffer_.consume(ws_buffer_.size()); - // P1-4:WS 读取设置空闲超时,清理长期不发消息的连接 - stream_.expires_after(server_->getReadTimeout()); + // P1-4:WS 读取设置空闲超时,清理长期不发消息的连接。 + // WebSocket 代理模式(如 aibox)不设超时:后端 AI 处理可能长时间无数据, + // 由前端心跳保活 + 后端自身超时机制管理连接生命周期。 + if (!is_ws_proxy_) { + stream_.expires_after(server_->getReadTimeout()); + } if (ssl_enabled_ && wss_stream_) { wss_stream_->async_read( @@ -329,6 +333,12 @@ void HttpServer::Session::onProxyHandshake(beast::error_code ec) { proxy_connected_ = true; proxy_closed_ = false; + + // 清除客户端侧流在 HTTP 升级阶段继承的读 deadline(body 阶段设置的 30s)。 + // 否则代理连接会在握手后约 30s 被强制关闭,导致长连接 WebSocket 断线。 + // 注意:后端侧 proxy_tcp_stream_ 已在 onProxyConnect 中 expires_never()。 + stream_.expires_never(); + std::cout << "[WS-Proxy] Connected to backend " << proxy_host_ << ":" << proxy_port_ << proxy_path_ << std::endl; diff --git a/static/public/js/aibox.js b/static/public/js/aibox.js index c9d1717..41c48ba 100644 --- a/static/public/js/aibox.js +++ b/static/public/js/aibox.js @@ -44,6 +44,21 @@ let ws = null; let manualClose = false; /* 手动断开则抑制自动重连 */ let reconnectDelay = RECONNECT_MIN; let reconnectTimer = null; +let heartbeatTimer = null; /* 心跳定时器,25s 发送 ping 保持 WS 连接 */ +const HEARTBEAT_INTERVAL = 25000; /* 心跳间隔(< gateway 30s 空闲超时) */ + +function startHeartbeat() { + stopHeartbeat(); + heartbeatTimer = setInterval(() => { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: 'ping' })); + } + }, HEARTBEAT_INTERVAL); +} + +function stopHeartbeat() { + if (heartbeatTimer) { clearTimeout(heartbeatTimer); heartbeatTimer = null; } +} /* 流式状态:不预创建空气泡,第一个 chunk 到来才挂载(保证思考区必然在正式回答之前) */ let aiBubble = null, aiText = '', aiActive = false; @@ -204,6 +219,7 @@ function connect() { setStatus('connected', '已连接'); connBtn.textContent = '断开'; connBtn.classList.add('on'); + startHeartbeat(); }; ws.onmessage = (ev) => { let data; @@ -212,6 +228,7 @@ function connect() { routeMessage(data); }; ws.onclose = () => { + stopHeartbeat(); setStatus('disconnected', '已断开'); connBtn.textContent = '连接'; connBtn.classList.remove('on'); @@ -235,6 +252,7 @@ function scheduleReconnect() { function disconnect(manual) { manualClose = manual !== false; clearTimeout(reconnectTimer); + stopHeartbeat(); if (ws) { ws.close(); ws = null; } setStatus('disconnected', manualClose ? '已手动断开' : '已断开'); connBtn.textContent = '连接';