From 5bcf22f9a771c0cdcc042f3defdc0bef1a774c57 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Mon, 17 Aug 2026 22:47:37 +0800 Subject: [PATCH] fix(aibox): registry timeout cleanup + reader thread finalization (F2/N6+N4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit N6: both streaming-forward timeout paths (deadline hit, join(5) expiry) previously returned early and skipped the tool.streaming_output / tool.output_queue cleanup, leaving the daemon thread writing through the instance attribute — zombie output could leak into the next session's queue once the attribute got reassigned. Now unified through a timed_out flag with cleanup at the single exit; stream_tool_timeout becomes an instance attribute (3600s default, test-tunable). N4: reader-thread finalization after timeout/grace break. First implementation followed the audit suggestion (close stdout + join), but PoC regression proved close() blocks: CPython stream objects hold an internal lock, so closing while readline is blocked parks until readline returns (PoC-1 regressed 3s -> 15.2s, exactly the Start-Sleep duration). Corrected strategy: when reader already got EOF, close stdout to release the pipe fd; when reader is still blocked, only join(timeout=2) then give up (daemon-bounded, thread ends naturally when the grandchild exits and EOF arrives), keeping the main flow prompt. Audit report updated with this correction. Tests: new tests/test_f2_fixes.py (5 cases), PoC regression 5/5 (PoC-1/4 back to ~5s), full suite 272 passed. --- aibox/subsystems/executors.py | 20 ++++ aibox/tests/test_f2_fixes.py | 171 +++++++++++++++++++++++++++ aibox/tools/registry.py | 31 +++-- docs/reports/aibox_audit_20260817.md | 7 ++ 4 files changed, 221 insertions(+), 8 deletions(-) create mode 100644 aibox/tests/test_f2_fixes.py diff --git a/aibox/subsystems/executors.py b/aibox/subsystems/executors.py index 1cdef3a..ead795d 100644 --- a/aibox/subsystems/executors.py +++ b/aibox/subsystems/executors.py @@ -159,6 +159,26 @@ class SubprocessCommandExecutor(CommandExecutor): if output_queue: output_queue.put(("line", line.rstrip('\n') + "\n")) + # N4 修复（审计 2026-08-18，二次修正）：reader 线程收尾。 + # 超时/截断/宽限到期 break 后，reader 可能仍阻塞在 readline + # （Windows shell=True 下孙进程持管道写端，永远等不到 EOF）。 + # 注意：此时不能 close stdout——CPython 流对象内部有锁， + # readline 持锁阻塞时 close() 会一同挂起直到 readline 返回 + # （实测命令超时被拖满整个 Start-Sleep 时长）。reader 为 daemon + # 线程，孙进程退出收到 EOF 后自然结束，不阻塞主流程； + # reader 已正常退出的场景才 close 显式释放管道 fd。 + if reader_done.is_set(): + try: + process.stdout.close() + except Exception: + pass + else: + reader.join(timeout=2) + if reader.is_alive(): + self.logger.warning( + "reader 线程仍阻塞在管道读取（孙进程持写端），已放弃等待，" + "线程由 daemon 兜底、随管道 EOF 自然退出") + return ''.join(output_chunks), timeout_occurred, truncated def execute(self, command: str, timeout: int = 30, work_dir: str = None, env: Dict = None, **kwargs) -> str: diff --git a/aibox/tests/test_f2_fixes.py b/aibox/tests/test_f2_fixes.py new file mode 100644 index 0000000..00dbfea --- /dev/null +++ b/aibox/tests/test_f2_fixes.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""F2 修复回归测试：N6 registry 超时清理 + N4 reader 线程收尾。 + +N6：流式转发超时后必须清理 tool.streaming_output / tool.output_queue， + 防止僵尸 daemon 线程经实例属性把输出混入后续会话队列。 +N4：_read_with_limit 超时/宽限 break 后必须 close stdout + join reader， + 防止 Windows shell=True 孙进程持管道场景下 reader 线程与 fd 泄漏。 +""" + +import queue +import threading +import time + +import pytest + +from subsystems.executors import SubprocessCommandExecutor +from tools.base import Tool +from tools.registry import ToolRegistry + + +# ---------- N6：registry 超时路径清理 ---------- + +class _StuckStreamTool(Tool): + """execute_streaming 永不结束（模拟卡死的工具，如等待外部资源）""" + + def get_name(self): + return "stuck_tool" + + def get_description(self): + return "test only" + + def get_parameters(self): + return {"type": "object", "properties": {}} + + def execute(self, **kwargs): + time.sleep(30) + return "should not reach" + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + self.output_queue = output_queue + self.streaming_output = True + try: + time.sleep(30) # 卡死，永不产出 complete + finally: + self.streaming_output = False + + +def test_registry_timeout_cleans_tool_state(): + """转发总超时后：返回超时消息，且工具实例属性被清理""" + reg = ToolRegistry() + reg.stream_tool_timeout = 1 # 测试调小（默认 3600） + tool = _StuckStreamTool() + reg.register(tool) + reg.set_output_queue(queue.Queue()) + + t0 = time.time() + result = reg.execute_tool("stuck_tool") + elapsed = time.time() - t0 + + assert "超时" in result, f"应返回超时消息，实际: {result}" + assert elapsed < 10, f"应 ~1s 返回，实际 {elapsed:.1f}s" + # N6 核心：超时路径不得跳过清理 + assert tool.output_queue is None, "超时后 tool.output_queue 必须置 None（防僵尸输出串流）" + assert tool.streaming_output is False, "超时后 tool.streaming_output 必须复位" + + +def test_registry_normal_path_cleans_tool_state(): + """正常流式完成后：清理同样执行，结果透传""" + reg = ToolRegistry() + + class _OkTool(_StuckStreamTool): + def get_name(self): + return "ok_tool" + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + self.output_queue = output_queue + self.streaming_output = True + try: + output_queue.put(("line", "hello")) + output_queue.put(("complete", "done-result")) + finally: + pass + + tool = _OkTool() + reg.register(tool) + reg.set_output_queue(queue.Queue()) + + result = reg.execute_tool("ok_tool") + assert result == "done-result" + assert tool.output_queue is None + assert tool.streaming_output is False + + +# ---------- N4：reader 线程收尾 ---------- + +class _FakeStdout: + """模拟管道 stdout，可配置阻塞时长""" + + def __init__(self, block_seconds: float = 8): + self.closed = False + self._block = block_seconds + + def readline(self): + time.sleep(self._block) # > join 宽限 2s 即模拟"无 EOF 阻塞" + return "" + + def close(self): + self.closed = True + + +class _FakeProcess: + def __init__(self, block_seconds: float = 8): + self.stdout = _FakeStdout(block_seconds) + + def poll(self): + return None # 进程"活着" + + def terminate(self): + pass + + def kill(self): + pass + + +def test_read_with_limit_timeout_returns_promptly(): + """超时 break 后 reader 仍阻塞：只做有限 join，不 close、不挂死。 + + CPython 流锁决定了此时 close(stdout) 会阻塞到 readline 返回， + 故收尾策略为放弃等待（daemon 兜底），主流程必须及时返回。 + """ + ex = SubprocessCommandExecutor() + proc = _FakeProcess(block_seconds=8) + + t0 = time.time() + output, timeout_occurred, truncated = ex._read_with_limit( + proc, timeout=1, start_time=time.time()) + elapsed = time.time() - t0 + + assert timeout_occurred is True + assert truncated is False + # 超时 1s + join 宽限 2s，绝不能等到 block 8s + assert elapsed < 5, f"应收尾后及时返回，实际 {elapsed:.1f}s" + + +def test_read_with_limit_grace_path_returns_promptly(): + """shell 死亡宽限 break 后 reader 仍阻塞：同样及时返回""" + ex = SubprocessCommandExecutor() + proc = _FakeProcess(block_seconds=8) + proc.poll = lambda: 0 # shell 已死，但 reader 无 EOF（孙进程持管道） + + t0 = time.time() + output, timeout_occurred, truncated = ex._read_with_limit( + proc, timeout=0, start_time=time.time()) # timeout=0 不触发超时 + elapsed = time.time() - t0 + + assert timeout_occurred is False + # 宽限 1s + join 2s 上界，不能等到 block 8s + assert elapsed < 5, f"宽限+收尾应 ~3s 内返回，实际 {elapsed:.1f}s" + + +def test_read_with_limit_closes_stdout_after_normal_eof(): + """reader 正常收到 EOF 退出后：显式 close stdout 释放管道 fd""" + ex = SubprocessCommandExecutor() + proc = _FakeProcess(block_seconds=0) # readline 立即返回 ""（EOF） + + output, timeout_occurred, truncated = ex._read_with_limit( + proc, timeout=5, start_time=time.time()) + + assert timeout_occurred is False + assert output == "" + assert proc.stdout.closed is True, "reader 正常退出后必须 close stdout 释放 fd" diff --git a/aibox/tools/registry.py b/aibox/tools/registry.py index 5440eba..8286974 100644 --- a/aibox/tools/registry.py +++ b/aibox/tools/registry.py @@ -36,6 +36,8 @@ class ToolRegistry: self.logger = logger or Logger("ToolRegistry") self.current_output_queue = None # 用于流式输出的队列 self.skill_manager: Optional['SkillManager'] = None # 技能管理器引用 + # 单工具流式执行总上限（工具自身秒级超时之外的兜底；测试可调小） + self.stream_tool_timeout = 3600 def register(self, tool: Tool, is_builtin: bool = False): """注册工具，可以标记是否为内置工具""" @@ -100,14 +102,20 @@ class ToolRegistry: # 此前 while thread.is_alive() 在工具内部卡死（如子进程 # 等待 stdin）时永不退出；超时后放弃等待，避免本调用 # 线程被连带挂起。 - tool_timeout = 3600 # 单工具总上限 1 小时（工具自身的秒级超时之外的兜底） + # N6 修复（审计 2026-08-18）：超时路径此前直接 return， + # 跳过下方清理——daemon 线程继续持有 tool.output_queue + # 实例属性，属性被后续调用覆盖到新会话队列后，僵尸输出 + # 会混入后续对话流。现在统一在出口清理。 + tool_timeout = self.stream_tool_timeout deadline = time.time() + tool_timeout final_result = None + timed_out = False while thread.is_alive() or not output_queue.empty(): if time.time() > deadline: self.logger.warning( f"工具 {name} 流式执行超过 {tool_timeout}s 总上限，放弃等待") - return f"❌ 工具执行超时（>{tool_timeout}s，已放弃等待）" + timed_out = True + break try: msg_type, content = output_queue.get(timeout=0.1) if msg_type == "line": @@ -121,14 +129,21 @@ class ToolRegistry: self.current_output_queue.put(("complete", content)) except queue.Empty: continue - - thread.join(timeout=5) - if thread.is_alive(): - return "❌ 工具执行超时" - - result = final_result or "" + + if not timed_out: + thread.join(timeout=5) + if thread.is_alive(): + timed_out = True + + # 统一清理（正常与超时路径都必须执行）：断开实例属性， + # 僵尸 daemon 线程的 send_line 落到 None 上不再外写 tool.streaming_output = False tool.output_queue = None + + if timed_out: + result = f"❌ 工具执行超时（>{tool_timeout}s，已放弃等待）" + else: + result = final_result or "" else: # 普通执行 result = tool.execute(**kwargs) diff --git a/docs/reports/aibox_audit_20260817.md b/docs/reports/aibox_audit_20260817.md index b1ec668..72bdd45 100644 --- a/docs/reports/aibox_audit_20260817.md +++ b/docs/reports/aibox_audit_20260817.md @@ -219,3 +219,10 @@ aibox 全量单测 **264 passed**（零回归）。 - `tools/registry.py`（N11）：debug 日志对 `password/api_key/token/secret/authorization` 参数打码 - 测试：新增 `tests/test_config.py` 3 例（落盘无明文 / password_env 解析 / 旧配置迁移），全量 267 passed - 行为变化：已有明文密码的配置重启后代理需重设密码或改用环境变量（迁移日志有提示）——清除存量明文的必然代价 + +**F2 ✅ 已落地（2026-08-18）**——N6 + N4 一并修复： + +- N6（registry 超时路径补清理）：两个超时 return（deadline 命中、join(5) 超时）统一改为 `timed_out` 标记 + 出口统一执行 `tool.streaming_output=False` / `tool.output_queue=None`；僵尸 daemon 线程的 `send_line` 落到 None 上不再外写，消除僵尸输出串流路径。`stream_tool_timeout` 提为实例属性（默认 3600s，测试可调） +- N4（reader 线程收尾，**含二次修正**）：初版按审计建议"break 后 `stdout.close()` + `reader.join(2)`"实现，PoC 回归实证 **close 会阻塞**——CPython 流对象内部有锁，readline 持锁阻塞时 `close()` 一同挂起直到 readline 返回（PoC-1 从 3s 退化到 15.2s，恰为命令 Start-Sleep 时长）。修正后策略：reader 已正常退出（EOF）→ 显式 close 释放管道 fd；reader 仍阻塞 → 只做 `join(timeout=2)` 后放弃（daemon 兜底，孙进程退出收 EOF 后自然结束，fd 随管道消亡回收），主流程及时返回 +- **审计建议修正记录**：8.3 节 N4 与 8.7 节 F2 中"close stdout 可中断阻塞 readline（CPython 行为）"的表述**不成立**，已按实测修正——跨线程中断阻塞中的管道 read 无便携方案，正确做法是控制泄漏窗口而非强行中断 +- 测试：新增 `tests/test_f2_fixes.py` 5 例（N6 超时/正常清理 ×2、N4 及时返回 ×2 + EOF 后 close ×1），PoC 回归 5/5（PoC-1/4 恢复 ~5s：超时 3s + join 宽限 2s），全量 272 passed