From 5cb35945a46b3086105427384cae2064bd902474 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Fri, 21 Aug 2026 11:17:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(aibox):=20=E5=90=8E=E5=8F=B0=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E6=B1=A0=20v3.8=E2=80=94=E2=80=94=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=E8=BF=90=E8=A1=8C/=E7=AE=A1=E7=90=86=E9=95=BF?= =?UTF-8?q?=E9=A9=BB=E8=BF=9B=E7=A8=8B=E4=B8=8E=E4=BA=A4=E4=BA=92=E5=BC=8F?= =?UTF-8?q?=20I/O?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 ProcessManager 子系统 + 5 个 worker 执行面工具 （bg_run/bg_write/bg_list/bg_output/bg_kill），弥补 run_command 同步阻塞、超时即杀的不足；run_command 原路径零改动。 - 生命周期三决策：用户中断不影响后台进程；aibox 退出经 atexit 收尾（非 detach 全杀，detach=true 豁免）；注册表跨会话/跨子代理 全局共享 - 交互式 I/O：stdin 管道 + watermark 写读对齐（bg_output(since= watermark) 必然且仅为本条输入的响应）；wait_lines expect 式同步 等待；写入 watchdog 线程兜底防管道写满阻塞 - 输出缓冲：每进程 256KB 环形缓冲（单行截断 16KB）；log_file=true 同步 tee 无界全档，可被原生工具 tail/cat 读取 - 三级优雅终止：关 stdin → SIGTERM 宽限 → SIGKILL 兜底 - 安全边界：运行上限 16/条目上限 32/单次写入 64KB；background_ processes 配置节双闸；proc_name 规整规避 execute_tool 首参撞名 - 测试：test_process_manager.py 39 项（生命周期/游标续读/环形淘汰/ watermark 对齐/EOF 优雅终止/log tee/工具分发），全量 532 passed / 4 failed（平台环境项，与基线一致）零回归 --- aibox/README.md | 42 +- aibox/aibox.py | 11 +- aibox/common/constants.py | 25 + aibox/subsystems/__init__.py | 8 + aibox/subsystems/process_manager.py | 731 ++++++++++++++++++++++++++++ aibox/tests/test_process_manager.py | 448 +++++++++++++++++ aibox/tools/__init__.py | 13 + aibox/tools/process_tools.py | 408 ++++++++++++++++ 8 files changed, 1678 insertions(+), 8 deletions(-) create mode 100644 aibox/subsystems/process_manager.py create mode 100644 aibox/tests/test_process_manager.py create mode 100644 aibox/tools/process_tools.py diff --git a/aibox/README.md b/aibox/README.md index 3b97766..92738fd 100644 --- a/aibox/README.md +++ b/aibox/README.md @@ -96,8 +96,9 @@ aibox/ │ ├── db.py # MemoryDatabase + MemoDatabase（SQLite，FTS5 trigram 中文检索 + LIKE 兜底） │ ├── skill_manager.py # SkillManager（技能发现/加载/热重载） │ ├── conversation_history.py # ConversationHistory（对话历史管理） -│ └── conversation_compressor.py # ConversationCompressor（对话压缩，注入 llm_client 解耦） -├── tools/ # 工具系统包（A 批 + B 批共 17 个工具类） +│ ├── conversation_compressor.py # ConversationCompressor（对话压缩，注入 llm_client 解耦） +│ └── process_manager.py # ProcessManager（后台进程管理，全局单例 + 环形输出缓冲 + atexit 收尾） +├── tools/ # 工具系统包（A 批 + B 批 + 后台进程共 22 个工具类） │ ├── base.py # Tool 抽象基类 │ ├── registry.py # ToolRegistry 工具注册器 │ ├── basic_tools.py # A 批：ReadFile/WriteFile/ListFiles/SimpleWebCrawler/GetCurrentTime/Calculator @@ -106,8 +107,9 @@ aibox/ │ ├── db_tools.py # B 批：MemoryTool/MemoTool │ ├── prompt_tools.py # B 批：UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation │ ├── proxy_tools.py # B 批：SetProxy/GetProxy -│ └── misc_tools.py # B 批：AskUserTool -├── tests/ # 单元测试套件（490 项全过） +│ ├── misc_tools.py # B 批：AskUserTool +│ └── process_tools.py # 后台进程：BgRun/BgWrite/BgList/BgOutput/BgKill（bg_run/bg_write/bg_list/bg_output/bg_kill） +├── tests/ # 单元测试套件（529 项全过） │ ├── conftest.py # sys.path 配置 + websockets 桩 │ ├── test_basic_tools.py # A 批工具 43 项 │ ├── test_b_tools.py # B 批工具 58 项 @@ -126,7 +128,8 @@ aibox/ │ ├── test_i_fixes.py # 遗留收尾 Batch I（16 项） │ ├── test_subagent_max_turns.py # 子代理最大轮次行为回归（14 项） │ ├── test_k_fixes.py # 第五轮审计 Batch K（35 项） -│ └── test_model_config.py # 模型接入点管理工具（17 项） +│ ├── test_model_config.py # 模型接入点管理工具（17 项） +│ └── test_process_manager.py # 后台进程管理器与 bg_* 工具（39 项：生命周期/交互 I/O/收尾） ├── skills/ # 内置扩展技能（*.py 可执行脚本 / *.md Markdown 技能） │ ├── hello-world.py │ ├── aes-crypto.py @@ -161,8 +164,8 @@ common ← subsystems ← tools ← aibox.py ```bash # 从仓库根目录运行 py -m pytest aibox/tests -q -# 当前：490 项全过（Linux；test_execute_timeout / test_flat_api_still_reads_key_file 为 Windows 向用例，Linux 预存失败） -# Windows 开发机：493 passed / 4 failed（平台环境项：symlink 权限、python3 缺失、POSIX 可执行位语义 ×2）/ 1 skipped +# 当前：529 项全过（Linux；test_execute_timeout / test_flat_api_still_reads_key_file 为 Windows 向用例，Linux 预存失败） +# Windows 开发机：532 passed / 4 failed（平台环境项：symlink 权限、python3 缺失、POSIX 可执行位语义 ×2）/ 1 skipped ``` 运行期数据存放于 `~/.aibox/`： @@ -238,6 +241,13 @@ py -m pytest aibox/tests -q | `web` | 访问网页、获取标题/文本/链接（返回 markdown 指引，改用 `curl` 命令行访问） | | `calculator` | 数学计算 | | `get_current_time` | 获取当前时间 | +| `bg_run` | 后台启动长驻进程（dev server / 训练任务 / watch），立即返回句柄不阻塞；stdin 为管道可交互；默认 `detach=false`（aibox 退出时终止），`detach=true` 豁免但脱离句柄管理；`log_file=true` 同步镜像全程输出到 `workspace/bg_<句柄>.log`（无界全档，可被 run_command 的 tail/cat 原生读取）；名称参数为 `proc_name`（≤64 字符，防与 `execute_tool` 首参撞名） | +| `bg_write` | 向后台进程 stdin 写入一行（行协议交互：REPL / 问答式 CLI）；返回 **watermark 水位**——`bg_output(since=watermark)` 得到的必然且仅是本条输入引发的响应（写读对齐，不含 `[stdin]` 回显行）；`wait_lines=N` 同步等待 N 行响应（expect 式一步问答）；单次上限 64KB，超时写入不阻塞工具线程；无 TTY，全屏程序（vim/top）不可用，建议启动时强制行缓冲（`python -i -u` / `PYTHONUNBUFFERED=1` / `stdbuf -oL`） | +| `bg_list` | 列出全部后台进程（运行中在前，已退出保留尾部输出供验尸，LRU 淘汰） | +| `bg_output` | 拉取输出：`tail=N` 尾部 N 行 / `since=游标` 增量续读（游标为上次返回的 `next_cursor`）；单次默认 64KB、上限 1MB；环形缓冲（默认 256KB/进程）淘汰的早期行如实提示 | +| `bg_kill` | 终止后台进程（整棵进程组）：三级优雅终止——关 stdin（行协议进程见 EOF 常自行退出）→ SIGTERM 宽限 → SIGKILL 兜底；`force=true` 直接 SIGKILL；对已退出进程幂等 | + +> **后台进程语义（v3.8）**：注册表为全局单例（跨会话/跨子代理可见，任何 worker 可管理其他 worker 启动的进程）；用户中断（Ctrl+C/取消事件）**不**影响后台进程；aibox 进程退出经 atexit 统一收尾（非 detach 全杀，SIGKILL/SIGTERM 硬杀不覆盖）；单行输出截断 16KB；运行中上限 16 / 总条目上限 32 / 单次 stdin 写入上限 64KB（`background_processes` 配置节可调）；POSIX 按进程组终止（`start_new_session`，孙进程不悬挂），Windows 回退单进程终止。 ## 💬 WebSocket API @@ -615,6 +625,14 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`，通过 `update_i "commands": { "enabled": true }, + "background_processes": { + "enabled": true, + "max_processes": 16, + "max_entries": 32, + "buffer_per_process": 262144, + "max_lifetime": 0, + "max_write_bytes": 65536 + }, "web": { "timeout": 10, "delay": 0, @@ -914,6 +932,16 @@ elif command == "/mycommand": ## 📄 更新日志 +### v3.8 +- ✅ **后台进程池（智能体运行/管理长驻进程）**：新增 `subsystems/process_manager.py`（`ProcessManager` 全局单例）+ `tools/process_tools.py` 5 个 worker 执行面工具（`bg_run` / `bg_write` / `bg_list` / `bg_output` / `bg_kill`），弥补 `run_command` 同步阻塞、超时即杀的不足——dev server / 训练任务 / watch 等长驻进程可后台启动、跨会话管理： + - **生命周期语义（三决策）**：用户中断（Ctrl+C/取消事件）不影响后台进程；aibox 退出经 atexit 统一收尾（非 detach 全杀，`detach=true` 豁免但脱离句柄管理）；注册表跨会话/跨子代理共享（任何 worker 可管理其他 worker 启动的进程） + - **交互式 I/O（stdin 管道 + watermark 写读对齐）**：stdin 挂管道，`bg_write` 向行协议进程（REPL / 问答式 CLI）写入一行，返回 **watermark**（`[stdin]` 回显行序号）——`bg_output(since=watermark)` 得到的必然且仅是该输入引发的响应（写读配对）；`wait_lines=N` expect 式同步等待；回显行入环形缓冲与 log 镜像（tail 可见完整交互时序）；写入 watchdog 线程兜底（进程不读 stdin 管道写满时不阻塞工具线程）；单次写入上限 64KB + - **输出缓冲**：每进程有界环形缓冲（默认 256KB，`buffer_per_process` 可调），单行截断 16KB；`bg_output` 支持 `tail` 尾读与 `since` 游标增量续读（淘汰行如实提示），单次默认 64KB、上限 1MB；`bg_run log_file=true` 输出同步 tee 到 `workspace/bg_<句柄>.log`（无界全档，`[stdin]` 回显同入档，可被 run_command 的 tail/cat 原生读取） + - **三级优雅终止**：`bg_kill` 关 stdin（行协议进程见 EOF 常自行退出）→ SIGTERM 宽限（grace 比例 4:6）→ SIGKILL 兜底；`force=true` 直接 SIGKILL；shutdown 收尾同样先关 stdin + - **安全边界**：运行中上限 16 / 总条目上限 32（LRU 淘汰已退出条目）；`background_processes` 配置节（含 `enabled` 双闸：`commands.enabled` 关闭时一并拒绝）；`proc_name` 标签 ≤64 字符且拒绝控制字符（M5 惯例）；POSIX `start_new_session` + 进程组终止（对齐执行器 L9 语义），Windows 回退单进程终止；TTL `max_lifetime` 惰性清理（默认 0 不启用）；凭据类交互（sudo -S 喂密码）会留对话记录，敏感凭据走环境变量注入（工具描述已指引） + - **实现要点**：H2 reader 线程模式复用（EOF 即 poll 收割防僵尸）；`list_procs` 惰性刷新全表状态；`proc_name` 命名规避 `ToolRegistry.execute_tool(name=...)` 首参撞名；编排者误调拦截表 `WORKER_ONLY_TOOLS` 同步纳入；无 TTY 的适用边界（全屏程序不可用、行缓冲指引）写入工具描述供模型自检 + - **测试**：新增 `tests/test_process_manager.py` 39 项（生命周期/游标续读/环形淘汰/单行截断/上限/TTL/收尾/工具分发/禁用闸/watermark 对齐/回显/sync 等待/超时/EOF 优雅终止/log tee/注册表分发），Windows 开发机 532 passed / 4 failed（平台环境项，与代码无关） + ### v3.7 - ✅ **第六轮安全审计 + Batch L 修复（2026-08-20，交叉验证轮：Batch K 回归验证 / 并发与生命周期 / 降级路径 / 修复间相互作用，4 路并行审读 + 主审逐条复核）**（完整记录见 [docs/reports/aibox_audit_20260817.md](../docs/reports/aibox_audit_20260817.md) §16）： - **Batch K 回归验证**：J1~J16 逐项在位、零回归，无修复引入的缺陷；历届 Batch E~I 修复抽查（executors reader 收尾、1MB 输出上限、registry 流式预算、broadcast 快照等）均在位 diff --git a/aibox/aibox.py b/aibox/aibox.py index d423ab7..d04135f 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -55,6 +55,8 @@ from tools import ( UpdateIdentityTool, UpdateSoulTool, SaveMemoryAndEndConversationTool, SetProxyTool, GetProxyTool, AskUserTool, + # 后台进程管理工具（bg_run/bg_write/bg_list/bg_output/bg_kill，worker 执行面） + BgRunTool, BgWriteTool, BgListTool, BgOutputTool, BgKillTool, ) # 子系统（Wave 1 + Wave 2 抽离）：命令执行器 / 提示词管理器 / 数据库 / 技能管理器 / 对话历史 / 对话压缩 @@ -987,7 +989,8 @@ TOOL_PROFILE_ORCHESTRATOR = "orchestrator" TOOL_PROFILE_WORKER = "worker" # 仅 worker 子代理注册的执行面工具；编排者误调时拦截并纠正 -WORKER_ONLY_TOOLS = {"run_command", "read_file", "write_file", "list_files", "web", "calculator"} +WORKER_ONLY_TOOLS = {"run_command", "read_file", "write_file", "list_files", "web", "calculator", + "bg_run", "bg_write", "bg_list", "bg_output", "bg_kill"} # 每轮临时注入的互动风格提醒（仅编排者，不进对话历史） ORCHESTRATOR_TURN_GUIDANCE = ( @@ -1179,6 +1182,12 @@ class DeepSeekChat: SimpleWebCrawlerTool(self.logger), CalculatorTool(self.logger), GetCurrentTimeTool(self.logger), + # 后台进程管理（共享全局单例：跨子代理可管理彼此启动的进程） + BgRunTool(self.logger), + BgWriteTool(self.logger), + BgListTool(self.logger), + BgOutputTool(self.logger), + BgKillTool(self.logger), ] self.tool_registry.register_many(worker_tools, is_builtin=True) self.logger.info(f"worker 子代理已注册 {len(worker_tools)} 个执行面工具") diff --git a/aibox/common/constants.py b/aibox/common/constants.py index 351b701..71af689 100644 --- a/aibox/common/constants.py +++ b/aibox/common/constants.py @@ -236,6 +236,31 @@ class Constants: def COMPRESSOR_LLM_MAX_TOKENS(cls): return cls._get_config().get("compressor.llm_max_tokens", 1000) + # 后台进程管理（bg_run/bg_list/bg_output/bg_kill） + @property + def BG_PROCESSES_ENABLED(cls): + return cls._get_config().get("background_processes.enabled", True) + + @property + def BG_MAX_PROCESSES(cls): + return cls._get_config().get("background_processes.max_processes", 16) + + @property + def BG_MAX_ENTRIES(cls): + return cls._get_config().get("background_processes.max_entries", 32) + + @property + def BG_BUFFER_PER_PROCESS(cls): + return cls._get_config().get("background_processes.buffer_per_process", 262144) + + @property + def BG_MAX_LIFETIME(cls): + return cls._get_config().get("background_processes.max_lifetime", 0) + + @property + def BG_MAX_WRITE_BYTES(cls): + return cls._get_config().get("background_processes.max_write_bytes", 65536) + # 创建常量实例 constants = Constants() diff --git a/aibox/subsystems/__init__.py b/aibox/subsystems/__init__.py index 70a8293..b1cabc0 100644 --- a/aibox/subsystems/__init__.py +++ b/aibox/subsystems/__init__.py @@ -24,6 +24,11 @@ from .db import MemoryDatabase, MemoDatabase from .skill_manager import SkillManager from .conversation_history import ConversationHistory from .conversation_compressor import ConversationCompressor +from .process_manager import ( + ProcessManager, + get_process_manager, + reset_process_manager, +) __all__ = [ "CommandExecutor", @@ -34,4 +39,7 @@ __all__ = [ "SkillManager", "ConversationHistory", "ConversationCompressor", + "ProcessManager", + "get_process_manager", + "reset_process_manager", ] diff --git a/aibox/subsystems/process_manager.py b/aibox/subsystems/process_manager.py new file mode 100644 index 0000000..5910944 --- /dev/null +++ b/aibox/subsystems/process_manager.py @@ -0,0 +1,731 @@ +#!/usr/bin/env python3 +""" +subsystems.process_manager - 后台进程管理器 + +让智能体能够启动并管理长驻后台进程（dev server / 训练任务 / watch 等）， +弥补 run_command 同步阻塞、超时即杀的不足： + - bg 语义：start() 立即返回句柄，进程独立于对话轮次存活 + - 有界环形输出缓冲：每进程上限 buffer_per_process（默认 256KB）， + 解决 nohup+日志文件方案输出无界增长的问题 + - 注册表：自增句柄跨会话/跨子代理可见（全局单例） + - 生命周期：惰性状态收割（reader 线程 EOF 即 poll 收割防僵尸）+ + TTL 惰性清理 + 退出统一收尾（非 detach 进程默认全杀） + +安全边界（对齐 SubprocessCommandExecutor 既有惯例）： + - 输出单行截断 16KB，防超长单行撑爆内存 + - 运行中进程数上限（默认 16）+ 总条目上限（默认 32，已退出条目 + 保留尾部输出供验尸，LRU 淘汰） + - POSIX start_new_session + 进程组信号（L9 语义：孙进程不悬挂）； + Windows 无进程组概念，回退单进程 terminate/kill + - 中断语义：用户 Ctrl+C / 取消事件不影响后台进程（后台的本意就是 + 脱离本轮对话生命周期）；aibox 进程退出时经 atexit 统一收尾， + detach=true 的进程豁免 + +依赖：common (ConfigManager, constants, Logger) + 标准库 +不反向依赖 aibox 任何符号。 +""" + +import atexit +import collections +import os +import signal +import subprocess +import threading +import time +from typing import Dict, List, Optional + +from common import ConfigManager, constants, Logger + +# 单行输出最大字节数，超出截断（防超长单行撑爆环形缓冲内存预算） +MAX_LINE_BYTES = 16 * 1024 + +# 单次 bg_output 拉取的硬上限（对齐执行器 1MB 输出限制） +MAX_FETCH_BYTES = 1 * 1024 * 1024 + + +class _ProcEntry: + """单个后台进程的注册表条目（内部类）""" + + __slots__ = ( + "handle", "name", "command", "popen", "pgid", "started_at", + "detach", "max_lifetime", "owner", "status", "exit_code", "end_time", + "lines", "buffer_bytes", "max_buffer", "next_seq", "evicted", + "lock", "reader_done", "reader", "stdin_lock", "log_fh", + ) + + # 状态取值：running / exited（自然退出）/ killed（被管理器终止）/ expired（TTL 到期） + def __init__(self, handle: int, name: str, command: str, popen, pgid, + detach: bool, max_lifetime: float, buffer_bytes: int, + owner: Optional[str] = None): + self.handle = handle + self.name = name + self.command = command + self.popen = popen + self.pgid = pgid + self.started_at = time.time() + self.detach = detach + self.max_lifetime = max_lifetime + self.owner = owner + self.status = "running" + self.exit_code = None + self.end_time = None + # 环形输出缓冲：deque of (seq, line, nbytes)，seq 单调递增不回收， + # 淘汰最旧行仍保持 seq 连续性——bg_output 的 since 游标据此续读 + self.lines = collections.deque() + self.buffer_bytes = 0 + self.max_buffer = max(1024, int(buffer_bytes)) + self.next_seq = 1 + self.evicted = 0 + self.lock = threading.Lock() + self.reader_done = threading.Event() + self.reader = None + # 交互写入：单写锁保序（bg_write 串行化，防交错输入） + self.stdin_lock = threading.Lock() + # 可选全档镜像文件句柄（log_file=True 时由 reader 线程 tee 写入） + self.log_fh = None + + +class ProcessManager: + """后台进程管理器：启动 / 列举 / 增量拉取输出 / 终止 / 退出收尾""" + + def __init__(self, logger: Optional[Logger] = None, + max_processes: int = None, max_entries: int = None, + buffer_per_process: int = None, max_lifetime: float = None): + self.logger = logger or Logger("ProcessManager") + self.config = ConfigManager() + self.max_processes = int(max_processes if max_processes is not None + else constants.BG_MAX_PROCESSES) + self.max_entries = int(max_entries if max_entries is not None + else constants.BG_MAX_ENTRIES) + self.buffer_per_process = int(buffer_per_process if buffer_per_process is not None + else constants.BG_BUFFER_PER_PROCESS) + self.max_lifetime = float(max_lifetime if max_lifetime is not None + else constants.BG_MAX_LIFETIME) + self.max_write_bytes = int(constants.BG_MAX_WRITE_BYTES) + self._procs: Dict[int, _ProcEntry] = {} + self._next_id = 1 + self._lock = threading.Lock() + self._shutdown_done = False + + # ------------------------------------------------------------------ + # 启动 + # ------------------------------------------------------------------ + + def start(self, command: str, name: str = None, work_dir: str = None, + env: Dict = None, detach: bool = False, + log_file: bool = False, owner: str = None) -> Dict: + """后台启动命令，立即返回。成功返回 {ok, handle, pid, name, log_file}， + 失败返回 {ok: False, error}。 + + stdin 为管道：后续可经 write() 交互写入（行协议 REPL 问答）。 + log_file=True 时输出同步 tee 到 workspace/bg_<句柄>.log（无界 + 全档，可被原生工具 tail/cat 读取；环形缓冲仍为 bg_output 视图）。""" + if not command or not isinstance(command, str) or not command.strip(): + return {"ok": False, "error": "缺少或空的 command"} + + # 名称规整（M5 惯例）：标签仅用于展示回显，拒绝控制字符防终端/前端注入 + if name is not None: + name = str(name).strip() + if not name.isprintable() or len(name) > 64: + return {"ok": False, "error": "name 含控制字符或超过 64 字符"} + if not name: + name = None + if name is None: + name = command.strip()[:40] + + if work_dir is None or work_dir == ".": + work_dir = constants.WORKSPACE_DIR + + with self._lock: + if self._shutdown_done: + return {"ok": False, "error": "管理器已关闭"} + # TTL 惰性清理 + 状态收割 + self._sweep_locked() + running = [e for e in self._procs.values() if e.status == "running"] + if len(running) >= self.max_processes: + return {"ok": False, + "error": f"运行中后台进程已达上限（{self.max_processes}），" + f"请先用 bg_kill 清理"} + # 总条目上限：淘汰最老的已退出条目（保留其信息的价值最低） + if len(self._procs) >= self.max_entries: + exited = sorted( + (e for e in self._procs.values() if e.status != "running"), + key=lambda e: e.started_at) + while self._procs and len(self._procs) >= self.max_entries and exited: + victim = exited.pop(0) + self._procs.pop(victim.handle, None) + if len(self._procs) >= self.max_entries: + return {"ok": False, + "error": f"后台进程条目已达上限（{self.max_entries}）且全部运行中"} + handle = self._next_id + self._next_id += 1 + + exec_env = os.environ.copy() + if env: + exec_env.update(env) + + try: + popen = subprocess.Popen( + command, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + # 交互式 I/O：stdin 挂管道供 write() 写入。与 H3 的差异： + # 后台进程由管理器显式管理生命周期，不存在"等待输入的 + # 命令永久挂起无人知晓"问题——不读 stdin 的进程不受影响， + # 读 stdin 的进程获得交互能力，kill 路径三级优雅终止兜底。 + stdin=subprocess.PIPE, + text=True, + bufsize=1, + universal_newlines=True, + env=exec_env, + cwd=os.path.expanduser(work_dir), + # 对齐 L9：POSIX 下 shell 成为会话首进程，kill 按进程组 + # 一次覆盖整棵进程树（含 & 后台孙进程） + start_new_session=(os.name == "posix"), + ) + except FileNotFoundError as e: + return {"ok": False, "error": f"命令未找到: {e}"} + except PermissionError as e: + return {"ok": False, "error": f"权限不足: {e}"} + except Exception as e: + return {"ok": False, "error": f"启动失败: {e}"} + + pgid = None + if os.name == "posix": + try: + pgid = os.getpgid(popen.pid) + except Exception: + pgid = None + + entry = _ProcEntry( + handle, name, command.strip(), popen, pgid, + detach=bool(detach), max_lifetime=self.max_lifetime, + buffer_bytes=self.buffer_per_process, owner=owner, + ) + # 可选全档镜像（失败降级为仅环形缓冲，不阻断启动） + log_path = None + if log_file: + try: + log_dir = constants.WORKSPACE_DIR + os.makedirs(log_dir, exist_ok=True) + log_path = os.path.join(log_dir, f"bg_{handle}.log") + entry.log_fh = open(log_path, "a", encoding="utf-8", + errors="replace", buffering=1) + except Exception as e: + entry.log_fh = None + log_path = None + self.logger.warning(f"输出镜像文件开启失败，仅保留环形缓冲: {e}") + with self._lock: + # start() 竞态兜底：期间并发 shutdown 则立即回收刚拉起的进程 + if self._shutdown_done: + self._close_stdin(entry) + self._terminate_tree(entry, force=True) + try: + popen.wait(timeout=2) + except Exception: + pass + return {"ok": False, "error": "管理器已关闭"} + self._procs[handle] = entry + + entry.reader = threading.Thread( + target=self._reader, args=(entry,), daemon=True, + name=f"bg-proc-{handle}-reader") + entry.reader.start() + self.logger.info(f"后台进程已启动: #{handle} pid={popen.pid} " + f"detach={entry.detach} cmd={entry.command[:60]}") + return {"ok": True, "handle": handle, "pid": popen.pid, "name": name, + "detach": entry.detach, "log_file": log_path} + + # ------------------------------------------------------------------ + # 输出读取 + # ------------------------------------------------------------------ + + def get_output(self, handle, tail: int = None, since: int = None, + max_bytes: int = 65536) -> Dict: + """拉取后台进程输出。 + + - tail=N：取最近 N 行（缺省 50） + - since=S：取行序号 > S 的全部缓冲行（增量续读，游标为上次 + 返回的 next_cursor；被环形淘汰的早期行以 evicted 提示） + - max_bytes：本次拉取字节上限（缺省 64KB，硬顶 1MB） + """ + entry = self._lookup(handle) + if entry is None: + return {"ok": False, "error": f"句柄 #{handle} 不存在"} + + try: + max_bytes = int(max_bytes or 65536) + except (TypeError, ValueError): + max_bytes = 65536 + max_bytes = max(1024, min(max_bytes, MAX_FETCH_BYTES)) + + with entry.lock: + snapshot = list(entry.lines) + evicted = entry.evicted + next_seq = entry.next_seq + + selected = [] + evicted_note = 0 + if since is not None: + try: + since = int(since) + except (TypeError, ValueError): + return {"ok": False, "error": "since 游标必须是整数"} + selected = [(s, l) for (s, l, _) in snapshot if s > since] + # 请求范围早于环形缓冲起点：被淘汰部分如实告知 + oldest = snapshot[0][0] if snapshot else next_seq + if since + 1 < oldest: + evicted_note = oldest - since - 1 + else: + try: + tail = int(tail) if tail is not None else 50 + except (TypeError, ValueError): + tail = 50 + tail = max(1, min(tail, 100000)) + selected = [(s, l) for (s, l, _) in snapshot[-tail:]] + + # 字节上限截断（保留最新行优先：从尾部回溯收集） + kept = [] + total = 0 + truncated = False + for s, line in reversed(selected): + nb = len(line.encode("utf-8", errors="replace")) + if total + nb > max_bytes: + truncated = True + break + kept.append((s, line)) + total += nb + kept.reverse() + + output = "".join(line for _, line in kept) + result = { + "ok": True, + "handle": entry.handle, + "name": entry.name, + "status": entry.status, + "exit_code": entry.exit_code, + "output": output, + "lines": len(kept), + "next_cursor": kept[-1][0] if kept else max(since or 0, next_seq - 1), + "truncated": truncated, + } + if evicted_note > 0 or (since is None and evicted > 0): + result["evicted"] = evicted_note if since is not None else evicted + return result + + # ------------------------------------------------------------------ + # 交互写入（stdin 管道 + watermark 对齐） + # ------------------------------------------------------------------ + + def write(self, handle, text: str, append_newline: bool = True, + wait_lines: int = 0, wait_timeout: float = 5.0, + write_timeout: float = 5.0) -> Dict: + """向后台进程 stdin 写入一行（交互式问答）。 + + 返回 {ok, handle, watermark, sent_bytes[, wait]}： + - watermark：本条输入的 [stdin] 回显行序号；bg_output(since= + watermark) 得到的必然且仅是本条输入引发的响应（写读对齐） + - wait_lines>0 时同步等待 N 行响应（wait 子 dict 含 output/ + timed_out），expect 式一步问答 + 失败返回 {ok: False, error}（进程已退出/写入超时/超长等）。""" + entry = self._lookup(handle) + if entry is None: + return {"ok": False, "error": f"句柄 #{handle} 不存在"} + if not isinstance(text, str) or text == "": + return {"ok": False, "error": "缺少或空的 text"} + data = text if (not append_newline or text.endswith("\n")) else text + "\n" + payload = data.encode("utf-8") + if len(payload) > self.max_write_bytes: + return {"ok": False, + "error": f"单次写入超过 {self.max_write_bytes} 字节上限（超长内容建议落文件后让进程从文件读）"} + + with self._lock: + self._update_status_locked(entry) + if entry.status != "running": + return {"ok": False, "error": f"进程已结束（{entry.status}），无法写入"} + + # watermark：回显行将占用的序号（写入前先记账，响应必然在其后） + with entry.lock: + watermark = entry.next_seq + # 回显行进环形缓冲 + log 镜像（多行输入折叠为单行展示） + echo = "[stdin] " + text.replace("\n", "⏎") + self._record_line(entry, echo) + + # 写入放 watchdog 线程：进程不读 stdin 且管道缓冲（~64KB）写满时 + # write() 会阻塞，不能拖死工具线程；超时后 daemon 线程由系统兜底 + error_box = {} + done = threading.Event() + + def _do_write(): + try: + with entry.stdin_lock: + stdin = entry.popen.stdin + if stdin is None or stdin.closed: + raise BrokenPipeError("stdin 已关闭") + stdin.write(data) + stdin.flush() + except Exception as e: + error_box["e"] = e + finally: + done.set() + + t = threading.Thread(target=_do_write, daemon=True, + name=f"bg-proc-{entry.handle}-writer") + t.start() + try: + write_timeout = max(0.1, min(60.0, float(write_timeout))) + except (TypeError, ValueError): + write_timeout = 5.0 + if not done.wait(timeout=write_timeout): + return {"ok": False, + "error": f"写入超时（>{write_timeout}s）：进程可能不读取 stdin，管道缓冲已满"} + if error_box: + with self._lock: + self._update_status_locked(entry) + return {"ok": False, + "error": f"写入失败: {error_box['e']}"} + + result = {"ok": True, "handle": entry.handle, "watermark": watermark, + "sent_bytes": len(payload)} + if wait_lines and int(wait_lines) > 0: + result["wait"] = self._wait_lines(entry, watermark, + int(wait_lines), wait_timeout) + return result + + def _wait_lines(self, entry: _ProcEntry, watermark: int, count: int, + timeout: float) -> Dict: + """等待 watermark 之后出现 count 行新输出（进程退出且 reader 排空 + 即提前返回，如实标记 timed_out）。""" + try: + timeout = max(0.1, min(60.0, float(timeout))) + except (TypeError, ValueError): + timeout = 5.0 + deadline = time.time() + timeout + + def _snap(): + with entry.lock: + return [(s, l) for (s, l, _) in entry.lines if s > watermark] + + lines = _snap() + timed_out = False + while len(lines) < count: + if entry.popen.poll() is not None and entry.reader_done.is_set(): + time.sleep(0.1) # 排空 reader 尾部在途行后取终值 + lines = _snap() + break + if time.time() > deadline: + timed_out = True + break + time.sleep(0.05) + lines = _snap() + + # 字节上限（复用单次拉取硬顶） + kept, total, truncated = [], 0, False + for s, line in lines: + nb = len(line.encode("utf-8", errors="replace")) + if total + nb > MAX_FETCH_BYTES: + truncated = True + break + kept.append((s, line)) + total += nb + out = { + "lines": len(kept), + "output": "".join(l for _, l in kept), + "timed_out": timed_out, + "next_cursor": kept[-1][0] if kept else watermark, + } + if truncated: + out["truncated"] = True + return out + + def _close_stdin(self, entry: _ProcEntry): + """关闭 stdin 管道（行协议进程见 EOF 常自行退出；幂等）""" + try: + with entry.stdin_lock: + stdin = entry.popen.stdin + if stdin is not None and not stdin.closed: + stdin.close() + except Exception: + pass + + # ------------------------------------------------------------------ + # 列举 / 终止 / 收尾 + # ------------------------------------------------------------------ + + def list_procs(self) -> List[Dict]: + """列出全部条目（运行中在前，其余按句柄序）""" + with self._lock: + self._sweep_locked() + # 惰性收割：列举即刷新全部条目状态（此前仅 kill/get_output 路径更新） + for entry in self._procs.values(): + self._update_status_locked(entry) + entries = sorted(self._procs.values(), + key=lambda e: (e.status != "running", e.handle)) + return [self._entry_info(e) for e in entries] + + def kill(self, handle, force: bool = False, grace: float = 3.0) -> Dict: + """终止后台进程。SIGTERM → 宽限轮询 → SIGKILL；force=True 直接 SIGKILL。 + 对已退出条目幂等，返回当前状态。""" + entry = self._lookup(handle) + if entry is None: + return {"ok": False, "error": f"句柄 #{handle} 不存在"} + + with self._lock: + self._update_status_locked(entry) + if entry.status != "running": + return {"ok": True, "handle": entry.handle, "status": entry.status, + "exit_code": entry.exit_code, "already_finished": True} + + try: + grace = min(30.0, max(0.1, float(grace))) + except (TypeError, ValueError): + grace = 3.0 + + if force: + self._terminate_tree(entry, force=True) + else: + # 三级优雅终止：关 stdin（行协议进程见 EOF 常自行退出）→ + # SIGTERM 宽限 → SIGKILL 兜底 + self._close_stdin(entry) + deadline = time.time() + grace * 0.4 + while time.time() < deadline and entry.popen.poll() is None: + time.sleep(0.05) + if entry.popen.poll() is None: + self._terminate_tree(entry) + deadline = time.time() + grace * 0.6 + while time.time() < deadline and entry.popen.poll() is None: + time.sleep(0.05) + if entry.popen.poll() is None: + self._terminate_tree(entry, force=True) + try: + entry.popen.wait(timeout=2) + except Exception: + pass + + with self._lock: + self._update_status_locked(entry) + if entry.status == "running": + # 进程仍未死（极端：不可杀的 D 状态），如实报告 + return {"ok": False, "handle": entry.handle, + "error": "进程未能终止（可能处于不可中断状态）"} + entry.status = "killed" + return {"ok": True, "handle": entry.handle, "status": entry.status, + "exit_code": entry.exit_code} + + def shutdown(self) -> Dict: + """退出收尾：终止全部非 detach 运行中进程（SIGTERM→SIGKILL）， + 清空注册表；detach 进程保留运行、仅注销登记。幂等。""" + with self._lock: + if self._shutdown_done: + return {"killed": 0, "detached": 0} + self._shutdown_done = True + entries = list(self._procs.values()) + self._procs.clear() + + killed = detached = 0 + for entry in entries: + self._update_status(entry) + if entry.status != "running": + continue + if entry.detach: + detached += 1 + self.logger.info(f"后台进程 #{entry.handle}（detach）保留运行: " + f"pid={entry.popen.pid}") + continue + # 优雅收尾：先关 stdin（行协议进程见 EOF 常自行退出）再走信号 + self._close_stdin(entry) + try: + entry.popen.wait(timeout=1) + except Exception: + pass + if entry.popen.poll() is None: + self._terminate_tree(entry) + try: + entry.popen.wait(timeout=3) + except Exception: + self._terminate_tree(entry, force=True) + try: + entry.popen.wait(timeout=2) + except Exception: + pass + entry.status = "killed" + killed += 1 + if killed or detached: + self.logger.info(f"后台进程收尾: 终止 {killed} 个, detach 保留 {detached} 个") + return {"killed": killed, "detached": detached} + + # ------------------------------------------------------------------ + # 内部实现 + # ------------------------------------------------------------------ + + def _lookup(self, handle) -> Optional[_ProcEntry]: + try: + handle = int(handle) + except (TypeError, ValueError): + return None + with self._lock: + entry = self._procs.get(handle) + if entry is not None: + self._update_status(entry) + return entry + + def _update_status(self, entry: _ProcEntry): + with self._lock: + self._update_status_locked(entry) + + def _update_status_locked(self, entry: _ProcEntry): + """惰性收割：poll() 即 waitpid(WNOHANG)，自然退出时记录状态。""" + if entry.status != "running": + return + rc = entry.popen.poll() + if rc is not None: + entry.exit_code = rc + entry.end_time = time.time() + entry.status = "exited" + + def _sweep_locked(self): + """TTL 惰性清理：超期运行中进程标记 expired 并终止（max_lifetime<=0 不启用）。 + 调用方须持有 self._lock。""" + if self.max_lifetime <= 0: + return + now = time.time() + for entry in list(self._procs.values()): + if entry.status == "running" and now - entry.started_at > self.max_lifetime: + self._terminate_tree(entry) + entry.status = "expired" + entry.end_time = now + entry.exit_code = entry.popen.poll() + self.logger.info(f"后台进程 #{entry.handle} TTL 到期已终止") + + def _terminate_tree(self, entry: _ProcEntry, force: bool = False): + """按进程组终止整棵进程树（对齐 executors L9 语义与 Windows 回退）""" + proc = entry.popen + try: + if os.name == "posix": + sig = signal.SIGKILL if force else signal.SIGTERM + pgid = entry.pgid + if pgid is None: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, sig) + elif force: + proc.kill() + else: + proc.terminate() + except Exception: + try: + proc.kill() + except Exception: + pass + + def _reader(self, entry: _ProcEntry): + """后台 reader 线程（对齐 executors H2 模式）：阻塞 readline 入环形 + 缓冲（+ 可选 log 镜像）；EOF/异常即 poll() 收割防僵尸，并显式关闭 + 管道与镜像文件句柄。""" + try: + for line in iter(entry.popen.stdout.readline, ""): + self._record_line(entry, line) + except Exception: + pass + finally: + entry.reader_done.set() + try: + entry.popen.poll() # 立即收割退出子进程，防僵尸悬挂 + except Exception: + pass + try: + entry.popen.stdout.close() + except Exception: + pass + if entry.log_fh is not None: + try: + entry.log_fh.flush() + entry.log_fh.close() + except Exception: + pass + + def _record_line(self, entry: _ProcEntry, line: str): + """统一记录：环形缓冲 + 可选 log 全档镜像（stdin 回显同样入档）""" + if entry.log_fh is not None: + try: + entry.log_fh.write(line) + except Exception: + pass + self._append_line(entry, line) + + @staticmethod + def _append_line(entry: _ProcEntry, line: str): + # 单行截断：环形缓冲内存预算不被超长单行击穿 + nbytes = len(line.encode("utf-8", errors="replace")) + if nbytes > MAX_LINE_BYTES: + allowed = line.encode("utf-8", errors="replace")[:MAX_LINE_BYTES] + line = allowed.decode("utf-8", errors="ignore") + "…[单行截断]\n" + nbytes = len(line.encode("utf-8", errors="replace")) + with entry.lock: + entry.lines.append((entry.next_seq, line, nbytes)) + entry.next_seq += 1 + entry.buffer_bytes += nbytes + while entry.buffer_bytes > entry.max_buffer and len(entry.lines) > 1: + _, old, old_nb = entry.lines.popleft() + entry.buffer_bytes -= old_nb + entry.evicted += 1 + + @staticmethod + def _entry_info(entry: _ProcEntry) -> Dict: + now = time.time() + return { + "handle": entry.handle, + "name": entry.name, + "pid": entry.popen.pid, + "status": entry.status, + "exit_code": entry.exit_code, + "detach": entry.detach, + "started_at": entry.started_at, + "runtime_sec": round((entry.end_time or now) - entry.started_at, 1), + "buffered_lines": len(entry.lines), + "evicted_lines": entry.evicted, + "command": entry.command, + } + + +# ---------------------------------------------------------------------- +# 模块级单例（工具层共享；测试经 reset_process_manager() 重置） +# ---------------------------------------------------------------------- + +_manager: Optional[ProcessManager] = None +_manager_lock = threading.Lock() +_atexit_registered = False + + +def get_process_manager(logger: Optional[Logger] = None) -> ProcessManager: + """获取全局 ProcessManager 单例（懒加载）。 + + 后台进程注册表必须跨会话/跨子代理共享——任何 worker 子代理启动的 + 进程，后续任意子代理都能 bg_list/bg_output/bg_kill 管理。""" + global _manager, _atexit_registered + with _manager_lock: + if _manager is None: + _manager = ProcessManager(logger) + if not _atexit_registered: + # 退出语义：aibox 进程退出（含 --once sys.exit / Ctrl+C）时 + # 非 detach 后台进程统一终止；detach 进程保留运行仅注销登记。 + # SIGKILL/SIGTERM 硬杀无法覆盖（与既有线程池等资源同级）。 + atexit.register(_shutdown_atexit) + _atexit_registered = True + return _manager + + +def reset_process_manager(manager: Optional[ProcessManager] = None) -> None: + """替换/清空全局单例（测试隔离用）。传入 None 仅解除引用，不触发收尾。""" + global _manager + with _manager_lock: + _manager = manager + + +def _shutdown_atexit(): + m = _manager + if m is not None: + try: + m.shutdown() + except Exception: + pass diff --git a/aibox/tests/test_process_manager.py b/aibox/tests/test_process_manager.py new file mode 100644 index 0000000..b1b0b9a --- /dev/null +++ b/aibox/tests/test_process_manager.py @@ -0,0 +1,448 @@ +"""后台进程管理器（subsystems.process_manager）与 bg_* 工具回归测试 + +覆盖三决策语义与安全边界： + - bg 生命周期：启动立即返回 / 惰性收割 / kill 幂等 / TTL 过期 + - 输出缓冲：tail/since 游标续读 / 环形淘汰提示 / 单行截断 / 拉取字节上限 + - 上限：运行中进程数上限 / 总条目 LRU 淘汰 + - 收尾：shutdown 杀非 detach、保留 detach；atexit 单例共享 + - 工具层：参数规整（字符串句柄）/ 双闸禁用 / 注册表分发 + +跨平台：长驻进程用 sys.executable -c "time.sleep(...)"（echo/ping 不可移植）。 +Windows 注意：shell=True 下 terminate 只杀 shell 包装进程（与执行器 L9 +回退语义一致），被杀判定以 popen.poll() 为准。 +""" + +import queue +import sys +import time +import types + +import pytest + +if "websockets" not in sys.modules: + sys.modules["websockets"] = types.ModuleType("websockets") + +from subsystems import ProcessManager, get_process_manager, reset_process_manager # noqa: E402 +from subsystems.process_manager import MAX_LINE_BYTES # noqa: E402 +from tools import ( # noqa: E402 + ToolRegistry, + BgRunTool, BgWriteTool, BgListTool, BgOutputTool, BgKillTool, +) + + +PY = sys.executable + + +@pytest.fixture() +def echo_srv(tmp_path): + """行协议回环服务器：stdin 读一行 → stdout 回 ECHO:行，EOF 退出。 + 交互式 I/O（bg_write/watermark/优雅终止）测试的标准对端。""" + script = ( + "import sys\n" + "while True:\n" + " line = sys.stdin.readline()\n" + " if not line:\n" + " break\n" + " sys.stdout.write('ECHO:' + line.strip() + '\\n')\n" + " sys.stdout.flush()\n" + ) + f = tmp_path / "echo_srv.py" + f.write_text(script, encoding="utf-8") + return f'"{PY}" "{f}"' + + +def _py(code: str) -> str: + """跨平台 python 命令（code 内只用单引号）""" + return f'"{PY}" -c "{code}"' + + +ECHO_10 = _py("for i in range(10): print('line-%d' % i)") +SLEEP_10 = _py("import time; time.sleep(10)") +HELLO = _py("print('hello-bg')") + + +@pytest.fixture() +def mgr(): + """每测试独立管理器（不走单例），收尾兜底清进程""" + m = ProcessManager() + try: + yield m + finally: + m.shutdown() + + +def _wait_status(m: ProcessManager, handle: int, status: str, timeout: float = 10.0): + """轮询等待句柄达到指定状态""" + deadline = time.time() + timeout + while time.time() < deadline: + for p in m.list_procs(): + if p["handle"] == handle and p["status"] == status: + return p + time.sleep(0.1) + return None + + +# ==================== 启动与生命周期 ==================== + +class TestStart: + def test_start_returns_handle_immediately(self, mgr): + t0 = time.time() + r = mgr.start(SLEEP_10, name="sleeper") + assert r["ok"] is True + assert time.time() - t0 < 2 # 立即返回，不等进程结束 + assert isinstance(r["handle"], int) and r["pid"] > 0 + assert r["name"] == "sleeper" + + def test_start_default_name_from_command(self, mgr): + r = mgr.start(HELLO) + assert r["ok"] is True + # 默认名称 = 命令前 40 字符 + assert r["name"] == HELLO.strip()[:40] + + def test_start_rejects_empty_command(self, mgr): + assert mgr.start(" ")["ok"] is False + assert mgr.start(None)["ok"] is False + + def test_start_rejects_bad_name(self, mgr): + # 名称含控制字符（M5 惯例：防终端/前端注入） + assert mgr.start(HELLO, name="bad\x1b[31m")["ok"] is False + assert mgr.start(HELLO, name="x" * 65)["ok"] is False + + def test_start_missing_binary_returns_error(self, mgr): + r = mgr.start("definitely-not-a-command-xyz --version") + # shell 本身能启动（rc=127/9009），故启动成功、退出码非零 + assert r["ok"] is True + p = _wait_status(mgr, r["handle"], "exited") + assert p is not None + assert p["exit_code"] != 0 + + +class TestLifecycle: + def test_exited_status_and_exit_code(self, mgr): + r = mgr.start(HELLO) + p = _wait_status(mgr, r["handle"], "exited") + assert p is not None + assert p["exit_code"] == 0 + + def test_kill_running_process(self, mgr): + r = mgr.start(SLEEP_10) + time.sleep(0.5) # 等进程真正起来 + k = mgr.kill(r["handle"]) + assert k["ok"] is True and k["status"] == "killed" + assert mgr._procs[r["handle"]].popen.poll() is not None + + def test_kill_is_idempotent_on_exited(self, mgr): + r = mgr.start(HELLO) + assert _wait_status(mgr, r["handle"], "exited") is not None + k = mgr.kill(r["handle"]) + assert k["ok"] is True and k["already_finished"] is True + + def test_kill_missing_handle(self, mgr): + assert mgr.kill(999)["ok"] is False + + def test_ttl_expiry(self): + m = ProcessManager(max_lifetime=0.5) + try: + r = m.start(SLEEP_10) + time.sleep(0.8) + # list 触发惰性 sweep + assert _wait_status(m, r["handle"], "expired", timeout=2) is not None + finally: + m.shutdown() + + def test_max_processes_limit(self): + m = ProcessManager(max_processes=2) + try: + assert m.start(SLEEP_10)["ok"] is True + assert m.start(SLEEP_10)["ok"] is True + r3 = m.start(SLEEP_10) + assert r3["ok"] is False and "上限" in r3["error"] + finally: + m.shutdown() + + def test_max_entries_evicts_oldest_exited(self): + m = ProcessManager(max_entries=2) + try: + h1 = m.start(HELLO)["handle"] + h2 = m.start(HELLO)["handle"] + assert _wait_status(m, h1, "exited") is not None + assert _wait_status(m, h2, "exited") is not None + # 第三条触发最老已退出条目（#1）LRU 淘汰 + h3 = m.start(HELLO)["handle"] + assert h3 > 0 + handles = {p["handle"] for p in m.list_procs()} + assert h1 not in handles and h2 in handles and h3 in handles + finally: + m.shutdown() + + +# ==================== 输出缓冲与游标 ==================== + +class TestOutput: + def test_tail_and_since_cursor(self, mgr): + r = mgr.start(ECHO_10) + assert _wait_status(mgr, r["handle"], "exited") is not None + time.sleep(0.3) # reader 排空末尾行 + + out = mgr.get_output(r["handle"], tail=3) + assert out["ok"] is True + assert out["lines"] == 3 + assert "line-9" in out["output"] and "line-6" not in out["output"] + + full = mgr.get_output(r["handle"], since=0) + assert full["lines"] == 10 + assert full["next_cursor"] == 10 + + inc = mgr.get_output(r["handle"], since=full["next_cursor"]) + assert inc["lines"] == 0 and inc["next_cursor"] == 10 + + def test_since_before_eviction_reports_gap(self): + m = ProcessManager(buffer_per_process=2048) + try: + r = m.start(_py("for i in range(50): print('pad-%03d-%s' % (i, 'x'*80))")) + assert _wait_status(m, r["handle"], "exited") is not None + time.sleep(0.3) + out = m.get_output(r["handle"], since=0) + assert out["ok"] is True + assert out["evicted"] > 0 # 早期行被环形淘汰 + assert out["lines"] < 50 + finally: + m.shutdown() + + def test_single_line_truncated(self, mgr): + r = mgr.start(_py("print('A' * (1024*64))")) + assert _wait_status(mgr, r["handle"], "exited") is not None + time.sleep(0.3) + out = mgr.get_output(r["handle"], tail=5) + assert "单行截断" in out["output"] + # 缓冲内存预算不被超长单行击穿 + entry = mgr._procs[r["handle"]] + assert entry.buffer_bytes <= MAX_LINE_BYTES + 1024 + + def test_fetch_byte_limit_truncates(self, mgr): + r = mgr.start(_py("for i in range(100): print('y' * 200)")) + assert _wait_status(mgr, r["handle"], "exited") is not None + time.sleep(0.3) + out = mgr.get_output(r["handle"], since=0, max_bytes=4096) + assert out["truncated"] is True + + def test_missing_handle(self, mgr): + assert mgr.get_output(999)["ok"] is False + + def test_invalid_since(self, mgr): + r = mgr.start(HELLO) + assert mgr.get_output(r["handle"], since="abc")["ok"] is False + + +# ==================== 交互式 I/O（stdin 管道 + watermark 对齐） ==================== + +class TestInteractive: + def test_write_roundtrip_watermark_alignment(self, mgr, echo_srv): + """核心对齐语义：bg_output(since=watermark) = 本条输入的响应且不含回显""" + h = mgr.start(echo_srv)["handle"] + r = mgr.write(h, "hello", wait_lines=1, wait_timeout=10) + assert r["ok"] is True + assert "ECHO:hello" in r["wait"]["output"] + assert r["wait"]["timed_out"] is False + + out = mgr.get_output(h, since=r["watermark"]) + assert "ECHO:hello" in out["output"] + assert "[stdin]" not in out["output"] # 回显行序号 == watermark，不包含 + + def test_echo_line_visible_in_tail(self, mgr, echo_srv): + """[stdin] 回显进环形缓冲，tail 视图可见完整交互时序""" + h = mgr.start(echo_srv)["handle"] + assert mgr.write(h, "abc", wait_lines=1)["ok"] is True + out = mgr.get_output(h, tail=10) + assert "[stdin] abc" in out["output"] + assert "ECHO:abc" in out["output"] + + def test_write_wait_sync_roundtrip(self, mgr, echo_srv): + """wait_lines 同步等待：一步问答（expect 式）""" + h = mgr.start(echo_srv)["handle"] + r = mgr.write(h, "1+1", wait_lines=1) + assert r["wait"]["lines"] == 1 + assert r["wait"]["output"].strip() == "ECHO:1+1" + assert r["wait"]["next_cursor"] == r["watermark"] + 1 + + def test_write_wait_timeout_on_silent_process(self, mgr): + """不读不响应的进程：写入成功（管道缓冲），等待如实超时""" + h = mgr.start(SLEEP_10)["handle"] + r = mgr.write(h, "ping", wait_lines=1, wait_timeout=0.5) + assert r["ok"] is True + assert r["wait"]["timed_out"] is True + assert r["wait"]["lines"] == 0 + + def test_write_to_exited_process(self, mgr): + h = mgr.start(HELLO)["handle"] + assert _wait_status(mgr, h, "exited") is not None + assert mgr.write(h, "x")["ok"] is False + + def test_write_rejects_empty_and_oversize(self, mgr, echo_srv): + h = mgr.start(echo_srv)["handle"] + assert mgr.write(h, "")["ok"] is False + assert mgr.write(h, None)["ok"] is False + assert mgr.write(h, "A" * (64 * 1024 + 1))["ok"] is False + + def test_stdin_eof_graceful_kill(self, mgr, echo_srv): + """三级优雅终止第一级：关 stdin → 回环服务器见 EOF 自行退出""" + h = mgr.start(echo_srv)["handle"] + time.sleep(0.5) + k = mgr.kill(h) # 不 force：先关 stdin + assert k["ok"] is True + assert mgr._procs[h].popen.poll() is not None + assert _wait_status(mgr, h, "killed") is not None + + def test_write_after_kill_rejected(self, mgr, echo_srv): + h = mgr.start(echo_srv)["handle"] + assert mgr.kill(h)["ok"] is True + assert mgr.write(h, "late")["ok"] is False + + def test_log_file_tee(self, mgr, monkeypatch, tmp_path): + """log_file 全档镜像：输出与 [stdin] 回显均落盘，可被原生工具读取""" + from common import constants as cm + monkeypatch.setattr(type(cm), "WORKSPACE_DIR", + property(lambda self: str(tmp_path))) + script = tmp_path / "srv.py" + script.write_text( + "import sys\n" + "for line in sys.stdin:\n" + " sys.stdout.write('ECHO:' + line.strip() + '\\n')\n" + " sys.stdout.flush()\n", + encoding="utf-8") + cmd = f'"{PY}" "{script}"' + r = mgr.start(cmd, log_file=True) + assert r["ok"] is True and r["log_file"] is not None + assert mgr.write(r["handle"], "tee-me", wait_lines=1)["ok"] is True + mgr.kill(r["handle"]) + log = (tmp_path / f"bg_{r['handle']}.log").read_text(encoding="utf-8") + assert "[stdin] tee-me" in log + assert "ECHO:tee-me" in log + + def test_bg_write_tool_flow(self, mgr, echo_srv): + """工具层冒烟：BgWriteTool 参数规整 + 对齐读取提示""" + assert BgRunTool(manager=mgr).execute(command=echo_srv).startswith("✅") + out = BgWriteTool(manager=mgr).execute(handle="1", text="hi", + wait_lines="1", wait_timeout=10) + assert "ECHO:hi" in out + assert "watermark" in out + # 描述里的行缓冲指引（给模型看的适用前提） + assert "PYTHONUNBUFFERED" in BgWriteTool(manager=mgr).get_description() + + def test_registry_dispatch_bg_write(self, mgr, echo_srv): + reg = ToolRegistry() + reg.register_many([BgRunTool(manager=mgr), BgWriteTool(manager=mgr)], + is_builtin=True) + assert reg.execute_tool("bg_run", command=echo_srv).startswith("✅") + r = reg.execute_tool("bg_write", handle=1, text="via-reg", + wait_lines=1, wait_timeout=10) + assert "ECHO:via-reg" in r + + +# ==================== 退出收尾 ==================== + +class TestShutdown: + def test_shutdown_kills_non_detach_keeps_detach(self): + m = ProcessManager() + try: + h1 = m.start(SLEEP_10)["handle"] # 常规：退出时终止 + h2 = m.start(SLEEP_10, detach=True)["handle"] # detach：保留 + time.sleep(0.5) + e1, e2 = m._procs[h1], m._procs[h2] + + r = m.shutdown() + assert r["killed"] == 1 and r["detached"] == 1 + assert e1.popen.poll() is not None # 常规进程已终止 + assert e2.popen.poll() is None # detach 进程仍存活 + assert m.list_procs() == [] # 注册表已清空（detach 仅注销） + + # 幂等 + assert m.shutdown() == {"killed": 0, "detached": 0} + # 收尾后拒绝新进程 + assert m.start(HELLO)["ok"] is False + # 清理 detach 测试进程 + e2.popen.kill() + finally: + m.shutdown() + + def test_singleton_shared_across_callers(self): + m1 = get_process_manager() + m2 = get_process_manager() + assert m1 is m2 # 跨会话/跨子代理共享同一注册表 + reset_process_manager(None) + m3 = get_process_manager() + assert m3 is not m1 + m3.shutdown() + reset_process_manager(None) + + +# ==================== 工具层 ==================== + +class TestTools: + def test_bg_run_and_output_flow(self, mgr): + run = BgRunTool(manager=mgr) + r = run.execute(command=HELLO, proc_name="hello") + assert r.startswith("✅") and "#" in r + handle = 1 + assert _wait_status(mgr, handle, "exited") is not None + time.sleep(0.3) + + out = BgOutputTool(manager=mgr).execute(handle=handle, since=0) + assert "hello-bg" in out + assert "next_cursor" in out + + def test_bg_run_string_handle_coercion(self, mgr): + """LLM 可能传字符串数字句柄（W 批输入规整惯例）""" + run = BgRunTool(manager=mgr) + assert run.execute(command=HELLO).startswith("✅") + assert _wait_status(mgr, 1, "exited") is not None + out = BgOutputTool(manager=mgr).execute(handle="1", tail=5) + assert "hello-bg" in out + + def test_bg_run_missing_command(self, mgr): + r = BgRunTool(manager=mgr).execute() + assert "缺少" in r + + def test_bg_list_empty_and_filled(self, mgr): + lst = BgListTool(manager=mgr) + assert "没有后台进程" in lst.execute() + mgr.start(SLEEP_10, name="sleeper") + r = lst.execute() + assert "sleeper" in r and "运行中 1" in r + + def test_bg_kill_flow(self, mgr): + mgr.start(SLEEP_10, name="sleeper") + time.sleep(0.5) + r = BgKillTool(manager=mgr).execute(handle=1) + assert r.startswith("✅") and "已终止" in r + + def test_disabled_gate_blocks_all(self, mgr, monkeypatch): + from common import constants + monkeypatch.setattr(type(constants), "BG_PROCESSES_ENABLED", + property(lambda self: False)) + assert "禁用" in BgRunTool(manager=mgr).execute(command=HELLO) + assert "禁用" in BgListTool(manager=mgr).execute() + assert "禁用" in BgOutputTool(manager=mgr).execute(handle=1) + assert "禁用" in BgKillTool(manager=mgr).execute(handle=1) + + def test_registry_dispatch(self, mgr): + """4 个工具经 ToolRegistry 正常分发（worker 注册链路冒烟）""" + reg = ToolRegistry() + reg.register_many([ + BgRunTool(manager=mgr), BgListTool(manager=mgr), + BgOutputTool(manager=mgr), BgKillTool(manager=mgr), + ], is_builtin=True) + r = reg.execute_tool("bg_run", command=HELLO, proc_name="smoke") + assert r.startswith("✅") + assert _wait_status(mgr, 1, "exited") is not None + time.sleep(0.3) + r = reg.execute_tool("bg_output", handle=1, since=0) + assert "hello-bg" in r + r = reg.execute_tool("bg_list") + assert "smoke" in r + + +def test_worker_only_tools_includes_bg(): + """编排者误调拦截表包含 bg 工具（worker 执行面边界）""" + from aibox import WORKER_ONLY_TOOLS + assert {"bg_run", "bg_write", "bg_list", "bg_output", "bg_kill"} <= WORKER_ONLY_TOOLS diff --git a/aibox/tools/__init__.py b/aibox/tools/__init__.py index 9e10876..207565c 100644 --- a/aibox/tools/__init__.py +++ b/aibox/tools/__init__.py @@ -45,6 +45,13 @@ from .prompt_tools import ( ) from .proxy_tools import SetProxyTool, GetProxyTool from .misc_tools import AskUserTool +from .process_tools import ( + BgRunTool, + BgWriteTool, + BgListTool, + BgOutputTool, + BgKillTool, +) __all__ = [ "Tool", @@ -71,4 +78,10 @@ __all__ = [ "SetProxyTool", "GetProxyTool", "AskUserTool", + # 后台进程管理工具（bg_run/bg_write/bg_list/bg_output/bg_kill） + "BgRunTool", + "BgWriteTool", + "BgListTool", + "BgOutputTool", + "BgKillTool", ] diff --git a/aibox/tools/process_tools.py b/aibox/tools/process_tools.py new file mode 100644 index 0000000..6cb9548 --- /dev/null +++ b/aibox/tools/process_tools.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +""" +tools.process_tools - 后台进程管理工具 + +五个 worker 执行面工具，配合 subsystems.process_manager.ProcessManager： + - BgRunTool : bg_run 后台启动长驻进程，立即返回句柄 + - BgWriteTool : bg_write 向后台进程 stdin 写入（watermark 写读对齐） + - BgListTool : bg_list 列出全部后台进程（运行中/已退出） + - BgOutputTool : bg_output 拉取输出（tail 尾部 / since 游标增量） + - BgKillTool : bg_kill 终止指定后台进程（关stdin→SIGTERM→SIGKILL） + +设计语义（与用户约定的三决策）： + - 用户中断（Ctrl+C/取消事件）不影响后台进程 + - aibox 进程退出时默认终止全部，bg_run detach=true 豁免 + - 注册表全局共享：任何子代理可管理其他子代理启动的进程 + +交互语义： + - stdin 为管道：行协议程序（REPL/问答 CLI）可 bg_write 写入； + watermark = [stdin] 回显行序号，bg_output(since=watermark) 即本条 + 输入的响应（写读对齐） + - bg_run log_file=true 时输出 tee 到 workspace/bg_<句柄>.log（无界 + 全档），可被原生工具（run_command tail/cat）读取 + +依赖：common (ConfigManager, constants) + tools.base.Tool + +subsystems.process_manager（get_process_manager 单例） +不反向依赖 aibox 任何符号。 +""" + +from typing import Dict + +from common import ConfigManager, constants +from subsystems.process_manager import get_process_manager, ProcessManager +from .base import Tool + + +def _enabled() -> bool: + """后台进程功能开关：commands.enabled 与 background_processes.enabled + 双闸（后台执行本质仍是命令执行，前者关闭时一并拒绝）""" + return constants.COMMANDS_ENABLED and constants.BG_PROCESSES_ENABLED + + +def _fmt_duration(seconds: float) -> str: + """秒数 → 人读时长（如 3h05m / 12.5s）""" + try: + seconds = float(seconds) + except (TypeError, ValueError): + return "?" + if seconds < 60: + return f"{seconds:.1f}s" + m, s = divmod(int(seconds), 60) + if m < 60: + return f"{m}m{s:02d}s" + h, m = divmod(m, 60) + return f"{h}h{m:02d}m{s:02d}s" + + +def _coerce_int(value, default=None): + """命令面输入规整（W 批惯例）：LLM 可能传字符串数字，统一安全转 int""" + if value is None: + return default + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _coerce_bool(value, default: bool) -> bool: + """命令面输入规整：LLM 可能传字符串 "false"/"0"，统一安全转 bool""" + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.strip().lower() not in ("false", "0", "no", "off", "") + return bool(value) + + +class BgRunTool(Tool): + """后台启动长驻进程""" + + def __init__(self, manager: ProcessManager = None, logger=None): + super().__init__(logger) + self.manager = manager or get_process_manager(logger) + + def get_name(self) -> str: + return "bg_run" + + def get_description(self) -> str: + return ("后台启动长驻进程（如 dev server、训练任务、watch 监控），立即返回句柄不阻塞。" + "与 run_command 的区别：run_command 同步等待命令结束（超时即杀），适合短命令；" + "bg_run 适合需要持续运行的进程，启动后用 bg_list 查看状态、bg_output 拉取输出、" + "bg_kill 终止。stdin 为管道：行协议交互程序（REPL/问答式 CLI）可用 bg_write 写入。" + "注意：用户中断不影响后台进程；aibox 退出时后台进程默认被终止，" + "需要跨重启存活的进程用 detach=true（随后无法再经句柄管理）。") + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "要执行的命令（shell 命令行）", + }, + "proc_name": { + "type": "string", + "description": "进程标签（可选，便于识别，≤64 字符，默认取命令前 40 字符）", + }, + "work_dir": { + "type": "string", + "description": "工作目录（默认 ~/.aibox/workspace）", + }, + "env": { + "type": "object", + "description": "额外的环境变量", + }, + "detach": { + "type": "boolean", + "description": "aibox 退出时不终止该进程（默认 false；detach 后进程脱离句柄管理）", + }, + "log_file": { + "type": "boolean", + "description": "同时镜像全程输出到 workspace/bg_<句柄>.log（无界全档，可被 run_command 的 tail/cat 读取；默认 false 仅环形缓冲）", + }, + }, + "required": ["command"], + } + + def execute(self, command: str = None, proc_name: str = None, + work_dir: str = None, env: Dict = None, + detach: bool = False, log_file: bool = False, **kwargs) -> str: + if error := self.validate_params(["command"], command=command): + return error + if not _enabled(): + return "❌ 后台进程功能已禁用（commands.enabled 或 background_processes.enabled 为 false）" + + result = self.manager.start(command, name=proc_name, work_dir=work_dir, + env=env, detach=detach, log_file=log_file) + if not result.get("ok"): + return self.format_result(False, "后台进程启动失败", + {"错误": result.get("error", "未知错误")}) + data = { + "句柄": result["handle"], + "PID": result["pid"], + "名称": result["name"], + "detach": "是（aibox 退出不终止）" if result.get("detach") else "否（aibox 退出时终止）", + "后续": f"bg_output 拉取输出 / bg_list 查看状态 / bg_kill 终止（句柄 #{result['handle']}）", + } + if result.get("log_file"): + data["日志镜像"] = result["log_file"] + return self.format_result(True, f"后台进程已启动（#{result['handle']}）", data) + + +class BgWriteTool(Tool): + """向后台进程 stdin 写入（交互式问答）""" + + def __init__(self, manager: ProcessManager = None, logger=None): + super().__init__(logger) + self.manager = manager or get_process_manager(logger) + + def get_name(self) -> str: + return "bg_write" + + def get_description(self) -> str: + return ("向后台进程的标准输入写入一行（行协议交互：REPL / 问答式 CLI）。" + "返回 watermark 水位：bg_output(handle=..., since=watermark) 得到的必然且仅是" + "本条输入引发的进程响应（写读对齐，不含 [stdin] 回显行）。" + "适用前提：进程按行读 stdin 并及时输出——建议启动时强制行缓冲" + "（python -i -u、PYTHONUNBUFFERED=1、stdbuf -oL），否则响应可能滞留子进程缓冲。" + "限制：无 TTY，全屏程序（vim/top 等）不可用；喂密码类交互（如 sudo -S）会留在" + "对话记录中，敏感凭据请改用环境变量注入。") + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "handle": { + "type": "integer", + "description": "bg_run 返回的句柄编号", + }, + "text": { + "type": "string", + "description": "要写入的文本（默认追加换行；多行输入会折叠为单行回显）", + }, + "append_newline": { + "type": "boolean", + "description": "写入后是否自动补换行（默认 true；false 发送原始字节）", + }, + "wait_lines": { + "type": "integer", + "description": "写完同步等待 N 行响应再返回（默认 0 立即返回，用 bg_output 轮询）", + }, + "wait_timeout": { + "type": "number", + "description": "同步等待响应的超时秒数（默认 5，上限 60）", + }, + }, + "required": ["handle", "text"], + } + + def execute(self, handle=None, text: str = None, append_newline=True, + wait_lines=None, wait_timeout=None, **kwargs) -> str: + if error := self.validate_params(["handle", "text"], handle=handle, text=text): + return error + if not _enabled(): + return "❌ 后台进程功能已禁用" + + handle = _coerce_int(handle) + if handle is None: + return "❌ 错误：handle 必须是整数句柄" + wait_i = _coerce_int(wait_lines, 0) + wait_t = _coerce_int(wait_timeout, 5) + + result = self.manager.write( + handle, text, + append_newline=_coerce_bool(append_newline, True), + wait_lines=max(0, wait_i), + wait_timeout=wait_t, + ) + if not result.get("ok"): + return self.format_result(False, f"写入后台进程 #{handle} 失败", + {"错误": result.get("error", "未知错误")}) + + data = { + "句柄": handle, + "已写入": f"{result['sent_bytes']} 字节", + "watermark": result["watermark"], + "对齐读取": f"bg_output(handle={handle}, since={result['watermark']}) → 本条输入的响应", + } + if "wait" in result: + w = result["wait"] + data["等待输出"] = w.get("output") or "(超时未收到响应)" + data["next_cursor"] = w.get("next_cursor") + if w.get("timed_out"): + data["警告"] = "等待响应超时（进程可能无输出或缓冲滞留），可用 bg_output 轮询" + return self.format_result(True, f"已写入后台进程 #{handle}", data) + + +class BgListTool(Tool): + """列出后台进程""" + + def __init__(self, manager: ProcessManager = None, logger=None): + super().__init__(logger) + self.manager = manager or get_process_manager(logger) + + def get_name(self) -> str: + return "bg_list" + + def get_description(self) -> str: + return ("列出全部后台进程（含已退出的，保留尾部输出供验尸）。运行中在前。" + "输出包含句柄/名称/PID/状态/运行时长，用句柄配合 bg_output/bg_kill 管理。") + + def get_parameters(self) -> Dict: + return {"type": "object", "properties": {}} + + def execute(self, **kwargs) -> str: + if not _enabled(): + return "❌ 后台进程功能已禁用" + procs = self.manager.list_procs() + if not procs: + return "✅ 当前没有后台进程（bg_run 启动）" + lines = [] + status_emoji = {"running": "🟢", "exited": "⚪", "killed": "🔴", "expired": "⏹️"} + for p in procs: + emoji = status_emoji.get(p["status"], "❔") + extra = "" + if p["status"] == "running": + extra = f"运行 {_fmt_duration(p['runtime_sec'])}" + else: + extra = f"已结束（{_fmt_duration(p['runtime_sec'])}，返回码 {p['exit_code']}）" + detach_tag = " [detach]" if p["detach"] else "" + lines.append(f"{emoji} #{p['handle']} {p['name']}{detach_tag} " + f"(pid {p['pid']}, {extra})") + lines.append(f" 命令: {p['command']}") + running = sum(1 for p in procs if p["status"] == "running") + header = f"共 {len(procs)} 个后台进程（运行中 {running}）" + return "✅ " + header + "\n" + "\n".join(lines) + + +class BgOutputTool(Tool): + """拉取后台进程输出""" + + def __init__(self, manager: ProcessManager = None, logger=None): + super().__init__(logger) + self.manager = manager or get_process_manager(logger) + + def get_name(self) -> str: + return "bg_output" + + def get_description(self) -> str: + return ("拉取后台进程输出。两种模式：tail=N 取最近 N 行（默认 50）；" + "since=游标 增量拉取上次读取之后的新输出（游标用上次返回的 next_cursor）。" + "长驻进程轮询建议用 since 模式避免重复传输旧输出。") + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "handle": { + "type": "integer", + "description": "bg_run 返回的句柄编号", + }, + "tail": { + "type": "integer", + "description": "取最近 N 行（默认 50；与 since 二选一）", + }, + "since": { + "type": "integer", + "description": "增量游标：上次返回的 next_cursor，取此序号之后的新行（与 tail 二选一）", + }, + "max_bytes": { + "type": "integer", + "description": "本次拉取最大字节数（默认 64KB，上限 1MB）", + }, + }, + "required": ["handle"], + } + + def execute(self, handle=None, tail=None, since=None, max_bytes=None, **kwargs) -> str: + if error := self.validate_params(["handle"], handle=handle): + return error + if not _enabled(): + return "❌ 后台进程功能已禁用" + + handle = _coerce_int(handle) + if handle is None: + return "❌ 错误：handle 必须是整数句柄" + since_i = _coerce_int(since) + tail_i = _coerce_int(tail) + + result = self.manager.get_output(handle, tail=tail_i, since=since_i, + max_bytes=_coerce_int(max_bytes, 65536)) + if not result.get("ok"): + return self.format_result(False, "拉取输出失败", + {"错误": result.get("error", "未知错误")}) + + data = { + "状态": result["status"] + (f"（返回码 {result['exit_code']}）" + if result["exit_code"] is not None else ""), + "本次行数": result["lines"], + "next_cursor": result["next_cursor"], + } + if result.get("truncated"): + data["警告"] = "输出超过本次字节上限，已截断（提高 max_bytes 或用 since 分段拉取）" + if result.get("evicted"): + data["提示"] = (f"有 {result['evicted']} 行早期输出因环形缓冲淘汰不可读取" + "（需要完整日志请让进程重定向到文件）") + output = result.get("output") or "(本次无新输出)" + data["输出"] = output + return self.format_result(True, f"后台进程 #{handle} 输出", data) + + +class BgKillTool(Tool): + """终止后台进程""" + + def __init__(self, manager: ProcessManager = None, logger=None): + super().__init__(logger) + self.manager = manager or get_process_manager(logger) + + def get_name(self) -> str: + return "bg_kill" + + def get_description(self) -> str: + return ("终止后台进程（整个进程组，含子进程）。默认先 SIGTERM 优雅终止，" + "宽限期内未退出再 SIGKILL；force=true 直接 SIGKILL。对已退出进程幂等。") + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "handle": { + "type": "integer", + "description": "bg_run 返回的句柄编号", + }, + "force": { + "type": "boolean", + "description": "跳过优雅终止直接 SIGKILL（默认 false）", + }, + "grace": { + "type": "number", + "description": "SIGTERM 后等待进程退出的宽限秒数（默认 3，上限 30）", + }, + }, + "required": ["handle"], + } + + def execute(self, handle=None, force: bool = False, grace: float = 3.0, **kwargs) -> str: + if error := self.validate_params(["handle"], handle=handle): + return error + if not _enabled(): + return "❌ 后台进程功能已禁用" + + handle = _coerce_int(handle) + if handle is None: + return "❌ 错误：handle 必须是整数句柄" + + result = self.manager.kill(handle, force=force, grace=grace) + if not result.get("ok"): + return self.format_result(False, f"终止后台进程 #{handle} 失败", + {"错误": result.get("error", "未知错误")}) + data = { + "句柄": handle, + "状态": result["status"], + } + if result.get("exit_code") is not None: + data["返回码"] = result["exit_code"] + if result.get("already_finished"): + return self.format_result(True, f"后台进程 #{handle} 此前已结束", data) + return self.format_result(True, f"后台进程 #{handle} 已终止", data)