feat(aibox): 后台进程池 v3.8——智能体运行/管理长驻进程与交互式 I/O

新增 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(平台环境项,与基线一致)零回归
This commit is contained in:
qing committed 2026-08-21 11:17:11 +08:00
1 parent 6f1f2bd296
commit 5cb35945a4
8 files changed
+1678 -8

No files matched your search

+35 -7
View File
@@ -96,8 +96,9 @@ aibox/
│ ├── db.py # MemoryDatabase + MemoDatabase(SQLite,FTS5 trigram 中文检索 + LIKE 兜底) │ ├── db.py # MemoryDatabase + MemoDatabase(SQLite,FTS5 trigram 中文检索 + LIKE 兜底)
│ ├── skill_manager.py # SkillManager(技能发现/加载/热重载) │ ├── skill_manager.py # SkillManager(技能发现/加载/热重载)
│ ├── conversation_history.py # ConversationHistory(对话历史管理) │ ├── conversation_history.py # ConversationHistory(对话历史管理)
│ └── conversation_compressor.py # ConversationCompressor(对话压缩,注入 llm_client 解耦) │ ├── conversation_compressor.py # ConversationCompressor(对话压缩,注入 llm_client 解耦)
├── tools/ # 工具系统包(A 批 + B 批共 17 个工具类) │ └── process_manager.py # ProcessManager(后台进程管理,全局单例 + 环形输出缓冲 + atexit 收尾)
├── tools/ # 工具系统包(A 批 + B 批 + 后台进程共 22 个工具类)
│ ├── base.py # Tool 抽象基类 │ ├── base.py # Tool 抽象基类
│ ├── registry.py # ToolRegistry 工具注册器 │ ├── registry.py # ToolRegistry 工具注册器
│ ├── basic_tools.py # A 批:ReadFile/WriteFile/ListFiles/SimpleWebCrawler/GetCurrentTime/Calculator │ ├── basic_tools.py # A 批:ReadFile/WriteFile/ListFiles/SimpleWebCrawler/GetCurrentTime/Calculator
@@ -106,8 +107,9 @@ aibox/
│ ├── db_tools.py # B 批:MemoryTool/MemoTool │ ├── db_tools.py # B 批:MemoryTool/MemoTool
│ ├── prompt_tools.py # B 批:UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation │ ├── prompt_tools.py # B 批:UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation
│ ├── proxy_tools.py # B 批:SetProxy/GetProxy │ ├── proxy_tools.py # B 批:SetProxy/GetProxy
│ └── misc_tools.py # B 批:AskUserTool │ ├── misc_tools.py # B 批:AskUserTool
├── tests/ # 单元测试套件(490 项全过) │ └── 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 桩 │ ├── conftest.py # sys.path 配置 + websockets 桩
│ ├── test_basic_tools.py # A 批工具 43 项 │ ├── test_basic_tools.py # A 批工具 43 项
│ ├── test_b_tools.py # B 批工具 58 项 │ ├── test_b_tools.py # B 批工具 58 项
@@ -126,7 +128,8 @@ aibox/
│ ├── test_i_fixes.py # 遗留收尾 Batch I(16 项) │ ├── test_i_fixes.py # 遗留收尾 Batch I(16 项)
│ ├── test_subagent_max_turns.py # 子代理最大轮次行为回归(14 项) │ ├── test_subagent_max_turns.py # 子代理最大轮次行为回归(14 项)
│ ├── test_k_fixes.py # 第五轮审计 Batch K(35 项) │ ├── 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 技能) ├── skills/ # 内置扩展技能(*.py 可执行脚本 / *.md Markdown 技能)
│ ├── hello-world.py │ ├── hello-world.py
│ ├── aes-crypto.py │ ├── aes-crypto.py
@@ -161,8 +164,8 @@ common ← subsystems ← tools ← aibox.py
```bash ```bash
# 从仓库根目录运行 # 从仓库根目录运行
py -m pytest aibox/tests -q py -m pytest aibox/tests -q
# 当前:490 项全过(Linux;test_execute_timeout / test_flat_api_still_reads_key_file 为 Windows 向用例,Linux 预存失败) # 当前:529 项全过(Linux;test_execute_timeout / test_flat_api_still_reads_key_file 为 Windows 向用例,Linux 预存失败)
# Windows 开发机:493 passed / 4 failed(平台环境项:symlink 权限、python3 缺失、POSIX 可执行位语义 ×2)/ 1 skipped # Windows 开发机:532 passed / 4 failed(平台环境项:symlink 权限、python3 缺失、POSIX 可执行位语义 ×2)/ 1 skipped
``` ```
运行期数据存放于 `~/.aibox/`: 运行期数据存放于 `~/.aibox/`:
@@ -238,6 +241,13 @@ py -m pytest aibox/tests -q
| `web` | 访问网页、获取标题/文本/链接(返回 markdown 指引,改用 `curl` 命令行访问) | | `web` | 访问网页、获取标题/文本/链接(返回 markdown 指引,改用 `curl` 命令行访问) |
| `calculator` | 数学计算 | | `calculator` | 数学计算 |
| `get_current_time` | 获取当前时间 | | `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 ## 💬 WebSocket API
@@ -615,6 +625,14 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`,通过 `update_i
"commands": { "commands": {
"enabled": true "enabled": true
}, },
"background_processes": {
"enabled": true,
"max_processes": 16,
"max_entries": 32,
"buffer_per_process": 262144,
"max_lifetime": 0,
"max_write_bytes": 65536
},
"web": { "web": {
"timeout": 10, "timeout": 10,
"delay": 0, "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 ### v3.7
- ✅ **第六轮安全审计 + Batch L 修复(2026-08-20,交叉验证轮:Batch K 回归验证 / 并发与生命周期 / 降级路径 / 修复间相互作用,4 路并行审读 + 主审逐条复核)**(完整记录见 [docs/reports/aibox_audit_20260817.md](../docs/reports/aibox_audit_20260817.md) §16): - ✅ **第六轮安全审计 + 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 快照等)均在位 - **Batch K 回归验证**:J1~J16 逐项在位、零回归,无修复引入的缺陷;历届 Batch E~I 修复抽查(executors reader 收尾、1MB 输出上限、registry 流式预算、broadcast 快照等)均在位
+10 -1
View File
@@ -55,6 +55,8 @@ from tools import (
UpdateIdentityTool, UpdateSoulTool, SaveMemoryAndEndConversationTool, UpdateIdentityTool, UpdateSoulTool, SaveMemoryAndEndConversationTool,
SetProxyTool, GetProxyTool, SetProxyTool, GetProxyTool,
AskUserTool, AskUserTool,
# 后台进程管理工具(bg_run/bg_write/bg_list/bg_output/bg_kill,worker 执行面)
BgRunTool, BgWriteTool, BgListTool, BgOutputTool, BgKillTool,
) )
# 子系统(Wave 1 + Wave 2 抽离):命令执行器 / 提示词管理器 / 数据库 / 技能管理器 / 对话历史 / 对话压缩 # 子系统(Wave 1 + Wave 2 抽离):命令执行器 / 提示词管理器 / 数据库 / 技能管理器 / 对话历史 / 对话压缩
@@ -987,7 +989,8 @@ TOOL_PROFILE_ORCHESTRATOR = "orchestrator"
TOOL_PROFILE_WORKER = "worker" TOOL_PROFILE_WORKER = "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 = ( ORCHESTRATOR_TURN_GUIDANCE = (
@@ -1179,6 +1182,12 @@ class DeepSeekChat:
SimpleWebCrawlerTool(self.logger), SimpleWebCrawlerTool(self.logger),
CalculatorTool(self.logger), CalculatorTool(self.logger),
GetCurrentTimeTool(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.tool_registry.register_many(worker_tools, is_builtin=True)
self.logger.info(f"worker 子代理已注册 {len(worker_tools)} 个执行面工具") self.logger.info(f"worker 子代理已注册 {len(worker_tools)} 个执行面工具")
+25
View File
@@ -236,6 +236,31 @@ class Constants:
def COMPRESSOR_LLM_MAX_TOKENS(cls): def COMPRESSOR_LLM_MAX_TOKENS(cls):
return cls._get_config().get("compressor.llm_max_tokens", 1000) 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() constants = Constants()
+8
View File
@@ -24,6 +24,11 @@ from .db import MemoryDatabase, MemoDatabase
from .skill_manager import SkillManager from .skill_manager import SkillManager
from .conversation_history import ConversationHistory from .conversation_history import ConversationHistory
from .conversation_compressor import ConversationCompressor from .conversation_compressor import ConversationCompressor
from .process_manager import (
ProcessManager,
get_process_manager,
reset_process_manager,
)
__all__ = [ __all__ = [
"CommandExecutor", "CommandExecutor",
@@ -34,4 +39,7 @@ __all__ = [
"SkillManager", "SkillManager",
"ConversationHistory", "ConversationHistory",
"ConversationCompressor", "ConversationCompressor",
"ProcessManager",
"get_process_manager",
"reset_process_manager",
] ]
+731
View File
@@ -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
+448
View File
@@ -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
+13
View File
@@ -45,6 +45,13 @@ from .prompt_tools import (
) )
from .proxy_tools import SetProxyTool, GetProxyTool from .proxy_tools import SetProxyTool, GetProxyTool
from .misc_tools import AskUserTool from .misc_tools import AskUserTool
from .process_tools import (
BgRunTool,
BgWriteTool,
BgListTool,
BgOutputTool,
BgKillTool,
)
__all__ = [ __all__ = [
"Tool", "Tool",
@@ -71,4 +78,10 @@ __all__ = [
"SetProxyTool", "SetProxyTool",
"GetProxyTool", "GetProxyTool",
"AskUserTool", "AskUserTool",
# 后台进程管理工具(bg_run/bg_write/bg_list/bg_output/bg_kill)
"BgRunTool",
"BgWriteTool",
"BgListTool",
"BgOutputTool",
"BgKillTool",
] ]
+408
View File
@@ -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)