diff --git a/CHANGELOG.md b/CHANGELOG.md index dc0cffc..1f937ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.56.0 — 2026-07-10 + +- 长时间运行的脚本 / 命令支持**后台执行**:耗时任务(大批量数据处理、长时间计算等)自动转入后台独立运行,**不再被超时打断,服务更新也不中断**。 +- 后台任务运行期间体验与普通执行一致:卡片实时跳秒、发送按钮变"停止"可一键终止;**完成时右下角弹窗提醒**(切到其他对话也能收到,点击跳转),对话随即解锁可继续。 +- 视频生成等待上限从 10 分钟延长到 20 分钟;等待超时或中断后**可以继续取回结果**,不重复计费、不用重新生成。 + ## 0.55.2 — 2026-07-10 - 文件面板根目录不再显示定时任务 / 微信对话自动生成的工作文件夹(如 `scheduled-xxxx`),减少干扰;点开对应定时任务或微信对话,右侧文件面板照常查看其中的文件。 diff --git a/DESIGN.md b/DESIGN.md index 999dde4..a7a317f 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -327,6 +327,23 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) **实施时对账清单**:①子循环禁注 `delegate` 自身(防自我繁殖,同 8.5 定时 run 禁 schedule_create);②工具集只读白名单,不给 shell/fs 写/run_python;③计费入 usage_events 归属主 task(kind 区分);④主 run 的 cancel_check 传导进子循环;⑤子循环事件不直播 SSE(或只 emit 一条聚合摘要事件),防前端刷屏;⑥子循环产出若超长仍走"落文件留路径"纪律(§8.2 质量边界)。 +### 8.12 后台进程 bg proc:detach + 文件系统状态,不引队列组件(2026-07-10) + +**根因**:模型写的长脚本(批量数据处理/模拟计算)在工具内同步跑,三个结构性问题:①工具超时(shell 60s / run_python 120s)把真实长任务掐死,长程复杂任务做不了;②就算放大 timeout,run 被占死几十分钟(单活 run 锁,用户 409);③进程是 zcbot 实例的子进程,蓝绿切换旧实例退出时**必然陪葬**——超时调多大都躲不过部署窗口。 + +**决策**:把长进程从 zcbot 进程树上摘下来,OS 就是"任务组件",不引 Celery/RQ/队列。`shell`/`run_python` 加 `background=true`(**模型判断**:预计 >~1min 走后台;前台超时报错里提示改后台——判断错了纠错路径只有一步;用户显式指令永远优先)。状态协议**纯文件**:`/.zcbot_procs///{proc.json, output.log, exit_code}`——dotfile 用户不可见(同 `.zcbot_tmp` 惯例),`exit_code` 文件出现是唯一终态信号,状态判定全靠文件 + 现场探测(pid / 容器 running),**无常驻登记,天然扛重启**。查询/终止走配套 `check_process` 工具(host in-process,两种 backend 通吃)。 + +- **host backend**:detach 独立 wrapper(`core/proc_wrapper.py`,stdlib-only,sys.executable 直跑不依赖 PYTHONPATH):限时(默认 7200s / cap 86400s)、超时杀进程树记 124、日志截尾 10MB、最后写 exit_code。 +- **docker backend**:**专用容器** `zcbot-proc-`(pool.run_proc_container,同款硬化 + iptables init),`product=proc` + 无 instance label —— 与 sandbox 容器的 idle reaper / shutdown_all 生命周期**解耦**,dockerd 托管,蓝绿切换/实例重启不中断。不用 `docker exec -d` 进 sandbox 容器:idle 5min reaper + 启动 shutdown_all 会把长进程随容器带走。 +- **回收**:check_process 见终态顺手 rm 容器;web lifespan 每小时 `procs.sweep`(终态目录 7d TTL / exited 孤儿容器),幂等,蓝绿双实例同时跑无害。 +- **通知/可视**:不做服务端推送 —— 前端轮询 `GET /v1/procs`(用户级,纯文件读取,仅有 running proc 时 5s 一拉):`[Background]` 工具结果卡本身活化(spinner+跳秒+停止按钮,与前台工具卡同体验,历史重渲同样恢复;`POST .../procs//kill`)、running→终态弹 toast(跨 task 也提醒,点击跳转)。proc 完成时刻往往没有活跃 run,SSE 通道根本不在,轮询是诚实的选型。 +- **对话锁(前端)**:bg proc 运行期间该 task 的 composer 锁定(发送→停止,Enter 拦截),观感与前台执行完全一致 —— 后台化的收益定位为「进程扛超时/服务重启」,**不改变"一个任务同时只做一件事"的对话心智**;完成的那次轮询解锁 + toast「可继续对话」。锁只在前端,服务端不 409:「停止」入口必须可达,且多设备/渠道绕过前端锁属可接受边缘(等的是同一个进程,发了消息也不冲突)。 +- **防失控**:每用户并发 running 上限(`ZCBOT_MAX_BG_PROCS` 默 3);前台默认超时不放大(它是逼模型做前台/后台选择的杠杆)。 + +**边界(防滑坡)**:只覆盖「单个本地长进程」。①**外部异步作业**(seedance 等 submit/poll 形态)不进这里——工具内轮询 + `resume_task_id` 续查已够;②**job 链/依赖/自动重试**不做——那是 workflow 引擎,编排的唯一归属是 agent loop(模型 check 后自己决定下一步),同 §6 拒绝编排的理由;③**完成后自动续跑 run**不做——zcbot 的长任务产物多为终点交付物(与 Claude Code"build 是中间步骤"不同),自动续跑=无人在场烧 token,通知给人、下一步由人/下次对话决定。 + +**不选**:Celery/RQ(多机分发/任务序列化/框架重试——单机 + 模型现写脚本的场景一个都用不上,还多两个常驻组件的部署/蓝绿适配);工具层 async 化 run 内等待(run 不结束,409 照旧,重启照丢);DB 表 + 守护(文件已是事实源,detach 进程写 PG 还得给它凭证)。升级触发:要跨机器跑计算集群时,①②的工具接口不变,只换执行后端。 + --- ## 附录:DeepSeek V4 关键事实(2026-04-24) diff --git a/PROGRESS.md b/PROGRESS.md index 0eb8109..e1a7b26 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-07-10(per-user 磁盘配额 5GB→20GB,bump 0.55.1) +最后更新:2026-07-10(后台进程 bg proc + seedance 续查 + LLM 显式超时,bump 0.56.0) --- @@ -23,6 +23,7 @@ ### 2026-07 +- **07-10 / 0.56.0**:**后台进程 bg proc(DESIGN §8.12)**——解掉"长脚本被工具超时掐死 / 占死 run / 蓝绿切换陪葬"三连:`shell`/`run_python` 加 `background=true`(模型按预计时长自选,前台超时报错里提示改走后台;默认上限 7200s、cap 86400s、每用户并发 3 个 `ZCBOT_MAX_BG_PROCS`),新 `check_process` 工具查状态/日志尾部/kill。状态协议纯文件(`/.zcbot_procs///{proc.json,output.log,exit_code}`,dotfile 用户不可见,exit_code 出现即终态),无 DB 无队列组件。host 模式 detach 独立 wrapper(`core/proc_wrapper.py`,stdlib-only,限时+杀树+截尾 10MB);docker 模式**专用容器** `zcbot-proc-`(`pool.run_proc_container`,同款硬化/iptables init,product=proc + 无 instance label → 与 sandbox 的 idle reaper/shutdown_all 生命周期解耦,dockerd 托管扛蓝绿)。回收:check_process 见终态顺手 rm 容器 + web lifespan 每小时 `procs.sweep`(终态目录 7d TTL、exited 孤儿容器)。**前端可视 + 通知**:`GET /v1/procs`(用户级)+ `POST /v1/tasks/{id}/procs/{pid}/kill` 两端点(纯文件读取无 DB);`procs.js` 把 `[Background]` 工具结果卡活化(spinner+summary 跳秒+停止按钮,与前台工具卡同体验;历史重渲同样恢复活态/终态定格 exit·耗时)+ running→终态弹 toast(右下角,跨 task 也提醒、点击跳转对应任务)+ **对话锁**(bg proc 运行期 composer 发送→停止、Enter 拦截,观感与前台执行一致,完成的那次轮询解锁;锁仅前端,服务端不 409——停止入口须可达);轮询式(仅有 running proc 时 5s 一拉,触发点:登录/选 task/run 收尾/[Background] 工具结果)——不走 SSE,proc 完成时刻往往没有活跃 run。LLM 超时最终定为**显式化默认 600s + env 可调**(长工具已走后台,LLM 单调用无正当 600s 静默;被掐再调 `ZCBOT_LLM_TIMEOUT_S`)。手工验证 `scripts/test_bgproc_manual.py` 全过(win host 模式:启动/探活/exit 0/日志/kill 137/sweep),unittest 全量 201 过。同批带上:**seedance `resume_task_id` 续查**(超时/中断后拿 cgt_id 跳过提交直接续轮询,不过配额闸不重复计费;fast/pro poll_timeout 统一 1200s)+ **LLM 超时显式化**(`core/llm.py` 显式传 litellm timeout,env `ZCBOT_LLM_TIMEOUT_S` 可调,见后)。 - **07-10 / 0.55.2**:文件面板根目录隐藏系统工作目录:`/v1/files`(仅根层)与 `/v1/folders`(新建任务目录候选)过滤定时任务执行目录 / 渠道镜像对话目录(`web/app.py::_system_wd_names`,DB 回查 `scheduled_job_id`/`channel` 判定而非硬编码名字前缀 —— 孤儿目录(task 已删)无任务入口,刻意留在根目录可见)。只是列表降噪非权限拦截:带 path 直接访问照常放行,点定时任务运行历史 / 微信卡片时文件面板自动跳入该目录不受影响(chat.js 既有逻辑,前端零改动)。 - **07-10 / 0.55.1**:per-user 磁盘配额 5GB→20GB(`config/agent.yaml quotas.disk_bytes_per_user`;RUN 故障兜底行、CHANGELOG 同步)。仍是应用层软配额,OS 层 xfs prjquota 兜底照旧留待外部用户开放前(§7.5 #4)。 - **07-09 / 0.55.0**:**用户版更新日志**(点版本号看"更新了什么"):新增仓库根 `CHANGELOG.md`(面向用户口径,只记可感知变化,回填至 0.30;与工程笔记 PROGRESS 分离——PROGRESS 满是内部模块/env/部署细节不宜外露);`GET /v1/changelog`(公开无鉴权,`## <版本> — <日期>` 正则切段,mtime 缓存热改即生效,无 DB 零 migration);前端右栏底部版本号变可点 → 更新日志弹层(`changelog.js`,复用 marked+purify 渲染),localStorage 记 last-seen 版本、与 /healthz 版本不一致时版本号旁亮红点、打开即清(公测用户发现更新的唯一入口)。CLAUDE.md 加维护约定:bump 版本若用户可感知顺手补 CHANGELOG 条目。 diff --git a/RUN.md b/RUN.md index c382071..0fb9a6a 100644 --- a/RUN.md +++ b/RUN.md @@ -68,6 +68,12 @@ # 蓝绿部署强烈建议设:消掉切换窗口「刷新丢直播 / 停止失灵」两个边缘,也是将来稳态 # 双实例分流的前提。 # ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0 + # 单次 LLM 请求超时(秒),默 600(与 litellm 默认一致,显式化 + 可调)。长思考模型 + # 被掐("600s 无字节 → run 标 error")再调大;流式正常出 chunk 不触发。 + # ZCBOT_LLM_TIMEOUT_S=600 + # 后台进程(bg proc,DESIGN §8.12;shell/run_python background=true):每用户并发上限,默 3。 + # 状态锚 /.zcbot_procs/,后台默认限时 7200s(工具 timeout 参数可调,cap 86400)。 + # ZCBOT_MAX_BG_PROCS=3 # 定时任务守护循环(DESIGN §8.5,随 web 进程起,plain-asyncio 仿 _disk_scanner): # ZCBOT_DISABLE_SCHEDULER=1 # 可选,整体关掉调度(对照 Claude Code CLAUDE_CODE_DISABLE_CRON) # ZCBOT_SCHEDULER_TICK_SECONDS=10 # 可选,扫描间隔,默 10s(只决定最坏延迟≤1tick,不影响会否漏) @@ -240,6 +246,8 @@ curl --noproxy '*' -H "Authorization: Bearer $TOKEN" http://127.0.0.1:8765/v1/ta | `POST /v1/tasks/{id}/messages` | `{content, image_model?=""}` 发消息;返 `{events_url}`;**`run_status` 是 running/cancelling → 409**(单活 run;error 起新 run 时清);`image_model` 是 `config/media/doubao.yaml` image 段的 variant key(空 → 沿用 yaml 第一个),仅本 run 装配 SeedreamTool 时使用,不入 DB;UI 应 disable send 直到 SSE `done` | 必填 | | `GET /v1/tasks/{id}/events` | SSE 流(`event: ` + `data: `);订阅 task 当前活动 | 必填 | | `POST /v1/tasks/{id}/cancel` | 协作式 cancel;`run_status != running` → 409;LLM 走 streaming,chunk 间 poll cancel — 延迟 100ms 级,基本秒退 | 必填 | +| `GET /v1/procs` | 当前用户全部后台进程(bg proc,§8.12;shell/run_python `background=true` 启动);纯文件系统读取,前端运行条 5s 轮询用 | 必填 | +| `POST /v1/tasks/{id}/procs/{proc_id}/kill` | 强制终止后台进程(host 杀进程树 / docker rm 容器);幂等 | 必填 | | `POST /v1/tasks/{id}/clear` | 清空当前 task 全部 messages + reset `tasks.tokens_prompt/completion/cost_cny` 三列累计 + `run_status='idle'`;`usage_events`(账单记账)**不动**,只 `message_id` 列变 NULL;run 活跃中(running/cancelling)→ 409(先 cancel);FS 文件保留 | 必填 | | `POST /v1/tasks/{id}/optimize_prompt` | body `{text(req, ≤4000), image_model?=""}`;同步调当前 task model 润色草稿,返 `{optimized, model_profile, tokens_in, tokens_out, cost_cny}`;**不**写 messages、**不**累计 task 三列(顶栏数字不污染),只在 `usage_events` 写一行 `kind="prompt_optimize"`(对账可见);不与主对话 run 互斥(允许 streaming 中并行润色) | 必填 | | `WS /v1/asr/stream` | 流式语音转写(前端「🎙 语音」主通道):连上后首条文本帧发 `{"token":""}`(WS 塞不了 header,token 不走 query 防进 access log,失败 close 4401)→ 之后二进制帧 = PCM 分片(16k/16bit/mono)实时转发讯飞;任意文本帧(约定 `{"type":"end"}`)= 说完。服务端回推 `{"text","final"}` 增量全文(wpgs 动态修正已合并),final=true 为最终结果;错误回 `{"error"}` 后关闭。nginx 反代该路径需支持 WS Upgrade(`proxy_set_header Upgrade/Connection`) | 首消息 | @@ -822,6 +830,8 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_" /opt | prod 想把 workspace 落独立数据盘 | **别用 env / 别指 ROOT 外绝对路径**(workspace 锚定 ROOT,ROOT 外会让文件面板 / agent / 新建 task 三家分叉)。用 **bind mount** 把 `/data/...` 接到 `ROOT/workspace`,逻辑路径不变,DB 不用改。详「workspace 落独立数据盘」段 | | 文件面板"目录尚未创建"但文件确实在 / agent 写的文件面板看不到 | workspace 被指到了 ROOT 外(旧 `ZCBOT_WORKSPACE_DIR` 绝对路径残留)→ 文件面板走 `resolve_workspace` 看一处、agent 走 DB `from_db_path`(锚 ROOT)看另一处。删掉 env、改用 bind mount(见上段),三家归一 | | `docker run zcbot-sandbox:latest` 报 `Unable to find image` | 镜像没 build。`sudo -u zcbot docker build -f deploy/sandbox/Dockerfile --build-arg HOST_UID=$(id -u zcbot) --build-arg HOST_GID=$(id -g zcbot) -t zcbot-sandbox:latest .` | +| 后台进程(bg proc)疑似残留 / 想手工排查 | docker 模式:`docker ps --filter label=zcbot.product=proc` 列在跑的 proc 容器,`docker rm -f zcbot-proc-` 手杀;host 模式:状态锚在 `/.zcbot_procs///`(proc.json 有 pid,exit_code 文件在 = 已结束)。web 进程每小时 sweep 自动回收(终态目录 7 天 TTL);正常终止走前端停止按钮 / `check_process(action="kill")` | +| 后台进程状态显示 lost | 进程没留退出码就没了 —— 宿主重启 / OOM killer / docker daemon 重启把它带走(bg proc 扛 zcbot 重启和蓝绿,但不扛宿主级重启)。output.log 保留到中断为止,需重新 background=true 发起 | | 镜像 build pip 报 `THESE PACKAGES DO NOT MATCH THE HASHES FROM THE REQUIREMENTS FILE`(本仓 requirements 未钉 hash) | **不是被篡改、也不是 require-hashes**:镜像 index 声明的 wheel hash 与它实际吐出的文件字节不符 = 该镜像存的文件损坏 / 截断(2026-06-03 腾讯源就这么坏过 litellm-1.87.0)。换源重 build:`PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple/ sudo -E bash deploy/update.sh`。验真伪:`https://pypi.org/pypi///json` 看官方 sha256 是哪边对。与下面"版本滞后(Could not find)"是两回事 | | 镜像 build pip 报 `ReadTimeoutError: HTTPSConnectionPool(host='files.pythonhosted.org', ...)` | 境内访问 PyPI 抖动。加 `--build-arg PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple/`(清华,现默认)或腾讯 / 阿里源,详 RUN.md「镜像构建」段。Dockerfile 已把 pip timeout 拉到 60s,主因仍是源不通而非超时 | | pip 报 `Could not find a version that satisfies the requirement litellm>=1.83.0`(伴随一串 `Ignored ... yanked versions: 0.1.xxxx`) | 用的镜像源同步滞后,没有该新版本。**阿里 PyPI 一度只到 litellm 1.82.6** —— update.sh 默认已是清华源(同步及时)。若手动 build 撞到:换 `PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple/` 或腾讯源。那串 `0.1.xxxx` 是 litellm 远古版本,纯干扰信息 | diff --git a/config/media/doubao.yaml b/config/media/doubao.yaml index b0ee03b..98aa087 100644 --- a/config/media/doubao.yaml +++ b/config/media/doubao.yaml @@ -84,7 +84,7 @@ video: # 轮询参数 request_timeout_s: 60 # submit POST 超时(异步,只是提交) poll_interval_s: 5 # 单次 GET 间隔(秒);典型 30-90s 出片 - poll_timeout_s: 600 # 总等待上限(10min)→ 超时返 [Error] + poll_timeout_s: 1200 # 总等待上限(20min)→ 超时返 [Error]+cgt_id,可 resume_task_id 续查 seedance_2_pro: model_id: doubao-seedance-2-0-260128 @@ -113,4 +113,4 @@ video: # 轮询参数:Pro 出片慢于 Fast(更精细),拉长超时 request_timeout_s: 60 poll_interval_s: 5 - poll_timeout_s: 900 # Pro 上限拉到 15min 保险 + poll_timeout_s: 1200 # Pro 出片更慢,上限 20min;超时后同样可 resume_task_id 续查 diff --git a/core/__init__.py b/core/__init__.py index 03cdd09..604ca1b 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.55.2" +__version__ = "0.56.0" diff --git a/core/agent_builder.py b/core/agent_builder.py index 065f352..595de49 100644 --- a/core/agent_builder.py +++ b/core/agent_builder.py @@ -46,6 +46,7 @@ from tools.materials_project import ( MaterialsProjectSearchSummaryTool, ) from tools.look_at_image import LookAtImageTool +from tools.check_process import CheckProcessTool from tools.run_python import RunPythonTool from tools.seedance import SeedanceTool from tools.seedream import SeedreamTool @@ -500,10 +501,18 @@ def build_agent( au = AskUserTool(base_dir=tool_base, user_root=ur_path) tools[au.name] = au - for cls in (ReadTool, WriteTool, EditTool, GlobTool, GrepTool, ShellTool): + for cls in (ReadTool, WriteTool, EditTool, GlobTool, GrepTool): t = cls(base_dir=tool_base, user_root=ur_path) tools[t.name] = t + # shell/run_python 带 task_id:background=true 的 bg proc 状态锚定 + # `/.zcbot_procs//`(DESIGN §8.12)。check_process 是它们的 + # 配套查询/终止工具,host in-process(docker 模式下状态文件也在宿主侧)。 + sh = ShellTool(base_dir=tool_base, user_root=ur_path, task_id=task_id) + tools[sh.name] = sh + cp = CheckProcessTool(task_id=task_id, base_dir=tool_base, user_root=ur_path) + tools[cp.name] = cp + # web_fetch 无需 API key,始终可用 wf = WebFetchTool(base_dir=tool_base, user_root=ur_path) tools[wf.name] = wf @@ -583,7 +592,7 @@ def build_agent( tools[wp.name] = wp if caps.enable_run_python: - rp = RunPythonTool(base_dir=tool_base, user_root=ur_path) + rp = RunPythonTool(base_dir=tool_base, user_root=ur_path, task_id=task_id) tools[rp.name] = rp # 每账号每日配额(yaml `quotas` 段,跨 task 跨 variant 全口径合计; diff --git a/core/executor_docker.py b/core/executor_docker.py index 4d3862a..2364400 100644 --- a/core/executor_docker.py +++ b/core/executor_docker.py @@ -174,6 +174,8 @@ class DockerExecutor(Executor): content="[Error] bad arguments to shell: command must be non-empty string", exit_code=2, ) + if args.get("background"): + return self._exec_background("shell", args, ctx) timeout = int(args.get("timeout") or 60) container = self.pool.ensure(self.user_id) @@ -192,6 +194,8 @@ class DockerExecutor(Executor): # ── run_python ─────────────────────────────────────────── def _exec_python(self, args: Dict[str, Any], ctx: ExecCtx) -> ToolResult: + if args.get("background"): + return self._exec_background("run_python", args, ctx) code = args.get("code") script_path = args.get("script_path") if script_path is not None: @@ -242,6 +246,121 @@ class DockerExecutor(Executor): except OSError: pass + # ── background(bg proc,DESIGN §8.12)────────────────── + + def _exec_background(self, name: str, args: Dict[str, Any], ctx: ExecCtx) -> ToolResult: + """background=true:专用容器跑长进程,立即返回 proc_id。 + + 为什么不是 `docker exec -d` 进 sandbox 容器:sandbox 有 idle 5min reaper + + 启动时 shutdown_all,长进程会随容器陪葬。专用容器 product=proc 与这两条 + 生命周期解耦(pool.run_proc_container),状态协议同 host 模式(core/procs.py): + runner.sh 结束时写 exit_code,check_process/sweep 负责回收容器。 + """ + from core import procs + + anchor = self.user_root + if procs.count_running(anchor) >= procs.MAX_RUNNING_PER_USER: + return ToolResult( + content=( + f"[Error] 已有 {procs.MAX_RUNNING_PER_USER} 个后台进程在跑(上限)。" + f"用 check_process 查看,等待完成或 kill 掉不需要的再启动。" + ), + exit_code=2, + ) + + raw_timeout = args.get("timeout") + fg_default = 60 if name == "shell" else 120 + try: + raw_timeout = int(raw_timeout) if raw_timeout else 0 + except (TypeError, ValueError): + raw_timeout = 0 + timeout_s = procs.clamp_timeout(raw_timeout if raw_timeout > fg_default else None) + + proc_id = procs.new_proc_id() + d = procs.procs_root(anchor) / str(ctx.task_id) / proc_id + d.mkdir(parents=True, exist_ok=True) + cdir = f"/workspace/{procs.PROCS_SUBDIR}/{ctx.task_id}/{proc_id}" + + if name == "shell": + cmd = args.get("command") + if not isinstance(cmd, str) or not cmd.strip(): + return ToolResult( + content="[Error] bad arguments to shell: command must be non-empty string", + exit_code=2, + ) + (d / "cmd.sh").write_text(cmd + "\n", encoding="utf-8", newline="\n") + inner = f"bash {cdir}/cmd.sh" + display, kind = cmd, "shell" + else: + script_path = args.get("script_path") + code = args.get("code") + if isinstance(script_path, str) and script_path.strip(): + inner = f"python {self._container_script_path(script_path)}" + display, kind = f"python {script_path.strip()}", "python" + elif isinstance(code, str) and code.strip(): + (d / "script.py").write_text(code, encoding="utf-8", newline="\n") + inner = f"python {cdir}/script.py" + display, kind = f"python ", "python" + else: + return ToolResult( + content="[Error] bad arguments to run_python: code or script_path must be provided", + exit_code=2, + ) + + # runner.sh:限时跑 + 截日志 + 写 exit_code(exit_code 是唯一终态信号,最后写) + runner = ( + "#!/bin/bash\n" + f"cd {self.container_workdir} || {{ echo '[runner] cd failed' > {cdir}/output.log; " + f"echo 1 > {cdir}/exit_code; exit 1; }}\n" + f"timeout {timeout_s}s {inner} > {cdir}/output.log 2>&1\n" + "ec=$?\n" + f"if [ $ec -eq 124 ]; then echo '[runner] timeout after {timeout_s}s, killed' >> {cdir}/output.log; fi\n" + f"tail -c 10485760 {cdir}/output.log > {cdir}/output.log.trunc 2>/dev/null " + f"&& mv {cdir}/output.log.trunc {cdir}/output.log\n" + f"echo $ec > {cdir}/exit_code\n" + ) + (d / "runner.sh").write_text(runner, encoding="utf-8", newline="\n") + + meta: Dict[str, Any] = { + "proc_id": proc_id, + "task_id": str(ctx.task_id), + "kind": kind, + "backend": "docker", + "command": display[:500], + "cwd": self.container_workdir, + "timeout_s": timeout_s, + "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"), + "created_ts": time.time(), + } + procs.write_meta(d, meta) + + try: + container = self.pool.run_proc_container(self.user_id, proc_id, str(d)) + except Exception as e: + return ToolResult(content=f"[Error] 后台容器启动失败: {e}", exit_code=1) + meta["container"] = container + procs.write_meta(d, meta) + + argv = self._docker_exec_argv( + container, extra_env=_sandbox_env({"PYTHONIOENCODING": "utf-8"}), detach=True + ) + ["bash", f"{cdir}/runner.sh"] + r = subprocess.run(argv, capture_output=True, text=True, timeout=60) + if r.returncode != 0: + subprocess.run(["docker", "rm", "-f", container], capture_output=True) + return ToolResult( + content=f"[Error] 后台进程启动失败: {(r.stderr or '').strip()[:300]}", + exit_code=1, + ) + return ToolResult( + content=( + f"[Background] 已启动后台进程 proc_id={proc_id}({display[:150]})," + f"最长运行 {timeout_s}s。\n" + f"用 check_process(proc_id=\"{proc_id}\") 查进度和日志。进程独立于本轮对话运行," + f"服务重启也不中断。现在可以继续其他工作;若无事可做,结束回合并告知用户稍后询问进度。" + ), + exit_code=0, + ) + def _container_script_path(self, script_path: str) -> str: p = script_path.replace("\\", "/").strip() if p.startswith("/"): @@ -291,8 +410,10 @@ class DockerExecutor(Executor): container: str, extra_env: Optional[Dict[str, str]] = None, stdin_open: bool = False, + detach: bool = False, ) -> List[str]: - """`stdin_open=True` 时加 `-i` 让 stdin 通到容器(fs tool_runner 用)。""" + """`stdin_open=True` 时加 `-i` 让 stdin 通到容器(fs tool_runner 用); + `detach=True` 加 `-d`(bg proc 的 runner.sh,不接 stdio 立即返回)。""" argv = [ "docker", "exec", "--user", self.exec_user, @@ -300,6 +421,8 @@ class DockerExecutor(Executor): ] if stdin_open: argv.append("-i") + if detach: + argv.append("-d") env: Dict[str, str] = {} if extra_env: env.update(extra_env) diff --git a/core/llm.py b/core/llm.py index 0753800..41e55bc 100644 --- a/core/llm.py +++ b/core/llm.py @@ -26,6 +26,12 @@ from litellm.exceptions import ( from .capabilities import ModelCapabilities +# 单次 LLM 请求超时(秒),默认与 litellm 一致(600s)但显式化 + env 可调 ── +# 长思考模型真被掐("600s 无字节 → run 标 error")时调大 ZCBOT_LLM_TIMEOUT_S 即可, +# 不用改代码。流式场景它主要约束"无字节间隔",正常出 chunk 不触发。 +# 长工具调用不受此限:>1min 的脚本走 background=true(DESIGN §8.12),不占 LLM 超时。 +_REQUEST_TIMEOUT_S = int(os.getenv("ZCBOT_LLM_TIMEOUT_S", "600")) + class LLM: def __init__(self, capabilities: ModelCapabilities) -> None: @@ -50,6 +56,7 @@ class LLM: "messages": messages, "temperature": self.caps.optimal_temperature, "api_key": self.api_key, + "timeout": _REQUEST_TIMEOUT_S, } if self.api_base: kwargs["api_base"] = self.api_base diff --git a/core/proc_wrapper.py b/core/proc_wrapper.py new file mode 100644 index 0000000..c4dd21e --- /dev/null +++ b/core/proc_wrapper.py @@ -0,0 +1,117 @@ +"""bg proc wrapper:被 detach 启动的独立进程,负责跑子进程 + 写 exit_code。 + +协议见 core/procs.py。设计约束: +- **stdlib-only、无 zcbot import**:用 `python core/proc_wrapper.py ` + 直跑,不依赖 PYTHONPATH / 虚拟环境激活(解释器路径由 launch_host 用 + sys.executable 定死,与 web 进程同一个 venv) +- 自身 stdout/stderr 已被 launch_host 接到 DEVNULL,不打印任何东西; + 所有输出走 output.log(bytes 写,绕开 Windows GBK console 编码问题) +- zcbot 重启/蓝绿切换不影响本进程(detach + 新 session/进程组) + +流程:读 proc.json → 起子进程(stdout+stderr → output.log)→ 回写 child_pid +→ wait(timeout_s) → 超时杀进程树记 124 → 截断日志到最后 10MB → 写 exit_code。 +exit_code 文件的出现是唯一"已结束"信号,必须最后写。 +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +MAX_LOG_BYTES = 10 * 1024 * 1024 + + +def _read_meta(d: Path) -> dict: + return json.loads((d / "proc.json").read_text(encoding="utf-8")) + + +def _write_meta(d: Path, meta: dict) -> None: + tmp = d / "proc.json.tmp" + tmp.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") + os.replace(tmp, d / "proc.json") + + +def _kill_tree(child: subprocess.Popen) -> None: + if os.name == "posix": + import signal + try: + os.killpg(child.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + try: + child.kill() + except OSError: + pass + else: + subprocess.run( + ["taskkill", "/PID", str(child.pid), "/T", "/F"], + capture_output=True, + ) + + +def _truncate_log(p: Path, cap: int = MAX_LOG_BYTES) -> None: + try: + size = p.stat().st_size + if size <= cap: + return + with open(p, "rb") as f: + f.seek(size - cap) + data = f.read() + with open(p, "wb") as f: + f.write(b"[proc_wrapper] log truncated to last %d bytes\n" % cap) + f.write(data) + except OSError: + pass + + +def main() -> int: + d = Path(sys.argv[1]) + meta = _read_meta(d) + timeout_s = int(meta.get("timeout_s") or 7200) + log = open(d / "output.log", "ab", buffering=0) + + spawn_kw: dict = dict( + cwd=meta.get("cwd") or None, + stdin=subprocess.DEVNULL, + stdout=log, + stderr=subprocess.STDOUT, + ) + # 子进程独立进程组:超时/kill 时能整树带走(posix killpg / win taskkill /T) + if os.name == "posix": + spawn_kw["start_new_session"] = True + else: + spawn_kw["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + + try: + if meta.get("argv"): + child = subprocess.Popen(meta["argv"], **spawn_kw) + else: + child = subprocess.Popen(meta["shell_cmd"], shell=True, **spawn_kw) + except Exception as e: # 启动失败也要写 exit_code,否则状态永远悬着 + log.write(f"[proc_wrapper] spawn failed: {type(e).__name__}: {e}\n".encode("utf-8")) + log.close() + (d / "exit_code").write_text("127", encoding="utf-8") + return 127 + + meta["child_pid"] = child.pid + _write_meta(d, meta) + + try: + ec = child.wait(timeout=timeout_s) + except subprocess.TimeoutExpired: + _kill_tree(child) + try: + child.wait(timeout=10) + except subprocess.TimeoutExpired: + pass + log.write(b"\n[proc_wrapper] timeout after %ds, process tree killed\n" % timeout_s) + ec = 124 + log.close() + _truncate_log(d / "output.log") + (d / "exit_code").write_text(str(ec), encoding="utf-8") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/core/procs.py b/core/procs.py new file mode 100644 index 0000000..63cdd69 --- /dev/null +++ b/core/procs.py @@ -0,0 +1,429 @@ +"""后台进程(bg proc):detach + 文件系统状态,不引队列组件(DESIGN §8.12)。 + +解决的问题:模型写的长脚本(数据处理/模拟计算)在工具内同步跑会被超时掐死、 +占死 run、蓝绿切换时随实例进程陪葬。方案是把进程从 zcbot 进程树上摘下来: + +- host 模式:detach 启动 `core/proc_wrapper.py`(stdlib-only),wrapper 负责跑 + 子进程、限时、写退出码 —— zcbot 重启不影响它 +- docker 模式:每个 bg proc 一个专用容器 `zcbot-proc-`(executor_docker 侧 + 实现),dockerd 托管,与 sandbox 容器的 idle reaper / shutdown_all 生命周期解耦 + +目录布局(dotfile,/v1/files API 天然隐藏,用户文件浏览器不可见): + /.zcbot_procs/// + proc.json # 元数据(kind/command/backend/pid 或 container/timeout/created_at) + output.log # stdout+stderr 合流(结束时截到最后 10MB) + exit_code # 结束后由 wrapper(host)/runner.sh(docker)写入 —— 存在即已结束 + script.py # run_python inline code 落盘(可选) + cmd.sh # docker 模式的用户命令(可选,绕开引号转义) + +状态判定不靠常驻登记,全靠文件 + 系统探测(list/check 时现算): + exit_code 存在 → finished;否则 host 查 wrapper pid 活着 / docker 查容器 + running → running;都探不到 → lost(宿主重启把进程带走了,产物文件仍在)。 + +边界(DESIGN §8.12,防滑坡):只覆盖「单个本地长进程」。job 链/依赖/重试策略 +不进这里 —— 编排的唯一归属是 agent loop 本身(模型在 check 后自己决定下一步)。 +""" +from __future__ import annotations + +import json +import os +import re +import secrets +import shutil +import subprocess +import sys +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +PROCS_SUBDIR = ".zcbot_procs" + +# bg 进程默认/最大运行时长(秒);前台 60/120s 的默认值不放大 —— 它是逼模型 +# 在"快命令"与"长任务"之间做选择的杠杆 +DEFAULT_TIMEOUT_S = 7200 +MAX_TIMEOUT_S = 86400 + +# 每用户并发 bg 进程上限(跨 task 合计);防模型失控起一堆 +MAX_RUNNING_PER_USER = int(os.getenv("ZCBOT_MAX_BG_PROCS", "3")) + +# 终态 proc 目录保留时长,sweep 超期删除 +FINISHED_TTL_S = 7 * 86400 + +_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") + + +def new_proc_id() -> str: + return "p" + secrets.token_hex(4) + + +def procs_root(anchor: Path) -> Path: + """anchor = user_root(常规)或 base_dir(CLI --working-dir 无 user_root 兜底)。""" + return Path(anchor) / PROCS_SUBDIR + + +def proc_dir(anchor: Path, task_id: str, proc_id: str) -> Optional[Path]: + """proc_id 来自模型输入,先校验再拼路径(防 ../ 越界)。非法返 None。""" + if not (_ID_RE.match(str(task_id) or "") and _ID_RE.match(proc_id or "")): + return None + return procs_root(anchor) / str(task_id) / proc_id + + +def write_meta(d: Path, meta: Dict[str, Any]) -> None: + """原子写(tmp + replace):wrapper 与 check_process 并发读写不撕裂。""" + tmp = d / "proc.json.tmp" + tmp.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") + os.replace(tmp, d / "proc.json") + + +def read_meta(d: Path) -> Optional[Dict[str, Any]]: + try: + return json.loads((d / "proc.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def clamp_timeout(timeout: Optional[int]) -> int: + try: + t = int(timeout) if timeout else DEFAULT_TIMEOUT_S + except (TypeError, ValueError): + t = DEFAULT_TIMEOUT_S + return max(60, min(t, MAX_TIMEOUT_S)) + + +def count_running(anchor: Path) -> int: + """该用户(anchor=user_root)当前 running 状态的 bg 进程数,跨 task 合计。""" + root = procs_root(anchor) + if not root.is_dir(): + return 0 + n = 0 + for tdir in root.iterdir(): + if not tdir.is_dir(): + continue + for d in tdir.iterdir(): + meta = read_meta(d) if d.is_dir() else None + if meta and status_of(meta, d)[0] == "running": + n += 1 + return n + + +# ───────────── 状态探测 ───────────── + +def _pid_alive(pid: int) -> bool: + if pid <= 0: + return False + if os.name == "posix": + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + # Windows:OpenProcess 探测(PID 复用误报概率低,可接受) + import ctypes + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + h = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not h: + return False + try: + code = ctypes.c_ulong() + ok = ctypes.windll.kernel32.GetExitCodeProcess(h, ctypes.byref(code)) + STILL_ACTIVE = 259 + return bool(ok) and code.value == STILL_ACTIVE + finally: + ctypes.windll.kernel32.CloseHandle(h) + + +def _container_running(name: str) -> bool: + try: + r = subprocess.run( + ["docker", "inspect", "--type=container", + "--format={{.State.Running}}", name], + capture_output=True, text=True, timeout=15, + ) + return r.returncode == 0 and r.stdout.strip() == "true" + except (OSError, subprocess.TimeoutExpired): + return False + + +def status_of(meta: Dict[str, Any], d: Path) -> Tuple[str, Optional[int]]: + """→ ("running"|"finished"|"lost", exit_code|None)。exit_code 文件是唯一终态锚。""" + try: + raw = (d / "exit_code").read_text(encoding="utf-8").strip() + return "finished", int(raw) + except (OSError, ValueError): + pass + if meta.get("backend") == "docker": + if _container_running(str(meta.get("container") or "")): + return "running", None + return "lost", None + pid = int(meta.get("wrapper_pid") or 0) + if _pid_alive(pid): + return "running", None + return "lost", None + + +def tail_log(d: Path, max_bytes: int = 8_000) -> str: + p = d / "output.log" + try: + size = p.stat().st_size + with open(p, "rb") as f: + if size > max_bytes: + f.seek(size - max_bytes) + data = f.read() + text = data.decode("utf-8", errors="replace") + if size > max_bytes: + text = f"[...前 {size - max_bytes} 字节省略...]\n" + text + return text + except OSError: + return "(暂无输出)" + + +def list_procs(anchor: Path, task_id: str) -> List[Dict[str, Any]]: + """该 task 的所有 bg proc,带现算 status/exit_code,按创建时间排。""" + tdir = procs_root(anchor) / str(task_id) + if not tdir.is_dir(): + return [] + out = [] + for d in sorted(tdir.iterdir()): + if not d.is_dir(): + continue + meta = read_meta(d) + if not meta: + continue + st, ec = status_of(meta, d) + meta["_status"], meta["_exit_code"], meta["_dir"] = st, ec, d + out.append(meta) + out.sort(key=lambda m: m.get("created_ts") or 0) + return out + + +def list_all_procs(anchor: Path) -> List[Dict[str, Any]]: + """该用户(anchor=user_root)全部 task 的 bg proc(web `/v1/procs` 用:跨 task + 通知——用户切到别的 task 也能收到完成提示)。""" + root = procs_root(anchor) + if not root.is_dir(): + return [] + out: List[Dict[str, Any]] = [] + for tdir in sorted(root.iterdir()): + if tdir.is_dir(): + out.extend(list_procs(anchor, tdir.name)) + return out + + +def finished_at(d: Path) -> Optional[float]: + """终态时刻 = exit_code 文件 mtime(结束时才写,天然就是完成时间戳)。""" + try: + return (d / "exit_code").stat().st_mtime + except OSError: + return None + + +# ───────────── host 模式启动 / 终止 ───────────── + +def launch_host( + anchor: Path, + task_id: str, + *, + kind: str, # "shell" | "python" + command_display: str, # 给 LLM/用户看的命令描述 + argv: Optional[List[str]] = None, # 二选一:argv(python) + shell_cmd: Optional[str] = None, # 或 shell 命令串 + cwd: Path, + timeout_s: int, + env: Optional[Dict[str, str]] = None, + inline_code: Optional[str] = None, # run_python inline → 落 script.py,argv 引用它 +) -> Tuple[str, Path]: + """detach 启动 wrapper,立即返回 (proc_id, proc_dir)。 + + wrapper 是独立脚本(core/proc_wrapper.py,stdlib-only),用当前解释器直跑 + (非 -m,不依赖 PYTHONPATH),zcbot 进程退出/重启不影响它。 + """ + proc_id = new_proc_id() + d = procs_root(anchor) / str(task_id) / proc_id + d.mkdir(parents=True, exist_ok=True) + + if inline_code is not None: + script = d / "script.py" + script.write_text(inline_code, encoding="utf-8") + argv = [sys.executable, str(script)] + + meta: Dict[str, Any] = { + "proc_id": proc_id, + "task_id": str(task_id), + "kind": kind, + "backend": "host", + "command": command_display[:500], + "cwd": str(cwd), + "timeout_s": timeout_s, + "created_at": datetime.now().isoformat(timespec="seconds"), + "created_ts": time.time(), + } + if argv: + meta["argv"] = argv + else: + meta["shell_cmd"] = shell_cmd + write_meta(d, meta) + + wrapper = Path(__file__).parent / "proc_wrapper.py" + spawn_kw: Dict[str, Any] = dict( + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=env, + ) + if os.name == "posix": + spawn_kw["start_new_session"] = True + else: + spawn_kw["creationflags"] = ( + subprocess.DETACHED_PROCESS + | subprocess.CREATE_NEW_PROCESS_GROUP + | subprocess.CREATE_NO_WINDOW + ) + p = subprocess.Popen([sys.executable, str(wrapper), str(d)], **spawn_kw) + meta["wrapper_pid"] = p.pid + write_meta(d, meta) + # 后台线程 wait 掉 wrapper,避免 POSIX 僵尸挂在 web 进程下 + threading.Thread(target=p.wait, daemon=True).start() + return proc_id, d + + +def kill_proc(meta: Dict[str, Any], d: Path) -> str: + """强杀。docker → rm -f 容器;host → taskkill /T(win)/ killpg(posix)。 + 杀完补写 exit_code=137,让状态收敛到 finished(killed)。""" + st, _ = status_of(meta, d) + if st == "finished": + return "进程已结束,无需终止" + if meta.get("backend") == "docker": + name = str(meta.get("container") or "") + if name: + subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30) + else: + wrapper_pid = int(meta.get("wrapper_pid") or 0) + child_pid = int(meta.get("child_pid") or 0) + if os.name == "posix": + import signal + for pid in (child_pid, wrapper_pid): + if pid > 0: + try: + os.killpg(pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + else: + for pid in (wrapper_pid, child_pid): + if pid > 0: + subprocess.run( + ["taskkill", "/PID", str(pid), "/T", "/F"], + capture_output=True, timeout=30, + ) + try: + (d / "exit_code").write_text("137", encoding="utf-8") + except OSError: + pass + meta["killed"] = True + write_meta(d, meta) + return "已终止" + + +# ───────────── 清扫(web lifespan 周期调用) ───────────── + +def sweep(user_root_base: Path, ttl_s: int = FINISHED_TTL_S) -> Dict[str, int]: + """两件事:① 终态超 TTL 的 proc 目录删掉 ② 已结束(exit_code 在)但容器还挂着 + 的 docker proc 容器 rm -f(runner 写完退出码后容器空转 sleep,等这里回收)。 + 另兜底:label=zcbot.product=proc 的孤儿容器(目录已删/exited)一并清。 + 幂等,蓝绿双实例同时跑无害(rm -f / rmtree 都幂等)。""" + removed_dirs = reaped_containers = 0 + now = time.time() + base = Path(user_root_base) + if base.is_dir(): + for uroot in base.iterdir(): + # 单条目异常(权限/竞态删除)不打断整轮清扫 + try: + root = uroot / PROCS_SUBDIR + if not root.is_dir(): + continue + for tdir in list(root.iterdir()): + if not tdir.is_dir(): + continue + for d in list(tdir.iterdir()): + if not d.is_dir(): + continue + meta = read_meta(d) + if not meta: + continue + st, _ = status_of(meta, d) + if st != "running" and meta.get("backend") == "docker": + name = str(meta.get("container") or "") + if name and _container_exists(name): + subprocess.run( + ["docker", "rm", "-f", name], + capture_output=True, timeout=30, + ) + reaped_containers += 1 + age = now - float(meta.get("created_ts") or now) + if st != "running" and age > ttl_s: + shutil.rmtree(d, ignore_errors=True) + removed_dirs += 1 + try: + tdir.rmdir() # 空了顺手收掉,非空抛 OSError 忽略 + except OSError: + pass + except OSError: + continue + # 孤儿容器兜底(proc 目录已被删,或容器已 exited) + try: + r = subprocess.run( + ["docker", "ps", "-aq", "--filter", "label=zcbot.product=proc", + "--filter", "status=exited"], + capture_output=True, text=True, timeout=30, + ) + ids = r.stdout.split() if r.returncode == 0 else [] + if ids: + subprocess.run(["docker", "rm", "-f", *ids], capture_output=True, timeout=60) + reaped_containers += len(ids) + except (OSError, subprocess.TimeoutExpired): + pass # docker 不在(host 模式部署)→ 只做文件清扫 + return {"removed_dirs": removed_dirs, "reaped_containers": reaped_containers} + + +def _container_exists(name: str) -> bool: + try: + r = subprocess.run( + ["docker", "inspect", "--type=container", name], + capture_output=True, timeout=15, + ) + return r.returncode == 0 + except (OSError, subprocess.TimeoutExpired): + return False + + +# ───────────── LLM 面向的描述格式化(check_process / 启动返回共用) ───────────── + +def format_proc_line(meta: Dict[str, Any]) -> str: + st = meta.get("_status", "?") + ec = meta.get("_exit_code") + elapsed = "" + ts = meta.get("created_ts") + if ts: + elapsed = f" · 已运行 {_fmt_elapsed(time.time() - float(ts))}" if st == "running" else "" + tail = f"(exit {ec})" if st == "finished" else "" + return ( + f"- {meta.get('proc_id')} [{st}{tail}] {meta.get('kind')}: " + f"{meta.get('command', '')[:120]}{elapsed}" + ) + + +def _fmt_elapsed(s: float) -> str: + s = int(s) + if s < 60: + return f"{s}s" + if s < 3600: + return f"{s // 60}m{s % 60:02d}s" + return f"{s // 3600}h{(s % 3600) // 60:02d}m" diff --git a/core/sandbox/pool.py b/core/sandbox/pool.py index f81cd5c..071aae7 100644 --- a/core/sandbox/pool.py +++ b/core/sandbox/pool.py @@ -43,6 +43,12 @@ from .network import NETWORK_NAME, ensure_network CONTAINER_NAME_PREFIX = "zcbot-sandbox-" LABEL_PRODUCT_KEY = "zcbot.product" LABEL_PRODUCT_VALUE = "sandbox" +# bg proc 专用容器(DESIGN §8.12):product=proc + 不带 instance label ── +# 与 sandbox 容器的 idle reaper / shutdown_all(按 product=sandbox 过滤)生命周期 +# 解耦,蓝绿切换/重启都不会误杀;回收由 core/procs.py::sweep 负责 +LABEL_PRODUCT_PROC = "proc" +LABEL_PROC_DIR_KEY = "zcbot.proc_dir" +PROC_CONTAINER_PREFIX = "zcbot-proc-" LABEL_USER_ID_KEY = "zcbot.user_id" LABEL_INSTANCE_KEY = "zcbot.instance" @@ -199,18 +205,46 @@ class SandboxPool: return None return resolv - def _docker_run(self, user_id: UUID, name: str) -> None: - """同步阻塞;由 ensure 在 to_thread 里调。""" + def run_proc_container(self, user_id: UUID, proc_id: str, proc_dir: str) -> str: + """bg proc 专用容器:与 sandbox 同款硬化(iptables init / read-only / 资源限额), + 但 product=proc + 无 instance label → reaper/shutdown_all 都不碰它,dockerd 托管, + zcbot 重启与蓝绿切换不中断。回收:procs.sweep / check_process 见 exit_code 后 rm -f。""" + name = f"{PROC_CONTAINER_PREFIX}{proc_id}" + self._docker_run( + user_id, name, + product=LABEL_PRODUCT_PROC, + extra_labels={LABEL_PROC_DIR_KEY: proc_dir}, + ) + return name + + def _docker_run( + self, + user_id: UUID, + name: str, + *, + product: str = LABEL_PRODUCT_VALUE, + extra_labels: Optional[Dict[str, str]] = None, + ) -> None: + """同步阻塞;由 ensure 在 to_thread 里调。instance label 只挂 sandbox 容器 + (proc 容器要跨实例存活,不能被任何实例的 shutdown_all 认领)。""" user_root = self.user_root_base / str(user_id) user_root.mkdir(parents=True, exist_ok=True) resolv_file = self._ensure_resolv_conf_file() + is_sandbox = product == LABEL_PRODUCT_VALUE + label_args: List[str] = [ + "--label", f"{LABEL_PRODUCT_KEY}={product}", + "--label", f"{LABEL_USER_ID_KEY}={user_id}", + ] + if INSTANCE and is_sandbox: + label_args += ["--label", f"{LABEL_INSTANCE_KEY}={INSTANCE}"] + for k, v in (extra_labels or {}).items(): + label_args += ["--label", f"{k}={v}"] + cmd: List[str] = [ "docker", "run", "-d", "--name", name, - "--label", f"{LABEL_PRODUCT_KEY}={LABEL_PRODUCT_VALUE}", - "--label", f"{LABEL_USER_ID_KEY}={user_id}", - *(["--label", f"{LABEL_INSTANCE_KEY}={INSTANCE}"] if INSTANCE else []), + *label_args, "--network", NETWORK_NAME, # §7.5 硬限制(任一缺失视为 hardening 未完成) "--read-only", # rootfs read-only diff --git a/scripts/test_bgproc_manual.py b/scripts/test_bgproc_manual.py new file mode 100644 index 0000000..f24de9d --- /dev/null +++ b/scripts/test_bgproc_manual.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +"""bg proc host 模式手工验证(ASCII-only 输出,GBK console 安全)。 + +场景: +1. shell background 启动一个 ~6s 的命令 -> 立即返回 +2. 期间 status = running +3. 结束后 status = finished, exit_code=0, 日志内容对 +4. inline python background + kill +5. count_running / list_procs / sweep 冒烟 +""" +import shutil +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from core import procs # noqa: E402 + +anchor = Path(tempfile.mkdtemp(prefix="zcbot_bgproc_test_")) +task_id = "11111111-1111-1111-1111-111111111111" +fails = [] + + +def check(name, cond): + print(("[ok] " if cond else "[FAIL] ") + name) + if not cond: + fails.append(name) + + +# 1. python inline bg: 3s 后打印再退出 +code = "import time\nprint('phase1', flush=True)\ntime.sleep(3)\nprint('phase2 done', flush=True)\n" +t0 = time.time() +pid1, d1 = procs.launch_host( + anchor, task_id, kind="python", command_display="python ", + cwd=anchor, timeout_s=120, env=None, inline_code=code, + argv=None, shell_cmd=None, +) +launch_elapsed = time.time() - t0 +check("launch returns fast (<2s), got %.2fs" % launch_elapsed, launch_elapsed < 2) + +time.sleep(1.5) +meta1 = procs.read_meta(d1) +st, ec = procs.status_of(meta1, d1) +check("running mid-flight (got %s)" % st, st == "running") +check("count_running == 1 (got %d)" % procs.count_running(anchor), procs.count_running(anchor) == 1) + +for _ in range(20): + time.sleep(0.5) + st, ec = procs.status_of(procs.read_meta(d1), d1) + if st == "finished": + break +check("finished with exit 0 (got %s/%s)" % (st, ec), st == "finished" and ec == 0) +log = procs.tail_log(d1) +check("log has phase2 output", "phase2 done" in log) + +# 2. timeout kill: sleep 60 but timeout_s=61 clamp floor -> use argv sleep via python +code2 = "import time\ntime.sleep(600)\n" +pid2, d2 = procs.launch_host( + anchor, task_id, kind="python", command_display="python ", + cwd=anchor, timeout_s=120, env=None, inline_code=code2, + argv=None, shell_cmd=None, +) +time.sleep(1.5) +meta2 = procs.read_meta(d2) +st2, _ = procs.status_of(meta2, d2) +check("sleeper running", st2 == "running") +msg = procs.kill_proc(meta2, d2) +time.sleep(1.0) +st2b, ec2 = procs.status_of(procs.read_meta(d2), d2) +check("killed -> finished 137 (got %s/%s, msg=%s)" % (st2b, ec2, msg), + st2b == "finished" and ec2 == 137) + +# 3. list + format +items = procs.list_procs(anchor, task_id) +check("list_procs sees 2 (got %d)" % len(items), len(items) == 2) +for m in items: + print(" " + procs.format_proc_line(m)) + +# 4. sweep: nothing removed (fresh), then with ttl=0 removes finished dirs +stats = procs.sweep(anchor.parent / "_nonexistent_users") +stats2 = procs.sweep(anchor.parent, ttl_s=10 ** 9) # anchor.parent as users base: anchor is one 'user' +# note: sweep expects user_root_base -> iterates children; anchor.parent contains anchor +check("sweep noop keeps dirs", (anchor / procs.PROCS_SUBDIR / task_id).is_dir()) +stats3 = procs.sweep(anchor.parent, ttl_s=0) +remaining = list((anchor / procs.PROCS_SUBDIR / task_id).glob("*")) if (anchor / procs.PROCS_SUBDIR / task_id).is_dir() else [] +check("sweep ttl=0 removed finished dirs (left %d)" % len(remaining), len(remaining) == 0) + +shutil.rmtree(anchor, ignore_errors=True) +print() +if fails: + print("RESULT: %d FAILURE(S): %s" % (len(fails), "; ".join(fails))) + sys.exit(1) +print("RESULT: ALL PASS") diff --git a/tools/check_process.py b/tools/check_process.py new file mode 100644 index 0000000..7e1f858 --- /dev/null +++ b/tools/check_process.py @@ -0,0 +1,118 @@ +"""check_process: 查看/终止 shell·run_python `background=true` 启动的后台进程。 + +host in-process 工具(不进 CONTAINER_TOOLS):状态文件在宿主侧 user_root/.zcbot_procs, +docker 模式的容器探测/回收走宿主 docker CLI —— 两种 backend 一个工具通吃。 +""" +from __future__ import annotations + +import subprocess +import time +from pathlib import Path +from typing import Optional + +from core import procs + +from .base import Tool, compact_tool_output + + +class CheckProcessTool(Tool): + name = "check_process" + description = ( + "Check status / output of background processes started with background=true " + "(shell or run_python), or kill one. " + "Call without proc_id to list this task's background processes. " + "With proc_id: returns status + tail of the process log. " + "If still running: do NOT poll in a tight loop — do other useful work, or end " + "the turn and tell the user to ask later ('跑完了吗'). " + "action='kill' force-terminates the process." + ) + parameters = { + "type": "object", + "properties": { + "proc_id": { + "type": "string", + "description": "后台进程 id(启动时返回的 p 开头短串)。缺省 = 列出本 task 全部后台进程。", + }, + "action": { + "type": "string", + "enum": ["status", "kill"], + "description": "status(默认)= 查状态+日志尾部;kill = 强制终止该进程。", + }, + "tail_bytes": { + "type": "integer", + "description": "返回日志尾部的最大字节数,默认 4000,上限 20000。", + }, + }, + "required": [], + } + + def __init__( + self, + *, + task_id: str, + base_dir: Optional[Path] = None, + user_root: Optional[Path] = None, + ) -> None: + super().__init__(base_dir, user_root=user_root) + self.task_id = str(task_id) + + @property + def _anchor(self) -> Path: + # 常规部署有 user_root;CLI --working-dir 指外部目录时退 base_dir + return self.user_root or self.base_dir + + def execute( + self, + proc_id: Optional[str] = None, + action: str = "status", + tail_bytes: int = 4000, + ) -> str: + if not proc_id: + items = procs.list_procs(self._anchor, self.task_id) + if not items: + return "本 task 没有后台进程。用 shell/run_python 的 background=true 可启动。" + return "后台进程列表:\n" + "\n".join(procs.format_proc_line(m) for m in items) + + d = procs.proc_dir(self._anchor, self.task_id, proc_id.strip()) + if d is None or not d.is_dir(): + return f"[Error] 后台进程不存在: {proc_id!r}(用 check_process 不带参数列出现有进程)" + meta = procs.read_meta(d) + if meta is None: + return f"[Error] 后台进程元数据损坏: {proc_id!r}" + + if action == "kill": + msg = procs.kill_proc(meta, d) + return f"[check_process] {proc_id}: {msg}" + + st, ec = procs.status_of(meta, d) + tail_bytes = max(500, min(int(tail_bytes or 4000), 20_000)) + tail = procs.tail_log(d, max_bytes=tail_bytes) + + # docker 模式:已结束但专用容器还在空转 → 顺手回收(幂等,sweep 也会兜底) + if st != "running" and meta.get("backend") == "docker": + name = str(meta.get("container") or "") + if name: + try: + subprocess.run(["docker", "rm", "-f", name], + capture_output=True, timeout=30) + except (OSError, subprocess.TimeoutExpired): + pass + + header = f"[check_process] {proc_id} · {meta.get('kind')} · {meta.get('command', '')[:150]}" + if st == "running": + elapsed = procs._fmt_elapsed(time.time() - float(meta.get("created_ts") or time.time())) + body = ( + f"状态: running(已运行 {elapsed},上限 {meta.get('timeout_s')}s)\n" + f"--- 日志尾部 ---\n{tail}\n" + f"提示:仍在运行。别循环轮询 —— 先做别的工作,或结束回合告知用户稍后询问进度。" + ) + elif st == "finished": + note = " (timeout killed)" if ec == 124 else (" (killed)" if meta.get("killed") else "") + body = f"状态: finished, exit_code={ec}{note}\n--- 日志尾部 ---\n{tail}" + else: + body = ( + f"状态: lost(进程已不在,且未留下退出码 —— 可能宿主重启把它带走了)\n" + f"--- 日志尾部(截至中断) ---\n{tail}\n" + f"如需结果请重新以 background=true 启动。" + ) + return compact_tool_output(f"{header}\n{body}") diff --git a/tools/run_python.py b/tools/run_python.py index 32c3e4c..616793b 100644 --- a/tools/run_python.py +++ b/tools/run_python.py @@ -13,6 +13,9 @@ import subprocess import sys import tempfile from pathlib import Path +from typing import Optional + +from core import procs from .base import Tool, compact_tool_output @@ -47,17 +50,87 @@ class RunPythonTool(Tool): "type": "string", "description": "Path to an existing .py file to execute (relative to task_dir). Prefer this for non-trivial code; keep such scripts under scripts/.", }, - "timeout": {"type": "integer", "default": 120, "description": "Seconds before kill"}, + "timeout": {"type": "integer", "default": 120, "description": "Seconds before kill. background=true 时含义变为最长运行时长,默认 7200s。"}, + "background": { + "type": "boolean", + "description": ( + "true = 后台执行:立即返回 proc_id,进程 detach 独立运行(不受本轮对话、" + "服务重启影响),用 check_process 查进度/结果。适用:预计运行超过约 1 分钟的" + "计算/训练/批处理。快速脚本保持默认 false(同步等结果)。前台超时被杀的" + "长任务应改用 background=true 重新发起。" + ), + }, }, "required": [], } + def __init__( + self, + base_dir: Optional[Path] = None, + user_root: Optional[Path] = None, + task_id: Optional[str] = None, + ) -> None: + super().__init__(base_dir, user_root=user_root) + self.task_id = str(task_id) if task_id else "default" + + def _filtered_env(self) -> dict: + env = os.environ.copy() + for k in list(env): + u = k.upper() + if u not in _ENV_ALLOWLIST and any(p in u for p in _SENSITIVE_PATTERNS): + del env[k] + env["PYTHONIOENCODING"] = "utf-8" + env["PYTHONPATH"] = str(self.base_dir) + os.pathsep + env.get("PYTHONPATH", "") + return env + + def _execute_background( + self, code: str | None, script_path: str | None, timeout: int | None + ) -> str: + anchor = self.user_root or self.base_dir + if procs.count_running(anchor) >= procs.MAX_RUNNING_PER_USER: + return ( + f"[Error] 已有 {procs.MAX_RUNNING_PER_USER} 个后台进程在跑(上限)。" + f"用 check_process 查看,等待完成或 kill 掉不需要的再启动。" + ) + timeout_s = procs.clamp_timeout(timeout if timeout and timeout > 120 else None) + if script_path: + script = self._resolve(script_path) + if not script.is_file(): + return f"[Error] script_path not found: {self._display(script)}" + proc_id, _ = procs.launch_host( + anchor, self.task_id, + kind="python", + command_display=f"python {self._display(script)}", + argv=[sys.executable, str(script)], + cwd=self.base_dir, timeout_s=timeout_s, env=self._filtered_env(), + ) + shown = self._display(script) + elif isinstance(code, str) and code.strip(): + proc_id, _ = procs.launch_host( + anchor, self.task_id, + kind="python", + command_display=f"python ", + cwd=self.base_dir, timeout_s=timeout_s, env=self._filtered_env(), + inline_code=code, + ) + shown = "" + else: + return "[Error] run_python requires code or script_path" + return ( + f"[Background] 已启动后台进程 proc_id={proc_id}({shown}),最长运行 {timeout_s}s。\n" + f"用 check_process(proc_id=\"{proc_id}\") 查进度和日志。进程独立于本轮对话运行," + f"服务重启也不中断。现在可以继续其他工作;若无事可做,结束回合并告知用户稍后询问进度。" + ) + def execute( self, code: str | None = None, script_path: str | None = None, timeout: int = 120, + background: bool = False, ) -> str: + if background: + return self._execute_background(code, script_path, timeout) cleanup_script = False if script_path: script = self._resolve(script_path) @@ -75,14 +148,7 @@ class RunPythonTool(Tool): return "[Error] run_python requires code or script_path" try: - env = os.environ.copy() - for k in list(env): - u = k.upper() - if u not in _ENV_ALLOWLIST and any(p in u for p in _SENSITIVE_PATTERNS): - del env[k] - env["PYTHONIOENCODING"] = "utf-8" - env["PYTHONPATH"] = str(self.base_dir) + os.pathsep + env.get("PYTHONPATH", "") - + env = self._filtered_env() result = subprocess.run( [sys.executable, str(script)], cwd=str(self.base_dir), @@ -94,7 +160,10 @@ class RunPythonTool(Tool): env=env, ) except subprocess.TimeoutExpired: - return f"[Error] python script timed out after {timeout}s" + return ( + f"[Error] python script timed out after {timeout}s. " + f"若任务本身需要长时间运行,用 background=true 重新发起(后台执行,check_process 查进度)。" + ) finally: if cleanup_script: try: diff --git a/tools/seedance.py b/tools/seedance.py index 77427a9..e5a3adc 100644 --- a/tools/seedance.py +++ b/tools/seedance.py @@ -109,6 +109,14 @@ class SeedanceTool(Tool): "广告 / 短剧 / 角色对白等场景传 true,模型会一并算音轨,cost 高于纯视频。" ), }, + "resume_task_id": { + "type": "string", + "description": ( + "续查:之前提交但轮询超时/被中断的任务 id(cgt-开头,错误信息里有)。" + "传了则跳过提交,直接继续等待并取回结果(24h 内有效,不重复计费)。" + "prompt 等参数请带上与原次相同的值(用于产物元数据)。" + ), + }, }, "required": ["prompt"], } @@ -145,15 +153,18 @@ class SeedanceTool(Tool): duration: Optional[int] = None, watermark: Optional[bool] = None, generate_audio: Optional[bool] = None, + resume_task_id: Optional[str] = None, ) -> str: if not (prompt or "").strip(): return "[Error] prompt 不能为空" + resume_id = (resume_task_id or "").strip() # 每账号每日配额(yaml quotas.videos_per_day)。失败 / cancel 不计,因为 # record_video_usage 只在 succeeded+下载完才落库。tool 返串会进 LLM 上下文 # → 模型据此向用户解释,所以**只暴露用户该看的部分**(已用/上限 + 重置时间), # 内部 yaml 路径不进对话(管理员要改的地方读代码/yaml 自己找)。 - if self.daily_limit > 0: + # resume 不过配额闸:原次提交已经占过额度,续查不是新生成。 + if self.daily_limit > 0 and not resume_id: used, over = check_daily_quota(user_id=self.user_id, kind="video", limit=self.daily_limit) if over: return ( @@ -193,11 +204,15 @@ class SeedanceTool(Tool): t0 = time.monotonic() try: with ArkClient(self.ark_cfg, timeout_s=submit_timeout) as client: - # 1. submit - submit_resp = client.post_json(submit_endpoint, body, timeout_s=submit_timeout) - cgt_id = self._extract_task_id(submit_resp) - if not cgt_id: - return f"[Error] seedance submit 响应缺 task id: {json.dumps(submit_resp, ensure_ascii=False)[:300]}" + # 1. submit(resume 时跳过 —— 拿着原 cgt_id 直接进轮询,远端任务 + # 24h 内结果都在,不重复提交不重复计费) + if resume_id: + cgt_id = resume_id + else: + submit_resp = client.post_json(submit_endpoint, body, timeout_s=submit_timeout) + cgt_id = self._extract_task_id(submit_resp) + if not cgt_id: + return f"[Error] seedance submit 响应缺 task id: {json.dumps(submit_resp, ensure_ascii=False)[:300]}" # 2. poll poll_url = f"{poll_endpoint_base}/{cgt_id}" @@ -207,13 +222,15 @@ class SeedanceTool(Tool): while True: if self.cancel_check is not None and self.cancel_check(): return ( - f"[Cancelled] seedance task {cgt_id} 用户取消(远端任务可能仍在跑;" + f"[Cancelled] seedance task {cgt_id} 等待被中断(远端任务仍在跑," + f"24h 内可用 resume_task_id=\"{cgt_id}\" 继续等待/取回结果,不重复计费;" f"Volcengine 失败/成功才计费,若仍出片可能产生 ~¥{self._rough_cost(chosen_resolution, chosen_ratio, chosen_duration, fps, price_t2v):.2f})" ) if time.monotonic() > deadline: return ( f"[Error] seedance 轮询超时(>{poll_timeout:.0f}s),最后 status={last_status!r}," - f"cgt_id={cgt_id}(24h 内可手工 GET {poll_url} 拉结果)" + f"cgt_id={cgt_id}。远端任务未取消,24h 内可用 " + f"resume_task_id=\"{cgt_id}\" 再次调用本工具继续等待/取回结果(不重复计费)。" ) time.sleep(poll_interval) poll_resp = client.get_json(poll_url, timeout_s=submit_timeout) diff --git a/tools/shell.py b/tools/shell.py index ba401d3..964f05c 100644 --- a/tools/shell.py +++ b/tools/shell.py @@ -6,6 +6,10 @@ import re import shlex import subprocess import sys +from pathlib import Path +from typing import Optional + +from core import procs from .base import Tool, compact_tool_output @@ -14,17 +18,36 @@ class ShellTool(Tool): name = "shell" description = ( "Execute a shell command and return stdout/stderr/exit_code. " - "Default 60s timeout. Working directory is the agent's base dir." + "Default 60s timeout. Working directory is the agent's base dir. " + "For long-running commands (expected >~1 min), set background=true and " + "poll with check_process." ) parameters = { "type": "object", "properties": { "command": {"type": "string"}, - "timeout": {"type": "integer", "default": 60, "description": "Seconds before kill"}, + "timeout": {"type": "integer", "default": 60, "description": "Seconds before kill. background=true 时含义变为最长运行时长,默认 7200s。"}, + "background": { + "type": "boolean", + "description": ( + "true = 后台执行:立即返回 proc_id,进程 detach 独立运行(不受本轮对话、" + "服务重启影响),用 check_process 查进度/结果。适用:预计运行超过约 1 分钟的" + "命令(编译/批处理/长计算)。快命令保持默认 false。" + ), + }, }, "required": ["command"], } + def __init__( + self, + base_dir: Optional[Path] = None, + user_root: Optional[Path] = None, + task_id: Optional[str] = None, + ) -> None: + super().__init__(base_dir, user_root=user_root) + self.task_id = str(task_id) if task_id else "default" + BLOCKED_PATTERNS = ( "rm -rf /", "rm -rf ~", @@ -58,7 +81,7 @@ class ShellTool(Tool): ) return command, None - def execute(self, command: str, timeout: int = 60) -> str: + def execute(self, command: str, timeout: int = 60, background: bool = False) -> str: normalized = command.lower() for pat in self.BLOCKED_PATTERNS: if pat in normalized: @@ -66,6 +89,27 @@ class ShellTool(Tool): command, note = self._windows_compat(command) + if background: + anchor = self.user_root or self.base_dir + if procs.count_running(anchor) >= procs.MAX_RUNNING_PER_USER: + return ( + f"[Error] 已有 {procs.MAX_RUNNING_PER_USER} 个后台进程在跑(上限)。" + f"用 check_process 查看,等待完成或 kill 掉不需要的再启动。" + ) + timeout_s = procs.clamp_timeout(timeout if timeout and timeout > 60 else None) + proc_id, _ = procs.launch_host( + anchor, self.task_id, + kind="shell", + command_display=command, + shell_cmd=command, + cwd=self.base_dir, timeout_s=timeout_s, env=None, + ) + return ( + f"[Background] 已启动后台进程 proc_id={proc_id},最长运行 {timeout_s}s。\n" + f"用 check_process(proc_id=\"{proc_id}\") 查进度和日志。进程独立于本轮对话运行," + f"服务重启也不中断。现在可以继续其他工作;若无事可做,结束回合并告知用户稍后询问进度。" + ) + try: result = subprocess.run( command, @@ -78,7 +122,10 @@ class ShellTool(Tool): errors="replace", ) except subprocess.TimeoutExpired: - return f"[Error] command timed out after {timeout}s" + return ( + f"[Error] command timed out after {timeout}s. " + f"若命令本身需要长时间运行,用 background=true 重新发起(后台执行,check_process 查进度)。" + ) except FileNotFoundError as e: return f"[Error] {e}" diff --git a/web/app.py b/web/app.py index 4c4e721..fed70b5 100644 --- a/web/app.py +++ b/web/app.py @@ -17,6 +17,7 @@ import mimetypes import os import re import tempfile +import time from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager from datetime import datetime as _dt @@ -1268,6 +1269,32 @@ def create_app() -> FastAPI: raise RuntimeError( f"ZCBOT_SANDBOX_BACKEND=docker but sandbox init failed: {e}" ) + + # bg proc 清扫(DESIGN §8.12):终态 proc 目录过 TTL 删除 + 已结束的 + # zcbot-proc-* 容器回收。host / docker 两种 backend 都要跑(host 模式只做 + # 文件清扫,docker CLI 不在时 sweep 内部静默跳过容器部分)。每小时一次; + # 幂等,蓝绿双实例同时跑无害。启动后先跑一轮,把上个进程周期留下的 + # 已结束容器/过期目录收掉。 + from core.procs import sweep as _procs_sweep + _procs_users_base = resolve_workspace(None, _cfg) / "users" + + async def _proc_sweeper() -> None: + loop = asyncio.get_running_loop() + while True: + try: + stats = await loop.run_in_executor( + None, _procs_sweep, _procs_users_base + ) + if stats["removed_dirs"] or stats["reaped_containers"]: + print(f"[proc-sweep] dirs={stats['removed_dirs']} " + f"containers={stats['reaped_containers']}") + except asyncio.CancelledError: + raise + except Exception as e: + print(f"[proc-sweep] error: {type(e).__name__}: {e}") + await asyncio.sleep(3600) + + proc_sweeper_task = asyncio.create_task(_proc_sweeper(), name="proc-sweeper") try: yield finally: @@ -1326,6 +1353,11 @@ def create_app() -> FastAPI: await sandbox_reaper_task except (asyncio.CancelledError, Exception): pass + proc_sweeper_task.cancel() + try: + await proc_sweeper_task + except (asyncio.CancelledError, Exception): + pass if sandbox_backend == "docker": pool = getattr(app.state, "sandbox_pool", None) if pool is not None: @@ -2820,6 +2852,71 @@ def create_app() -> FastAPI: broker.request_cancel(tid) return {"ok": True, "task_id": str(tid), "run_status": "cancelling"} + # ───────────── Background procs(bg proc,DESIGN §8.12)───────────── + + def _proc_view(m: dict) -> dict: + """core.procs.list_* 的条目 → API 形态。elapsed:running 现算,finished 用 + exit_code mtime 定格(前端秒数展示与工具卡同体验)。""" + from core import procs as _procs + d = m.get("_dir") + st, ec = m.get("_status"), m.get("_exit_code") + created = float(m.get("created_ts") or 0) or None + now = time.time() + if st == "finished" and d is not None: + fin = _procs.finished_at(d) or created or now + elapsed = max(0.0, fin - (created or fin)) + else: + elapsed = max(0.0, now - created) if created else 0.0 + return { + "task_id": m.get("task_id"), + "proc_id": m.get("proc_id"), + "kind": m.get("kind"), + "command": m.get("command") or "", + "status": st, + "exit_code": ec, + "killed": bool(m.get("killed")), + "created_at": m.get("created_at"), + "elapsed_s": int(elapsed), + "timeout_s": m.get("timeout_s"), + } + + @app.get("/v1/procs", tags=["tasks"]) + async def list_user_procs(user_id: UUID = Depends(require_user)): + """当前用户全部 task 的后台进程(shell/run_python background=true 启动)。 + + 用户级而非 task 级:前端全局轮询,切到别的 task 也能收到"后台任务完成"提示。 + 纯文件系统读取(user_root/.zcbot_procs),无 DB;docker backend 的 running + 探测走 docker inspect,放 to_thread 防塞 event loop。 + """ + anchor = _load_user_root(user_id) + from core import procs as _procs + items = await asyncio.to_thread(_procs.list_all_procs, anchor) + return {"procs": [_proc_view(m) for m in items]} + + @app.post("/v1/tasks/{task_id}/procs/{proc_id}/kill", tags=["tasks"]) + async def kill_task_proc( + task_id: str, + proc_id: str, + user_id: UUID = Depends(require_user), + ): + """强制终止一个后台进程(host 杀进程树 / docker rm -f 容器)。幂等:已结束返 ok。""" + try: + tid = UUID(task_id) + except ValueError: + raise HTTPException(404, f"invalid task id: {task_id!r}") + with session_scope() as s: + _assert_owns_task(s, tid, user_id) + anchor = _load_user_root(user_id) + from core import procs as _procs + d = _procs.proc_dir(anchor, str(tid), proc_id) + if d is None or not d.is_dir(): + raise HTTPException(404, f"background proc not found: {proc_id!r}") + meta = _procs.read_meta(d) + if meta is None: + raise HTTPException(404, f"background proc metadata missing: {proc_id!r}") + msg = await asyncio.to_thread(_procs.kill_proc, meta, d) + return {"ok": True, "proc_id": proc_id, "message": msg} + # ───────────── Clear conversation ───────────── @app.post("/v1/tasks/{task_id}/clear", tags=["messages"]) diff --git a/web/static/dev.html b/web/static/dev.html index 229d0a8..559f587 100644 --- a/web/static/dev.html +++ b/web/static/dev.html @@ -862,6 +862,28 @@ margin-top: 0; border-color: rgba(192,57,43,0.22); background: linear-gradient(180deg, #fff, #fffafa); } + /* 后台进程(bg proc)工具卡活化:summary 里跳秒 + 停止按钮(复用 .tool-call.running spinner) */ + .tool-call.bgproc .bp-state { font-variant-numeric: tabular-nums; } + .tool-call.bgproc .bp-kill { + margin-left: 8px; padding: 0 8px; font-size: 11px; line-height: 1.6; + vertical-align: baseline; cursor: pointer; + } + /* 后台进程完成 toast(右下角,10s 自动消失;非当前任务点击跳转) */ + #proc-toasts { + position: fixed; right: 16px; bottom: 16px; z-index: 120; + display: flex; flex-direction: column; gap: 8px; + } + .proc-toast { + background: #fff; border: 1px solid var(--border); border-radius: 8px; + padding: 10px 14px; box-shadow: 0 4px 16px rgba(0,0,0,.18); + font-size: 13px; max-width: 340px; cursor: pointer; + animation: dock-in .2s ease-out; + } + .proc-toast .pt-cmd { + font-family: var(--mono); font-size: 11px; color: var(--muted); + overflow: hidden; text-overflow: ellipsis; white-space: nowrap; margin-top: 2px; + } + .proc-toast.fail { border-color: rgba(192,57,43,0.5); } /* media tool 摘要 banner(model / size / cost / elapsed,折叠态也可见) */ .tool-banner { display: inline-flex; flex-wrap: wrap; gap: 6px; diff --git a/web/static/js/chat.js b/web/static/js/chat.js index 9292a72..e262c50 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -27,6 +27,7 @@ import { openFilePreview, openPasteFilePreview, closePreviewIfShowing } from "./ import { loadFiles, scheduleFilesRefresh, uploadFiles, formatUploadProgress } from "./files.js"; import { toolActivityLabel, _workingDirName, extractMediaBanner, extractArtifactRels, renderArtifactBarHtml, upgradeMediaArtifacts, ARTIFACT_PRODUCING_TOOLS, _flushMediaArtifactCache } from "./media.js"; import { applyProgressAction, cloneProgressSteps, progressActionsFromToolCalls } from "./progress.js"; +import { refreshProcs, decorateBgprocCard, hasRunningProc, killTaskProcs } from "./procs.js"; export async function loadModels() { try { @@ -403,6 +404,9 @@ export async function selectTask(tid) { if (mqPhone.matches) setMobileView("mv-mid"); // 立即清空 + 显示加载占位:切 task 体感瞬时跟手,不等 meta/messages 两个 await $("chat-stream").innerHTML = `
加载中…
`; + // 先锁后放:按钮先置禁用加载态,meta/proc 状态到齐后由下方流程放到正确状态 + // (idle/streaming/bgproc)—— 杜绝"刷新/切换瞬间按钮短暂可用"的时序窗口 + setActionMode("loading"); renderTaskProgressDock([]); state.outline = []; renderOutlineRail(); // 切 task 先清旧目录,refreshOutline 拉到再渲 try { @@ -413,6 +417,9 @@ export async function selectTask(tid) { api("GET", "/v1/tasks/" + tid), loadMessages(), refreshOutline(), + // proc 状态并入预取:后面 renderLiveRunIfVisible 决定按钮态时数据已就位, + // 消掉"刷新后按钮先短暂可用、锁随轮询迟到"的亚秒窗口 + refreshProcs(), ]); state.taskMeta = meta; renderChatMeta(); @@ -432,6 +439,7 @@ export async function selectTask(tid) { if (e.status === 401) { logout(); return; } renderTaskProgressDock([]); $("chat-stream").innerHTML = `
加载失败:${escapeHtml(e.message)}
`; + setActionMode("idle"); // 加载失败也要把"先锁后放"的锁解开,别把 composer 卡死 } } @@ -1004,7 +1012,17 @@ function createLiveAssistantCard(run) { function renderLiveRunIfVisible() { const run = getLiveRun(state.taskId); if (!run) { - setActionMode("idle"); + // run 收尾/切 task 后:本 task 还有 running 后台进程 → 保持对话锁定不回 idle + if (hasRunningProc(state.taskId)) { + setActionMode("bgproc"); + if (!_bgprocLocked) { + _bgprocLocked = true; + $("chat-hint").textContent = "后台进程运行中,完成后可继续对话…"; + } + } else { + _bgprocLocked = false; + setActionMode("idle"); + } return; } const wrap = $("chat-stream"); @@ -1151,6 +1169,9 @@ function renderMessages(msgs) {
结果(${(txt || "").length} 字符)${banner}
${escapeHtml(txt || "")}
${renderArtifactBarHtml(rels, isProducer)} `; + // bg proc 启动结果 → 卡片活化(spinner/跳秒/停止,与直播态同构;真实状态 + // 由 selectTask 尾部的 refreshProcs 校正,终态卡显示 exit/耗时定格) + decorateBgprocCard(card.querySelector("details.tool-call"), txt || ""); wrap.appendChild(card); continue; } @@ -1253,12 +1274,45 @@ function setActionMode(mode) { btn.textContent = "停止中…"; btn.classList.add("danger"); btn.disabled = true; + } else if (mode === "bgproc") { + // 后台进程运行期:对话锁定,观感与前台执行一致(发送→停止)。后台化的收益 + // 定位是"进程扛超时/服务重启",不改变"一个任务同时只做一件事"的对话心智。 + btn.textContent = "停止"; + btn.classList.add("danger"); + btn.disabled = false; + btn.title = "终止后台进程(已产生的输出会保留)"; + } else if (mode === "loading") { + // 切 task / 刷新的加载窗口:先锁后放(悲观默认),状态到齐再切真实模式 + btn.textContent = "发送"; + btn.classList.add("primary"); + btn.disabled = true; + } +} + +// 无直播 run 时的 composer 状态:有本 task 的 running 后台进程 → 锁定(bgproc), +// 否则 idle。procs.js 每次拉取后回调,进程结束的那次拉取在这里解锁。 +let _bgprocLocked = false; +export function syncBgprocLock() { + if (!state.taskId || getLiveRun(state.taskId)) return; // 直播 run 拥有按钮 + if (hasRunningProc(state.taskId)) { + setActionMode("bgproc"); + // hint 只在进入锁定那一刻写一次 —— 5s 轮询重复写会盖掉"已转写"/"润色中" + // 等临时提示(润色/语音在锁定期与 streaming 期一样可用,只编辑草稿不发送) + if (!_bgprocLocked) { + _bgprocLocked = true; + $("chat-hint").textContent = "后台进程运行中,完成后可继续对话…"; + } + } else if (_bgprocLocked) { + _bgprocLocked = false; + setActionMode("idle"); + $("chat-hint").textContent = ""; } } function chatAction() { - if (isCurrentTaskStreaming()) cancelCurrentTask(); - else sendMessage(); + if (isCurrentTaskStreaming()) { cancelCurrentTask(); return; } + if (hasRunningProc(state.taskId)) { killTaskProcs(state.taskId); return; } + sendMessage(); } $("chat-form").addEventListener("submit", (e) => { e.preventDefault(); chatAction(); }); @@ -1266,7 +1320,9 @@ $("chat-input").addEventListener("keydown", (e) => { // streaming 期间 Enter 不触发停止 —— 用户可能正在编辑下一条草稿,误触发风险高 if (e.key === "Enter" && !e.shiftKey) { e.preventDefault(); - if (!isCurrentTaskStreaming()) sendMessage(); + // streaming / 后台进程锁定 / 加载态期间 Enter 不发送(与按钮语义一致,防误触) + if ($("chat-action").disabled) return; + if (!isCurrentTaskStreaming() && !hasRunningProc(state.taskId)) sendMessage(); } }); $("chat-input").addEventListener("input", syncOptimizeBtn); @@ -2065,6 +2121,7 @@ async function consumeSseStream(url, asstCard, ctx) { handleSseEvent(ev, asstCard, ctx); if (ev.event === "done" || ev.event === "error") { ctx.terminal = true; + refreshProcs(); // run 收尾顺手校正后台进程条(run 中可能起了/杀了 proc) return; } } @@ -2234,6 +2291,12 @@ function handleSseEvent(ev, asstCard, ctx) { if (isProducer) upgradeMediaArtifacts(asstCard); } if (visible) scheduleFilesRefresh(); // 工具调用结果回来,FS 可能被改了,debounce 刷新右侧 + // bg proc 刚启动([Background] 开头的结果)→ 结果卡活化(spinner/跳秒/停止) + // + 立刻拉一次状态,秒数即时起跳 + if (txtStr.startsWith("[Background]")) { + decorateBgprocCard(det, txtStr); + refreshProcs(); + } } else if (t === "context_fold") { // 长会话中段折叠(§8.8 Phase 2):start 阶段在状态行提示(摘要调用要几秒,别静止), // done 阶段插一条与 model_switch 同款的分隔线,让"上下文被整理过"在流里可见。 diff --git a/web/static/js/main.js b/web/static/js/main.js index d9b0305..3c1068d 100644 --- a/web/static/js/main.js +++ b/web/static/js/main.js @@ -16,6 +16,7 @@ import { closeSrcPicker, loadFiles } from "./files.js"; import { loadFolderSuggestions } from "./newtask.js"; import { embedInit } from "./embed.js"; import { loadTaskList, loadModels, loadChannelCards } from "./chat.js"; +import { refreshProcs } from "./procs.js"; // ───── enter app ───── export function enterApp() { @@ -29,6 +30,7 @@ export function enterApp() { loadFolderSuggestions(); // 灌 filter-wd select(modal 打开时会重拉,这里让左 pane 先有选项) loadStorage(); // 顶栏存储用量(后台扫描快照,非实时) loadRole(); // 拉 /v1/me,admin 才显「管理」入口(/static/admin.html) + refreshProcs(); // 后台进程(bg proc):上个会话遗留的 running proc 恢复轮询/展示 } // 顶栏用户名:默认显 name(兜底 user_name → email → uid8),title 悬浮给完整身份。 diff --git a/web/static/js/procs.js b/web/static/js/procs.js new file mode 100644 index 0000000..5146042 --- /dev/null +++ b/web/static/js/procs.js @@ -0,0 +1,180 @@ +// 后台进程(bg proc,DESIGN §8.12)前端:工具卡活化 + 完成 toast + 停止按钮。 +// 数据源 GET /v1/procs(用户级,纯文件系统读取)。轮询式:有 running proc 时每 5s +// 一拉,否则只靠触发点刷新(登录/选 task/run 收尾/[Background] 工具结果)—— +// 不走 SSE:proc 完成时刻往往没有活跃 run,SSE 通道根本不在;轮询成本忽略不计。 +// +// 展示沿用工具卡体验(用户反馈:别另起顶部条):[Background] 的工具结果卡本身 +// 变活卡 —— .running spinner + summary 跳秒 + 停止按钮,终态定格 exit/耗时。 +// 历史重渲(刷新页面)同样活化,updateCards 按最新拉取校正状态。 +// toast 用户级:切到别的任务也能收到完成提示,点击跳转对应任务。 +import { api } from "./api.js"; +import { state } from "./state.js"; +import { $ } from "./dom.js"; +import { escapeHtml } from "./format.js"; +import { selectTask, syncBgprocLock } from "./chat.js"; + +const POLL_MS = 5000; +let _pollTimer = null; +let _tickTimer = null; +let _known = new Map(); // proc_id -> 上次见到的 status(跨拉取比对出 running→终态) +let _byId = new Map(); // proc_id -> 最新 proc 对象(kill 需要 task_id / 卡片更新用) +let _fetchedAt = 0; // 上次拉取时刻(秒数本地推进的基准) + +export async function refreshProcs() { + if (!state.token) return; + let data; + try { data = await api("GET", "/v1/procs"); } catch (e) { return; } // 轮询失败静默 + const procs = data.procs || []; + procs.forEach((p) => { + const prev = _known.get(p.proc_id); + if (prev === "running" && p.status !== "running") notifyDone(p); + _known.set(p.proc_id, p.status); + _byId.set(p.proc_id, p); + }); + _fetchedAt = Date.now(); + updateCards(); + syncBgprocLock(); // 对话锁:本 task 有 running proc → composer 锁定;结束的那次拉取解锁 + if (procs.some((p) => p.status === "running")) startPolling(); else stopPolling(); +} + +// 本 task 是否有 running 后台进程(chat.js 对话锁 / Enter 拦截用) +export function hasRunningProc(taskId) { + if (!taskId) return false; + for (const p of _byId.values()) { + if (p.task_id === taskId && p.status === "running") return true; + } + return false; +} + +// 终止该 task 全部 running 后台进程(composer 主按钮"停止"入口;通常只有 1 个) +export async function killTaskProcs(taskId) { + const running = [..._byId.values()].filter( + (p) => p.task_id === taskId && p.status === "running" + ); + if (!running.length) return; + if (!confirm("确定终止后台进程?已产生的输出会保留,进程本身无法恢复。")) return; + for (const p of running) { + try { + await api("POST", `/v1/tasks/${p.task_id}/procs/${p.proc_id}/kill`); + } catch (e) { /* kill 幂等,失败下轮轮询自会校正 */ } + } + refreshProcs(); +} + +function startPolling() { + if (!_pollTimer) _pollTimer = setInterval(refreshProcs, POLL_MS); + if (!_tickTimer) _tickTimer = setInterval(tickElapsed, 1000); +} + +function stopPolling() { + if (_pollTimer) { clearInterval(_pollTimer); _pollTimer = null; } + if (_tickTimer) { clearInterval(_tickTimer); _tickTimer = null; } +} + +// ───── 工具卡活化(live 与历史重渲两个渲染点都调)───── + +// det = [Background] 工具结果的
。从结果文本抽 proc_id, +// 打上 data-proc-id + 状态 span + 停止按钮;真实状态由 updateCards 按拉取数据校正 +// (历史卡刚渲染时状态未知,先不转 spinner,refreshProcs 回来再定)。 +export function decorateBgprocCard(det, txt) { + if (!det || !txt || !txt.startsWith("[Background]")) return; + const m = txt.match(/proc_id=(\w+)/); + if (!m) return; + det.classList.add("bgproc"); + det.dataset.procId = m[1]; + const sum = det.querySelector("summary"); + if (!sum) return; + const st = document.createElement("span"); + st.className = "tool-elapsed bp-state"; + sum.appendChild(st); + const btn = document.createElement("button"); + btn.type = "button"; + btn.className = "small bp-kill"; + btn.textContent = "停止"; + btn.style.display = "none"; // 状态未知先藏,updateCards 见 running 再显 + btn.onclick = (e) => { + e.preventDefault(); // 别触发 details 开合 + e.stopPropagation(); + killProc(det.dataset.procId, btn); + }; + sum.appendChild(btn); + const p = _byId.get(m[1]); + if (p) updateCards(); // 已有该 proc 的状态(live 场景轮询先到)→ 立即定态 +} + +async function killProc(procId, btn) { + const p = _byId.get(procId); + if (!p) return; + if (!confirm("确定终止这个后台进程?已产生的输出会保留,进程本身无法恢复。")) return; + btn.disabled = true; + try { + await api("POST", `/v1/tasks/${p.task_id}/procs/${procId}/kill`); + } catch (e) { /* kill 幂等,失败下轮轮询自会校正 */ } + refreshProcs(); +} + +// 页面上所有 bgproc 卡按 _byId 最新状态校正(spinner / 跳秒基准 / 终态定格) +function updateCards() { + document.querySelectorAll(".tool-call.bgproc[data-proc-id]").forEach((det) => { + const p = _byId.get(det.dataset.procId); + const st = det.querySelector(".bp-state"); + const btn = det.querySelector(".bp-kill"); + if (!p || !st) return; + if (p.status === "running") { + det.classList.add("running"); + st.dataset.base = p.elapsed_s; + st.textContent = ` · 后台运行中 ${fmtElapsed(p.elapsed_s)}`; + if (btn) btn.style.display = ""; + } else { + det.classList.remove("running"); + delete st.dataset.base; + const label = p.killed ? "已终止" + : p.status === "finished" + ? (p.exit_code === 0 ? "已完成" : `失败(exit ${p.exit_code})`) + : "已中断"; + st.textContent = ` · 后台${label} · 耗时 ${fmtElapsed(p.elapsed_s)}`; + if (btn) btn.remove(); + } + }); +} + +// 秒数本地推进(与工具卡跳秒同体验):elapsed = 拉取时基准 + 本地流逝 +function tickElapsed() { + const dt = Math.floor((Date.now() - _fetchedAt) / 1000); + document.querySelectorAll(".tool-call.bgproc.running .bp-state").forEach((st) => { + const base = parseInt(st.dataset.base, 10); + if (!isNaN(base)) st.textContent = ` · 后台运行中 ${fmtElapsed(base + dt)}`; + }); +} + +function fmtElapsed(secs) { + secs = Math.max(0, secs | 0); + if (secs < 60) return secs + "s"; + if (secs < 3600) return Math.floor(secs / 60) + "m" + String(secs % 60).padStart(2, "0") + "s"; + return Math.floor(secs / 3600) + "h" + String(Math.floor((secs % 3600) / 60)).padStart(2, "0") + "m"; +} + +// ───── 完成 toast(右下角,10s 自动消失;非当前 task 点击可跳转)───── + +function notifyDone(p) { + let wrap = $("proc-toasts"); + if (!wrap) { + wrap = document.createElement("div"); + wrap.id = "proc-toasts"; + document.body.appendChild(wrap); + } + const ok = p.exit_code === 0; + const label = p.killed ? "已终止" : (ok ? "已完成" : `失败(exit ${p.exit_code})`); + const jump = p.task_id && p.task_id !== state.taskId; + const t = document.createElement("div"); + t.className = "proc-toast" + (ok || p.killed ? "" : " fail"); + t.innerHTML = ` +
后台任务${label} · 耗时 ${fmtElapsed(p.elapsed_s || 0)}${jump ? " · 点击查看" : " · 可继续对话"}
+
${escapeHtml(p.command)}
`; + t.onclick = () => { + t.remove(); + if (jump) selectTask(p.task_id); + }; + wrap.appendChild(t); + setTimeout(() => t.remove(), 10000); +}