fix(aibox): v2.6 - 修复Ctrl+C中断无法停止服务器的问题

问题根因： 1. WebSocketHandler.handle_message 完全没有处理 interrupt 消息类型（前端Ctrl+C发送的消息被丢弃） 2. DeepSeekChat 缺少取消机制，阻塞的 API 请求(iter_lines)和子进程无法被中断修复方案 - 三层取消机制： 1. threading.Event 取消标志：流式响应循环每轮检查 _is_cancelled()，工具调用前检查 2. HTTP连接关闭：保存 _current_response 引用，cancel() 时调用 response.close() 中断 iter_lines() 3. 子进程终止：SubprocessCommandExecutor 添加 cancel() 方法，保存 _current_process，cancel时kill 修改内容： - DeepSeekChat: 添加 _cancel_event/_current_response/_command_executor 字段，cancel()/_is_cancelled()/_check_cancelled() 方法 - SubprocessCommandExecutor: 添加 __init__/_process_lock/cancel()，execute()/execute_streaming() 中设置和清除进程引用 - WebSocketHandler: 添加 handle_interrupt() 处理 interrupt 消息，回复 interrupt_ack 后调用 ai_manager.cancel() - EnhancedAIManager: 添加 cancel() 统一入口，同时取消AI和命令执行器 - think_and_respond: 首轮清除cancel标志，流式循环检查取消，API/JSON/未知异常中均检查取消标志 - _handle_tool_calls: 每个工具调用前检查取消信号 - 更新README v2.6日志和API文档
2026-08-01 14:09:24 +08:00
1 parent b935fefb82
commit 9d737897df
@@ -21,6 +21,7 @@
 - 📦 **对话压缩**：消息数或 Token 数达到阈值时自动压缩，保持上下文质量
 - 🛡️ **内存安全防护**：文件读取按需加载不全载、命令输出 1MB 限制、单行超长截断、工具结果多层防护防止 OOM
 - 🔒 **错误自动恢复**：API 请求失败自动回滚消息，工具结果安全清洗（控制字符过滤、非字符串转换）
+- 🛑 **可靠中断机制**：支持 Ctrl+C 实时中断，通过 `threading.Event` + HTTP 连接关闭 + 子进程 kill 三层取消，流式响应和长时间运行命令均可立即终止

 ## 🚀 快速开始

@@ -192,6 +193,15 @@ const ws = new WebSocket('ws://localhost:8765');
 }
 ```

+#### 5. 中断（Ctrl+C）
+```json
+{
+  "type": "interrupt",
+  "content": "用户按下了 Ctrl+C 中断当前操作"
+}
+```
+服务端收到后立即回复 `interrupt_ack`，并通过三层取消机制（Event标志 + HTTP连接关闭 + 子进程kill）终止正在进行的AI思考和工具执行。
+
 ### 支持的命令

 > 注意：记忆搜索通过 `memory` 工具（在对话中由 AI 调用）完成，本系统**不存在** `/search` 斜杠命令。
@@ -706,6 +716,19 @@ elif command == "/mycommand":

 ## 📄 更新日志

+### v2.6
+- ✅ **修复 Ctrl+C 中断无法停止服务器的问题**：
+  - 新增 `interrupt` WebSocket 消息类型处理（BotOps 前端 Ctrl+C 发送此消息），立即回复 `interrupt_ack` 确认
+  - `DeepSeekChat` 新增 `cancel()` 方法和 `_cancel_event`（threading.Event），线程安全可跨线程调用
+  - **三层取消机制**：
+    1. 设置取消标志，流式响应循环每轮检查，发现取消立即 break
+    2. 关闭当前 HTTP response 对象，中断 `iter_lines()` 阻塞读取
+    3. 终止正在运行的子进程（`SubprocessCommandExecutor.cancel()` → `process.kill()`）
+  - 工具调用循环中每个工具执行前检查取消信号，中断后不再继续后续工具
+  - API 请求异常/JSON解析异常/未知异常中均检查取消标志，若因取消导致异常则按中断处理而非报错
+  - 中断后自动回滚本轮消息，防止污染对话历史
+  - 新增 `EnhancedAIManager.cancel()` 统一入口，同时终止 AI 和命令执行器
+
 ### v2.5
 - ✅ **新增 DeepSeek 思考模式支持**（Reasoning Mode）：
  - 新增 `api.reasoning_enabled` 配置项（默认 `false`），启用后在请求中添加 `thinking: {type: "enabled"}` 参数
@@ -3532,11 +3532,28 @@ class CommandExecutor(ABC):


 class SubprocessCommandExecutor(CommandExecutor):
-    """基于subprocess的命令执行器 - 支持超时、工作目录和输出大小限制"""
+    """基于subprocess的命令执行器 - 支持超时、工作目录、输出大小限制和取消"""
    
    # 命令输出最大字节数（1MB），超出截断
    MAX_OUTPUT_BYTES = 1 * 1024 * 1024
    
+    def __init__(self, logger=None):
+        super().__init__(logger)
+        self._current_process = None
+        self._process_lock = threading.Lock()
+    
+    def cancel(self):
+        """取消当前正在运行的命令（kill子进程）。线程安全。"""
+        with self._process_lock:
+            proc = self._current_process
+        if proc is not None:
+            try:
+                if proc.poll() is None:
+                    proc.kill()
+                    self.logger.info("已终止正在运行的子进程")
+            except Exception:
+                pass
+    
    def _read_with_limit(self, process, timeout: int, start_time: float, output_queue=None) -> Tuple[str, bool, bool]:
        """从进程读取输出，限制总大小和超时。
        返回 (output_str, timeout_occurred, truncated)
@@ -3612,6 +3629,9 @@ class SubprocessCommandExecutor(CommandExecutor):
                env=exec_env,
                cwd=os.path.expanduser(work_dir)
            )
+            # 保存进程引用以便cancel()可以kill它
+            with self._process_lock:
+                self._current_process = process
            
            start_time = time.time()
            output_str, timeout_occurred, truncated = self._read_with_limit(process, timeout, start_time)
@@ -3654,6 +3674,9 @@ class SubprocessCommandExecutor(CommandExecutor):
        finally:
            if process and process.poll() is None:
                process.kill()
+            # 清除当前进程引用
+            with self._process_lock:
+                self._current_process = None
    
    def execute_streaming(self, command: str, output_queue: queue.Queue, timeout: int = 30, 
                          work_dir: str = None, env: Dict = None, **kwargs):
@@ -3694,6 +3717,9 @@ class SubprocessCommandExecutor(CommandExecutor):
                cwd=os.path.expanduser(work_dir)
            )
            self.process = process
+            # 保存进程引用以便cancel()可以kill它
+            with self._process_lock:
+                self._current_process = process
            
            start_time = time.time()
            output_lines = []
@@ -3788,6 +3814,8 @@ class SubprocessCommandExecutor(CommandExecutor):
                except:
                    process.kill()
            self.process = None
+            with self._process_lock:
+                self._current_process = None

 # ==================== 工具基类 ====================

@@ -6750,11 +6778,42 @@ class DeepSeekChat:
        self.preserve_system = constants.PRESERVE_SYSTEM_PROMPTS
        self.stats = {"api_calls": 0, "tool_calls": 0, "errors": []}
        
+        # 取消机制：threading.Event 用于跨线程信号，_current_response 用于中断HTTP请求
+        self._cancel_event = threading.Event()
+        self._current_response = None
+        self._command_executor = command_executor  # 保存命令执行器引用，用于取消时kill子进程
+        
        self._init_system_prompts(system_prompt)
        
        self.tool_registry = ToolRegistry(self.logger)
        self._register_tools(exit_callback, command_executor)
    
+    def cancel(self):
+        """中断当前正在进行的AI思考/工具执行。线程安全，可从任意线程调用。"""
+        self.logger.info("收到中断信号，正在取消当前操作...")
+        self._cancel_event.set()
+        # 关闭当前HTTP响应，中断 iter_lines() 阻塞
+        if self._current_response is not None:
+            try:
+                self._current_response.close()
+            except Exception:
+                pass
+        # 终止正在运行的子进程
+        if self._command_executor is not None and hasattr(self._command_executor, 'cancel'):
+            try:
+                self._command_executor.cancel()
+            except Exception:
+                pass
+    
+    def _is_cancelled(self) -> bool:
+        """检查是否已被取消。"""
+        return self._cancel_event.is_set()
+    
+    def _check_cancelled(self):
+        """如果已取消，抛出异常中断当前流程。"""
+        if self._cancel_event.is_set():
+            raise InterruptedError(i18n.get('user_interrupted', '用户中断'))
+    
    def _init_system_prompts(self, system_prompt: str = None):
        """初始化系统提示词 - 从 PromptManager 加载（按 identity → 技能 → soul → status 顺序）"""
        if system_prompt:
@@ -6965,6 +7024,11 @@ class DeepSeekChat:
        MAX_RESULT_LENGTH = constants.TOOL_MAX_RESULT_LENGTH
        
        for tool_call in tool_calls:
+            # 每个工具调用前检查取消信号
+            if self._is_cancelled():
+                self.logger.info("工具调用被用户中断")
+                break
+            
            function = tool_call.get("function", {})
            tool_name = function.get("name", "unknown")
            arguments = function.get("arguments", "{}")
@@ -7076,6 +7140,13 @@ class DeepSeekChat:
            is_continuation: 是否为工具调用后的递归继续；为 True 时不注入冗长的工作流指令，
                            因为 system prompt 中已有完整行为规范，继续时只需传递工具结果
        """
+        # 首轮开始时清除取消标志
+        if not is_continuation:
+            self._cancel_event.clear()
+        
+        # 立即检查是否已被取消
+        self._check_cancelled()
+        
        if not self.conversation_active:
            self.logger.info("对话已结束，重置状态")
            self.conversation_active = True
@@ -7217,6 +7288,7 @@ class DeepSeekChat:
                stream=True,
                proxies=proxies   # 动态代理
            )
+            self._current_response = response  # 保存引用以便cancel()关闭连接
            
            if response.status_code != 200:
                error_msg = f"\n错误状态码: {response.status_code}\n错误响应: {response.text}"
@@ -7225,6 +7297,9 @@ class DeepSeekChat:
                response.raise_for_status()
            
            for line in response.iter_lines():
+                # 流式循环中检查取消信号
+                if self._is_cancelled():
+                    break
                if line:
                    line = line.decode('utf-8')
                    if line.startswith('data: '):
@@ -7315,6 +7390,16 @@ class DeepSeekChat:
                            self.logger.error(f"JSON解析错误: {e}, 数据: {data}")
                            continue
            
+            # 流式循环结束，检查是否被取消
+            if self._is_cancelled():
+                self._rollback_messages(msgs_before_turn)
+                interrupt_msg = i18n.get('user_interrupted', '用户中断')
+                if output_callback:
+                    output_callback("line", "\n" + interrupt_msg + "\n")
+                    output_callback("complete", interrupt_msg)
+                self.logger.info("AI响应被用户中断")
+                return interrupt_msg
+            
            # 处理工具调用
            if current_tool_calls:
                tool_calls_buffer = list(current_tool_calls.values())
@@ -7382,7 +7467,24 @@ class DeepSeekChat:
            
            return full_response if full_response else ""
                
+        except InterruptedError:
+            # 用户主动中断：回滚消息，通知前端
+            self._rollback_messages(msgs_before_turn)
+            interrupt_msg = i18n.get('user_interrupted', '用户中断')
+            if output_callback:
+                output_callback("line", "\n" + interrupt_msg + "\n")
+                output_callback("complete", interrupt_msg)
+            self.logger.info("AI响应被用户中断")
+            return interrupt_msg
        except requests.exceptions.RequestException as e:
+            # 如果是因为取消导致的连接关闭，按中断处理
+            if self._is_cancelled():
+                self._rollback_messages(msgs_before_turn)
+                interrupt_msg = i18n.get('user_interrupted', '用户中断')
+                if output_callback:
+                    output_callback("line", "\n" + interrupt_msg + "\n")
+                    output_callback("complete", interrupt_msg)
+                return interrupt_msg
            # 网络/HTTP 错误：回滚本轮添加的消息（user 消息 + 可能的 assistant/tool 消息），
            # 防止格式错误的消息污染历史导致后续请求持续失败
            self._rollback_messages(msgs_before_turn)
@@ -7402,6 +7504,14 @@ class DeepSeekChat:
            self.logger.error(f"API响应解析失败，已回滚消息到位置 {msgs_before_turn}: {e}")
            return error_msg
        except Exception as e:
+            # 如果是因为取消导致的异常，按中断处理
+            if self._is_cancelled():
+                self._rollback_messages(msgs_before_turn)
+                interrupt_msg = i18n.get('user_interrupted', '用户中断')
+                if output_callback:
+                    output_callback("line", "\n" + interrupt_msg + "\n")
+                    output_callback("complete", interrupt_msg)
+                return interrupt_msg
            # 未知错误：回滚消息，防止对话状态被破坏
            self._rollback_messages(msgs_before_turn)
            error_msg = f"\n{i18n.get('error_prefix')}{str(e)}"
@@ -7410,6 +7520,9 @@ class DeepSeekChat:
            self.stats["errors"].append(str(e))
            self.logger.error(f"未知错误，已回滚消息到位置 {msgs_before_turn}: {e}")
            return error_msg
+        finally:
+            # 清除当前响应引用
+            self._current_response = None
    
    def _rollback_messages(self, to_length: int):
        """回滚 self.messages 到指定长度，用于 API 失败时恢复对话状态。"""
@@ -7623,6 +7736,8 @@ class WebSocketHandler:
                await self.handle_command(websocket, data)
            elif msg_type == "message":
                await self.handle_user_message(websocket, data)
+            elif msg_type == "interrupt":
+                await self.handle_interrupt(websocket, data)
            elif msg_type == "task":
                await self.handle_task_message(websocket, data)
            elif msg_type == "ping":
@@ -7648,6 +7763,18 @@ class WebSocketHandler:
                "content": f"处理失败: {str(e)}"
            })
    
+    async def handle_interrupt(self, websocket, data):
+        """处理中断信号：取消正在进行的AI思考和工具执行"""
+        self.logger.info("收到客户端中断信号")
+        # 先发送确认，让前端立即得到反馈
+        await self.send_message(websocket, {
+            "type": "interrupt_ack",
+            "content": "已收到中断信号，正在停止..."
+        })
+        # 在事件循环线程外执行cancel（cancel是线程安全的，会设置Event并关闭response/kill进程）
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(None, self.ai_manager.cancel)
+    
    async def handle_task_message(self, websocket, data):
        """处理任务相关消息 - 简化版，所有任务都是对话任务"""
        action = data.get("action", "").strip()
@@ -8109,6 +8236,17 @@ class EnhancedAIManager:
        self.logger.info("AI已初始化")
        return ai
    
+    def cancel(self):
+        """中断当前AI操作（线程安全，可从任意线程调用）"""
+        if self.ai:
+            self.ai.cancel()
+        # 同时确保命令执行器中的进程也被终止
+        if self.command_executor and hasattr(self.command_executor, 'cancel'):
+            try:
+                self.command_executor.cancel()
+            except Exception:
+                pass
+    
    def show_welcome_message(self):
        """显示欢迎信息"""
        print("\n" + "="*70)