diff --git a/chv-guest/README.md b/chv-guest/README.md new file mode 100644 index 0000000..94222cb --- /dev/null +++ b/chv-guest/README.md @@ -0,0 +1,71 @@ +# Cloud Hypervisor 客机镜像（chv-guest）——内容清单与版本化构建说明 + +> 本目录只有清单与构建脚本骨架，镜像本身不在本轨内实际构建 +> 发布。原因写在前面，不藏着：当前没有带 /dev/kvm 的宿主 +> 可以构建并实测 VM 镜像（生产主机是 CVM、无嵌套虚拟化； +> 本 VM 容器内同样无 /dev/kvm）。构建与实测见 +> `docs/chv-production-readiness.md` 的“需镜像构建 / 需 KVM +> 宿主实测”项，绝不在本轨声称镜像已建成、已发布、已跑通。 + +## 1. 镜像定位 + +- 基础：Ubuntu 云镜像（与 `cloud/docker/Dockerfile` 的任务镜像 + 对齐能力，不是另起一套环境口径）。 +- 载体：Cloud Hypervisor 直接内核启动（vmlinux）+ raw 磁盘 + 镜像，执行器用 `CHV_KERNEL` / `CHV_ROOTFS` 指向它们。 +- 客内 host.py 位置：`/app/host.py`（与 Docker 的 `/app:ro` + 同款约定，只是载体从容器只读挂载换成 VM 镜像内置代码； + 代码版本 = 构建时仓库 commit，见 §3 版本化）。 +- 绝不进镜像的东西：任何模型 key、系统 key、用户账号数据、 + 生产凭证。模型 key 由客机 cloud-init 预置的运行环境变量名 + 注入（与 Docker 的 `-e <名>` 同款惯例），不写进镜像层。 + +## 2. 内容清单（按现有 Docker 任务镜像能力对齐） + +| 类别 | 内容 | 来源/版本口径 | +| --- | --- | --- | +| 语言与工具链 | Python 3.12、git、Node.js（Playwright 驱动用） | 与任务镜像同款大版本，构建时钉小版本并记入构建日志 | +| 文档/图片工具 | pandoc、poppler-utils（pdftotext）、latexmk、texlive-latex-base、python-docx、openpyxl、python-pptx、PyMuPDF、Pillow | 逐项照抄 `cloud/docker/Dockerfile` 的 apt/pip 清单，不擅自加减 | +| agent 引擎 | codex、codex-code-mode-host、codex-resources/bwrap（成套，约 367M，缺 sibling 直接构建失败） | codex 0.164.0-alpha.9（与生产任务镜像一致）；`CODEX_BIN_DIR` 由构建方提供，bin/ 不进仓库 | +| 浏览器 | Playwright + Chromium（浏览器自动化 MCP 用） | 与生产 playwright 镜像同款：浏览器装在固定路径、供租户用户可读；本骨架只列清单，浏览器版本在构建时钉死 | +| 客机基础 | openssh-server（SSH 通道用）、cloud-init、qemu-guest-agent（可选，宿主信息用） | sshd 默认只允许 key 登录，禁止密码登录 | +| 任务代码 | codex-app-server 仓库代码置于 `/app`（host.py 入口） | 版本见 §3，不挂载宿主代码目录（VM 与宿主无共享路径） | + +中文 LaTeX 说明：与 Docker 镜像同款口径——xelatex + CJK +字体会再增数百 MB，本清单不默认包含；需要时单独一版镜像， +不在本轨悄悄加。 + +## 3. 版本化 + +镜像 tag 格式（构建时三项缺一即失败，不生成无版本镜像）： + +``` +chv-guest---<仓库 commit 短 hash> +``` + +例如（示例格式，非已建成）：`chv-guest-24.04-0.164.0-alpha.9-aebc623`。 + +构建产物清单（构建方必须输出，执行器用它做 preflight 之外的 +版本核对，v1 执行器暂不校验 tag，见就绪清单）： + +- `vmlinux`（与 Ubuntu 云镜像配套的内核） +- `disk.raw`（根文件系统镜像） +- `BUILDINFO.json`：`{ubuntu, codex_version, repo_commit, built_at, + contents: [清单逐项版本]}` + +## 4. 构建脚本骨架 + +见 `build.sh`：只做参数校验、产物路径与版本 tag 计算、以及 +“需要 KVM 宿主 / virt-customize 才能执行的步骤”占位。脚本在 +无 KVM 的机器上运行时应在第一步就明确退出（`CHV_GUEST_BUILD +needs a KVM host`），绝不假装构建成功。 + +## 5. 与执行器的对接 + +- 执行器 preflight 只校验内核/镜像文件存在且可读，不校验镜像 + 内容版本——内容版本核对（BUILDINFO）是镜像发布流程的责任， + 在就绪清单里标为“需镜像构建”项，不在本轨假装已做。 +- 每任务磁盘：执行器 copy 模式下把 `disk.raw` 复制为每任务 + 磁盘，任务之间文件系统独立；reuse 模式仅单租户实验，生产禁用。 +- 快照基准：`CHV_SNAPSHOT_URL` 指向的快照必须由本镜像的干净 + 启动状态制成（pause 后 snapshot），预热池用它做干净 VM 来源。 diff --git a/chv-guest/build.sh b/chv-guest/build.sh new file mode 100755 index 0000000..b1c13fd --- /dev/null +++ b/chv-guest/build.sh @@ -0,0 +1,47 @@ +#!/bin/bash +# chv-guest 镜像构建脚本骨架（未在无 KVM 机器上实际执行过）。 +# +# 用法（需要带 /dev/kvm 的宿主，见 README）： +# CODEX_BIN_DIR=bin UBUNTU_VERSION=24.04 CODEX_VERSION=0.164.0-alpha.9 \ +# REPO_COMMIT=$(git rev-parse --short HEAD) ./build.sh out/ +# +# 本脚本只做三件事，绝不假装已建成镜像： +# 1. 参数校验（缺 CODEX_BIN_DIR 成套二进制、缺版本、缺 /dev/kvm +# 即退出，退出码非 0）； +# 2. 算出版本 tag（chv-guest---）； +# 3. 列出仍需在 KVM 宿主上执行的步骤（下载 Ubuntu 云镜像、 +# virt-customize 装清单软件、拷入 /app 代码、生成 vmlinux / +# disk.raw / BUILDINFO.json），每一步在骨架里以 TODO 标记， +# 不在本轨内静默跳过或谎称完成。 +set -eu + +OUT="${1:-out}" +: "${CODEX_BIN_DIR:?CODEX_BIN_DIR 未设：需 npm 安装 @openai/codex 后的配对二进制目录}" +: "${UBUNTU_VERSION:?UBUNTU_VERSION 未设}" +: "${CODEX_VERSION:?CODEX_VERSION 未设}" +: "${REPO_COMMIT:?REPO_COMMIT 未设：构建时仓库 commit 短 hash}" + +for bin in codex codex-code-mode-host; do + if [ ! -x "$CODEX_BIN_DIR/$bin" ]; then + echo "缺少配对二进制：$CODEX_BIN_DIR/$bin，构建失败" >&2 + exit 1 + fi +done +if [ ! -d "$CODEX_BIN_DIR/codex-resources" ]; then + echo "缺少 codex-resources（bwrap 载体），构建失败" >&2 + exit 1 +fi +if [ ! -e /dev/kvm ]; then + echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2 + exit 1 +fi + +TAG="chv-guest-${UBUNTU_VERSION}-${CODEX_VERSION}-${REPO_COMMIT}" +mkdir -p "$OUT" +echo "tag=$TAG" +echo "out=$OUT" +echo "TODO(KVM host): 下载 Ubuntu ${UBUNTU_VERSION} 云镜像并转 raw" +echo "TODO(KVM host): virt-customize 装 README §2 清单（git/python/pandoc/poppler/latex/codex/playwright/openssh）" +echo "TODO(KVM host): 拷入仓库代码到 /app（版本 $REPO_COMMIT），绝不拷入任何 key" +echo "TODO(KVM host): 生成 vmlinux + disk.raw + BUILDINFO.json，校验 tag 与清单版本" +echo "构建骨架结束：以上 TODO 步骤未执行，镜像未建成（见 README 与生产就绪清单）" diff --git a/cloud/README.md b/cloud/README.md index 820f0e9..180d455 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -1288,6 +1288,57 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 错误文本或返回结果。worker CLI 的 `--executor` 新增 cloud 选项， 构造 CloudExecutor；worker 的 run_one 走 dispatch_task， CloudExecutor 的 dispatch_task/run_task 直接委托给 execute。 +- CloudHypervisorExecutor（生产级 v1，`cloud/sandbox.py`， + 执行器名 `cloud-hypervisor`，绝不用 `cloud` 缩写）： + 单任务单 VM：一个任务一个 VM、独立内核/文件系统/网络， + 任务结束销毁，租户之间无共享路径、无东西向互访；VM 内 + agent 可有 root，但只能影响本 VM。密钥按现有惯例走运行 + 时环境变量名注入，不进镜像、不进代码、不进任务 prompt + 落库（SSH argv 里只有 key 文件路径，绝无 key 值）。 + 唤醒是 API 模式：preflight（见下）→ 容量门 → 磁盘配额门 + → 每任务磁盘供给（copy 模式复制基础镜像）→ 只以 + `--api-socket` 起 VMM → 等 `vmm.ping` 通 → `PUT vm.create` + （VmConfig）→ `PUT vm.boot` → 等 guest SSH 就绪（默认 + 120 秒）才跑 host.py；超时关 VM 并报分类失败。直接启动 + 命令（`build_vm_command`）只作手动等价写法，执行器不走它。 + 暂停/快照/休眠：`pause_vm`/`resume_vm` 走 ch-remote； + `restore_vm` 先 restore 再显式 resume（恢复后默认 paused， + 不 resume 不算醒）；CHV 无原生 hibernate 端点，`hibernate_vm` + 用 pause + snapshot + shutdown 等价实现，恢复走 restore。 + 预热池 `WarmPool`：预热条目是一次性干净快照 VM，acquire + 消耗、release 一律销毁，绝不跨任务复用脏 VM；池空回落 + 冷启动，绝不降级 Docker/Local。关闭与清理：任务结束/ + 失败/取消时 `shutdown_vm`（ch-remote shutdown + 杀进程 + + 删 socket），`cleanup_stale_vms` 处理已死/已关/socket 残留 + 三种情况。审计：VM 的 create/boot/ready/pause/resume/ + snapshot/restore/shutdown/destroy/failed/recycled 均经 + audit_callback 发出，worker 接线后转 AuthStore 审计；失败 + 带 kind 分类（kvm_missing/image_missing/guest_not_ready/ + api_timeout/host_capacity_exceeded/disk_quota_exceeded 等）， + 不笼统 failed。资源与成本：每 VM 的 CPU/内存/磁盘配额、 + 超时、最大轮次（goal 上限 100）复用 worker 现有 quota 语义， + 不另起一套计费；空闲休眠省成本；autoscaler 按 executor 过 + 滤队列深度并透传 `--executor` 给拉起的 worker。网络与治理： + VM 出口默认必须过 CHV_EGRESS_PROXY 白名单代理（未设且未 + 显式允许直连时 preflight 直接失败）；审批门仍走任务的 + approval 与 server 的 merge_gate，VM 内 agent 不能自批。 + 配置（环境变量，见该类 docstring 全表）：CHV_BINARY、 + CHV_KERNEL、CHV_ROOTFS（或 CHV_IMAGE 别名）、CHV_CPU、 + CHV_MEM、CHV_STATE_DIR、CHV_SSH_KEY/USER/PORT、CHV_CMDLINE、 + CHV_NET（直接命令用）/CHV_NET_JSON（API 用）、CHV_REMOTE、 + CHV_MAX_VMS、CHV_DISK_MODE（copy 默认/reuse 仅实验）、 + CHV_DISK_GB、CHV_EGRESS_PROXY、CHV_TASK_TIMEOUT、 + CHV_SNAPSHOT_URL、CHV_WARM_POOL_SIZE、CHV_API_TIMEOUT、 + CHV_READY_TIMEOUT/INTERVAL。默认关闭：server 默认执行器仍 + 是 docker，worker 默认 local，CHV 只在任务创建时显式选 + `executor=cloud-hypervisor` 且有 CHV worker 认领时才跑； + 任务内 LLM 绝不能发现/调用它（调度层选择，只在创建层可见： + POST /tasks 可选值、本文档、控制台选项，不新增任务内 MCP/ + 工具）。真实验证边界：未做真实 VM 启动验证——生产主机 + 本身是 CVM、无 /dev/kvm、无嵌套虚拟化，本 VM 容器内同样 + 无 /dev/kvm；真 VM、快照恢复、压测、安全审计均在 + `docs/chv-production-readiness.md` 里标为待办，绝不声称 + 已生产可用。客机镜像内容清单与构建骨架见 `chv-guest/`。 - 自动 PR（默认关闭，显式触发，`POST /tasks/{id}/pr`，需项目 maintainer 以上权限 approve_merge）：仅当任务状态为 merged 时 可调（其它状态 409；任务不存在 404；权限不足 403）。逻辑按顺序 diff --git a/cloud/autoscaler.py b/cloud/autoscaler.py index de0f2fc..57c7499 100644 --- a/cloud/autoscaler.py +++ b/cloud/autoscaler.py @@ -304,6 +304,14 @@ class Autoscaler: if not any(a == "--host-name" or a.startswith("--host-name=") for a in self.worker_args): cmd.extend(["--host-name", socket.gethostname()]) + # 执行器透传：scaler 的 --executor 只算该类型队列深度， + # 不透传时拉起的 worker 默认 local，永远认领不到该类型 + # 任务（cloud-hypervisor 池就是这样被饿死的）。worker_args + # 里已显式给 --executor 时以显式为准，不覆盖。 + if self.executor and not any( + a == "--executor" or a.startswith("--executor=") + for a in self.worker_args): + cmd.extend(["--executor", self.executor]) return cmd def _start_worker(self, worker_id): @@ -553,6 +561,8 @@ def main(argv=None): help="各 worker 子进程的日志目录") parser.add_argument("--executor", default=None, help="只按该执行器类型的就绪任务算深度；" + "同时透传给拉起的 worker（worker_args 里" + "已显式给 --executor 时以显式为准）；" "不给时按全部任务算（可能轻微多扩，" "受 max-workers 封顶）") parser.add_argument("worker_args", nargs=argparse.REMAINDER, diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 5d26e94..3e1a5c4 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -11,6 +11,11 @@ 时不启动容器）；超时与取消时 docker kill + rm -f，不留僵尸 容器。网络白名单不在本类里假装实现（见 cloud/README.md）。 - CloudExecutor：占位，未接通云端执行。 +- CloudHypervisorExecutor：生产级 v1，Cloud Hypervisor VM 内经 + SSH 调 host.py（API 唤醒流程见该类 docstring：preflight → + vmm.ping → vm.create → vm.boot → SSH 就绪 → 跑任务 → + finally 关 VM；任务内 LLM 绝不能发现/调用它，执行器选择只 + 在任务创建前由 server 白名单/环境默认值/worker 认领决定）。 """ import json @@ -18,8 +23,10 @@ import logging import os import re import shutil +import socket import subprocess import sys +import threading import time from cloud.marker_tools import extract_quota_status as _extract_quota_status @@ -2146,3 +2153,1285 @@ class CloudExecutor(Executor): def run_task(self, task) -> dict: return self.execute(task) + + +_CHV_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + + +class ChvFailure(RuntimeError): + """Cloud Hypervisor 分类失败：RuntimeError 的子类（worker + 现有 except Exception 路径照常捕获），另带 kind 字段供 + 审计与失败分类用，绝不笼统只写 failed。kind 取值见 + CloudHypervisorExecutor.preflight / wake_vm 的注释。""" + + def __init__(self, kind, message): + super().__init__("chv_failure:%s: %s" % (kind, message)) + self.kind = kind + + +def _parse_memory_bytes(value): + """内存字符串转字节：2G/512M/1024K/2048（无单位=字节）。 + 非法值 ValueError，绝不静默按默认算。""" + m = re.match(r"^\s*(\d+)\s*([KMGkmg]?)\s*[Bb]?\s*$", + str(value)) + if not m: + raise ValueError("CHV_MEM 格式非法：%r" % (value,)) + mult = {"": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3} + return int(m.group(1)) * mult[m.group(2).upper()] + + +def _unix_socket_api(socket_path, method, path, payload=None): + """Cloud Hypervisor REST API 的默认传输：标准库 raw HTTP + over Unix socket（无第三方依赖）。返回 (status, body)， + body 能解析 JSON 时为 dict，否则为原文本。socket 不存在 + 或连接失败时 raise RuntimeError（调用方据此判 api 超时 + /失败，不编造 200）。""" + if not os.path.exists(socket_path): + raise RuntimeError("api socket 不存在：%s" % socket_path) + body = b"" + if payload is not None: + body = json.dumps(payload).encode("utf-8") + req_lines = ["%s /api/v1%s HTTP/1.1" % (method, path), + "Host: localhost", + "Accept: application/json", + "Content-Type: application/json", + "Content-Length: %d" % len(body), + "Connection: close", "", ""] + raw = "\r\n".join(req_lines).encode("utf-8") + body + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(5) + try: + sock.connect(socket_path) + sock.sendall(raw) + chunks = [] + while True: + try: + data = sock.recv(65536) + except socket.timeout: + break + if not data: + break + chunks.append(data) + finally: + sock.close() + text = b"".join(chunks).decode("utf-8", errors="replace") + status = 0 + if text.startswith("HTTP/"): + try: + status = int(text.split()[1]) + except (IndexError, ValueError): + status = 0 + body_text = text.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in text \ + else "" + try: + parsed = json.loads(body_text) if body_text.strip() else {} + except ValueError: + parsed = body_text + return status, parsed + + +class CloudHypervisorExecutor(Executor): + """Cloud Hypervisor 执行器 v1（命名：cloud-hypervisor，绝不用 + cloud/ch 缩写——cloud 已是通用 HTTP runner 占位，撞名会让 + 认领过滤与白名单全部错乱）。 + + 调度层选择，不是任务内工具（补充口径，必读）： + - 任务内的 LLM 绝不能发现、调用本执行器：执行器是在任务 + 创建前由 server 白名单（POST /tasks 的 executor 字段）、 + 环境默认值（CLOUD_DEFAULT_EXECUTOR）或 worker CLI + --executor 认领决定的。任务内 agent 若能自选沙箱，等 + 于能从隔离强的执行器降级逃逸到隔离弱的执行器。 + - 本模块不提供任何 MCP/工具/skill 给任务内 agent；需要可 + 发现的是创建任务那一层：POST /tasks 的 executor 可选值、 + cloud/README.md 说明、控制台选项（只改文档与白名单， + 不新增任务内调用入口）。 + + 客内通道：只选 SSH，不做 vsock 第二套。 + - 选 SSH 的理由：客内 host.py 的调用就是一条普通命令， + SSH 直接复用 OpenSSH 的密钥认证与端口转发，云镜像自带 + sshd，无需在客内再写一个 vsock agent、也无需在宿主再写 + 一个 vsock 代理去翻译 host.py 协议。 + - 放弃 vsock 的代价：需要运维提供从宿主到客机的 SSH 通路 + （tap/桥接 + 端口转发，见下文“VM 生命周期”），且 v1 用 + StrictHostKeyChecking=no + BatchMode=yes：客机是临时 + 启动的、host key 每次都变，严格校验会误杀；生产加固时 + 应改为预置 known_hosts 钉死客机 key，本 v1 不假装已做。 + - 密钥红线：SSH 只传 -i 密钥文件路径，绝不把任何 key 值、 + 模型 key 值放进 argv、日志或返回结果；客内 host.py 的 + 模型 key 由客机镜像/cloud-init 预置（运维责任），v1 不 + 在 SSH 命令里注入任务级 key。 + + VM 唤醒流程（写进代码，不是只留文档；API 模式为准）： + - 主唤醒路径是 API 模式，不是“一条直接启动命令就完事”： + wake_vm() 先 preflight，再经 launcher 以 --api-socket + 只启动 VMM 进程（CLI 上不带内核/磁盘配置），等 + vmm.ping 通，再 PUT vm.create（VmConfig）+ PUT vm.boot。 + build_vm_command() 保留的直接启动命令只是文档里的手动 + 等价写法（运维手工起 VM 时用），执行器代码里不走它。 + - 就绪判定只选 SSH，不做 vsock 第二套：boot 后轮询 SSH + “true”命令，直到 CHV_READY_TIMEOUT（默认 120 秒， + 轮询间隔 CHV_READY_INTERVAL，默认 5 秒）内成功才跑 + host.py/任务；超时则关 VM 并报失败，绝不在未就绪时 + 就派任务。vsock 放弃理由同客内通道一节（需客内 agent + + 宿主代理两套新代码，v1 不拼半吊子）。 + - 暂停与快照唤醒是两条独立代码分支： + pause_vm()/resume_vm() 走 ch-remote pause/resume； + restore_vm() 先 ch-remote restore（source_url=...）， + 再显式 resume_vm()——恢复后的 VM 默认是 paused，不 + resume 就不算醒，绝不能只 restore 就返回成功。 + - 关闭与清理：任务结束/失败/取消时走 shutdown_vm()/ + stop_vm()：ch-remote shutdown + 终止登记进程 + 删 + socket。dispatch/run/goal 均走“wake → 跑任务 → + finally shutdown”的每任务 VM 生命周期；execute() 是 + 只读 status，仍走 SSH 直连（要求 VM 已醒，调用方先 + wake_vm），不为一次 status 就起一台 VM。 + - 清理要能处理三种情况：已死进程（poll() 非 None）、 + 已关 VM（无登记进程、无 socket，stop 返回 False）、 + socket 残留（有 socket、无登记进程，照常删 socket 并 + 计入清理）。cleanup_stale_vms() 只清 chv- 前缀的 + socket，绝不按前缀全删其它执行器的东西。 + + 生产级口径（对齐公开沙箱保障口径，不声称复刻某家内部架构）： + - 单任务单 VM：wake → 跑 → finally 关，同一任务不共享 VM； + 每任务磁盘 copy 模式独立，租户无共享路径、无东西向互访； + VM 内 root 只能影响本 VM。 + - 快速启动：WarmPool（本文件末尾）预热干净快照 VM，acquire + 消耗、release 销毁，池空回落冷启动，绝不降级。 + - 资源与成本：CPU/内存/磁盘配额门、宿主容量门（CHV_MAX_VMS）、 + SSH 超时、goal 100 轮上限、执行器内部不重试；成本与配额 + 复用 worker 现有 quota 语义，不另起计费。 + - 网络与治理：出口默认 fail-closed（必须过 CHV_EGRESS_PROXY + 白名单代理）；审批仍走任务 approval 与 server merge_gate， + VM 内 agent 不能自批。 + - 可观测：VM 生命周期事件经 audit_callback 发出（worker 接线 + 后转 AuthStore 审计），失败带 ChvFailure.kind 分类；断点 + 与 task_checkpoints/task_attempts 语义对齐（recycle 保留 + 两者，丢失临时文件）。 + - 默认关闭：server 默认仍 docker、worker 默认仍 local，CHV + 只在显式选择且有 CHV worker 时才跑；无 CHV_* 配置时 + preflight 直接失败。完整清单见 + docs/chv-production-readiness.md。 + - worker 层现状：通用路径已能走 dispatch_task/goal_task； + 断点续跑（goal-resume-breakpoint）在 worker 层对非 + Local 执行器仍明确失败（见 worker.py），本类的 + goal_resume_breakpoint_task 已实现，供直接调用与后续 + worker 接线，不通过 worker 迂回假装已通。 + - 网关现状：worker 通用路径在 sub2api 网关下对非 Local + 执行器直接 loud fail（与 CloudExecutor 同款），v1 不 + 改这条门，所以 CHV + sub2api 组合目前在 worker 层就 + 明确失败，不宣称生产可用。 + + 配置（只走构造参数或环境变量，不写文件；参数优先）： + - CHV_BINARY：cloud-hypervisor 二进制（默认同名 PATH 查找） + - CHV_KERNEL：客机内核路径（必填，文件必须存在） + - CHV_ROOTFS：客机根文件系统镜像路径（必填，文件必须 + 存在；CHV_IMAGE 是同义别名，仅当 CHV_ROOTFS 未设时用） + - CHV_CPU：vCPU 数（默认 2，命令里写 boot=N） + - CHV_MEM：内存（默认 2G，命令里写 size=2G） + - CHV_STATE_DIR：api-socket 目录（默认 /tmp/chv-vms） + - CHV_SSH_KEY：SSH 私钥文件路径（必填，文件必须存在） + - CHV_SSH_USER：SSH 用户（默认 root） + - CHV_SSH_PORT：宿主侧转发到客机的 SSH 端口（默认 2222） + - CHV_CMDLINE：客机内核命令行（默认 + "console=hvc0 root=/dev/vda1 rw"） + - CHV_NET：直接启动命令里的 --net 原样透传值（例如 + tap=,mac=...，运维按实际 tap 配置给；不给则命令里不带 + --net）。API 模式不用它，API 的网络配置走 CHV_NET_JSON。 + - CHV_NET_JSON：API 模式 VmConfig 的 net 列表 JSON（例如 + '[{"tap":"","mac":"12:34:56:78:90:01"}]'，运维按实际 + tap 给；不给则 VmConfig 不带 net，客机无网卡、SSH 就 + 绪必然超时失败——这是 fail-fast 前置条件，不是静默成功） + - CHV_REMOTE：ch-remote 二进制（默认同名 PATH 查找； + preflight 必查，缺失即 fail-fast；stop/cleanup 里调用 + 失败时尽力而为不炸） + - CHV_API_TIMEOUT：等 vmm.ping 通的上限秒数（默认 30） + - CHV_READY_TIMEOUT：等 guest SSH 就绪的上限秒数（默认 120） + - CHV_READY_INTERVAL：SSH 就绪轮询间隔秒数（默认 5） + + 前置校验（preflight，执行入口必调，缺一即 fail-fast）： + 收集全部缺失项一次报出具体缺哪一项，绝不笼统报错： + cloud-hypervisor 与 ch-remote 二进制存在、/dev/kvm 存在 + 且可读写、CPU 有 vmx/svm 标志、内核与磁盘镜像路径存在 + 且可读、SSH key 存在且可读。任一项缺失都 raise + RuntimeError，绝不静默回落 Docker/Local。_validate() + 即 preflight（旧名保留，供旧调用点用）。 + 命令构造（build_vm_command/build_ssh_command/ + build_vm_config）是纯函数，不做文件存在校验，单测可 + 直接断言命令与 VmConfig 形状。 + + 真实验证边界（写进 PR 描述，不可删）： + 未做真实 Cloud Hypervisor 启动验证。当前环境没有可跑真 + VM 的宿主：生产主机 124.222.165.170 本身就是 CVM + （systemd-detect-virt=kvm，DMI product=CVM），/dev/kvm + 不存在、无 vmx/svm 标志、无 kvm 模块、未装 + cloud-hypervisor，无嵌套虚拟化透传；本 VM 容器内同样 + 无 /dev/kvm。v1 验证只能是 mock runner + 干跑命令构造 + + 单元测试。真 VM 验证需另备带 /dev/kvm 的裸金属或 + 开启嵌套虚拟化的实例，那是用户侧开通决策，不在本轨内。 + """ + + def __init__(self, binary=None, kernel=None, rootfs=None, + image=None, cpus=None, memory=None, state_dir=None, + runner=None, launcher=None, ssh_key=None, + ssh_user=None, ssh_port=None, + guest_host_py="/app/host.py", cmdline=None, + net=None, net_json=None, kvm_path="/dev/kvm", + cpuinfo_path="/proc/cpuinfo", env=None, + model=None, ch_remote=None, api_client=None, + sleeper=None, api_timeout=None, ready_timeout=None, + ready_interval=None): + self._env = env if env is not None else os.environ + self.binary = binary or self._env.get("CHV_BINARY") \ + or "cloud-hypervisor" + self.kernel = kernel or self._env.get("CHV_KERNEL") + self.rootfs = rootfs or image \ + or self._env.get("CHV_ROOTFS") \ + or self._env.get("CHV_IMAGE") + self.cpus = str(cpus or self._env.get("CHV_CPU") or "2") + self.memory = memory or self._env.get("CHV_MEM") or "2G" + self.state_dir = state_dir or self._env.get("CHV_STATE_DIR") \ + or "/tmp/chv-vms" + self.ssh_key = ssh_key or self._env.get("CHV_SSH_KEY") + self.ssh_user = ssh_user or self._env.get("CHV_SSH_USER") \ + or "root" + try: + self.ssh_port = int(ssh_port + or self._env.get("CHV_SSH_PORT") + or 2222) + except (TypeError, ValueError): + raise ValueError("CHV_SSH_PORT 必须是整数") + self.guest_host_py = guest_host_py + self.cmdline = cmdline or self._env.get("CHV_CMDLINE") \ + or "console=hvc0 root=/dev/vda1 rw" + self.net = net or self._env.get("CHV_NET") + self.net_json = net_json or self._env.get("CHV_NET_JSON") + self.kvm_path = kvm_path + self.cpuinfo_path = cpuinfo_path + self.ch_remote = ch_remote or self._env.get("CHV_REMOTE") \ + or "ch-remote" + self.model = model + # 生产级配置（参数优先于环境变量；见类 docstring 与 + # cloud/README.md 的 CHV 一节，默认值全部显式写出）： + # - max_vms：本执行器实例同时最多几个 VM（宿主容量门， + # 超限明确拒绝，绝不降级 Docker/Local） + # - disk_mode：copy=每任务复制一份磁盘镜像（默认，慢但 + # 文件系统真正独立）；reuse=复用基础镜像路径（仅单租户 + # 实验用，多租户下磁盘不独立，绝不能用于生产） + # - disk_gb：基础镜像超过该 GB 数即拒绝（磁盘配额门） + # - egress_proxy：客机出口白名单代理地址（必填，除非 + # allow_direct_egress 显式开；直连外网默认不给） + # - task_timeout：SSH 单次任务命令超时秒数（None=不限， + # 但 worker 层 task_timeout 仍是外层硬护栏） + # - snapshot_url：快照池的基准快照（无则冷启动） + # - warm_pool_size：预热池目标数（0=关闭预热池） + try: + self.max_vms = int( + self._env.get("CHV_MAX_VMS") or 4) + except ValueError: + raise ValueError("CHV_MAX_VMS 必须是整数") + self.disk_mode = (self._env.get("CHV_DISK_MODE") or "copy") + if self.disk_mode not in ("copy", "reuse"): + raise ValueError( + "CHV_DISK_MODE 只能是 copy 或 reuse：%s" + % self.disk_mode) + disk_gb_raw = self._env.get("CHV_DISK_GB") + self.disk_gb = float(disk_gb_raw) if disk_gb_raw else None + self.egress_proxy = self._env.get("CHV_EGRESS_PROXY") + self.allow_direct_egress = ( + self._env.get("CHV_ALLOW_DIRECT_EGRESS") == "1") + task_timeout_raw = self._env.get("CHV_TASK_TIMEOUT") + self.task_timeout = float(task_timeout_raw) \ + if task_timeout_raw else None + self.snapshot_url = self._env.get("CHV_SNAPSHOT_URL") + try: + self.warm_pool_size = int( + self._env.get("CHV_WARM_POOL_SIZE") or 0) + except ValueError: + raise ValueError("CHV_WARM_POOL_SIZE 必须是整数") + self.api_client = api_client if api_client is not None \ + else _unix_socket_api + self.sleeper = sleeper if sleeper is not None else time.sleep + try: + self.api_timeout = float( + api_timeout if api_timeout is not None + else self._env.get("CHV_API_TIMEOUT") or 30) + self.ready_timeout = float( + ready_timeout if ready_timeout is not None + else self._env.get("CHV_READY_TIMEOUT") or 120) + self.ready_interval = float( + ready_interval if ready_interval is not None + else self._env.get("CHV_READY_INTERVAL") or 5) + except ValueError: + raise ValueError("CHV_*_TIMEOUT/INTERVAL 必须是数字") + if self.ready_interval <= 0: + self.ready_interval = 5 + self.audit_callback = None + self._vms_lock = threading.Lock() + self.runner = runner if runner is not None else _default_runner + if launcher is not None: + self.launcher = launcher + else: + def _popen(cmd): + return subprocess.Popen( + cmd, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + self.launcher = _popen + self._vms = {} + + # -- 名字 / 路径 --------------------------------------------------- + def _host_name(self, task): + return re.sub(r"[^A-Za-z0-9_.-]", "-", + str(task.get("id"))) + + def vm_name(self, task_or_name): + """VM 名：chv-<安全字符任务名>（与 Docker 的 codex- 同款 + 派生，同一任务每次算出同一名字，清理才能按名找）。 + 幂等：传入的已是 chv- 前缀名时不再二次加前缀（清理 + 扫描 socket 文件名时传的就是 chv- 名，二次加前缀会 + 把 chv-dead 算成 chv-chv-dead、永远清不掉）。""" + raw = task_or_name.get("id") if isinstance(task_or_name, dict) \ + else str(task_or_name) + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", raw) + if safe.startswith("chv-"): + return safe + return "chv-" + safe + + def api_socket(self, task_or_name): + """该 VM 的 api-socket 路径：state_dir/.sock。""" + return os.path.join(self.state_dir, + self.vm_name(task_or_name) + ".sock") + + # -- 前置校验 ------------------------------------------------------ + @staticmethod + def _binary_ok(path): + if "/" in path: + return os.path.isfile(path) and os.access(path, os.X_OK) + return shutil.which(path) is not None + + def preflight(self): + """预检（缺一不可则 fail-fast）：收集全部缺失项一次报出 + 具体缺哪一项，绝不笼统报错；失败为 ChvFailure（带 kind， + 供审计与失败分类用），绝不回落 Docker/Local。 + 检查项与 kind： + - binary_missing：cloud-hypervisor 二进制不存在/不可执行 + - binary_missing：ch-remote 二进制不存在/不可执行 + - kvm_missing：/dev/kvm 不存在 + - kvm_missing：/dev/kvm 不可读/不可写（分开报） + - cpu_flag_missing：/proc/cpuinfo 无 vmx/svm 标志 + - kernel_missing：CHV_KERNEL 未设/文件不存在/不可读 + - image_missing：CHV_ROOTFS 未设/文件不存在/不可读 + - ssh_key_missing：CHV_SSH_KEY 未设/文件不存在/不可读 + - egress_not_configured：CHV_EGRESS_PROXY 未设且未显式 + 允许直连（默认 fail-closed，VM 出口必须过白名单代理） + """ + missing = [] # (kind, 文本) 列表，全部收集完再一次抛 + if not self._binary_ok(self.binary): + missing.append(("binary_missing", + "cloud-hypervisor 二进制缺失或不可执行：%s" + % self.binary)) + if not self._binary_ok(self.ch_remote): + missing.append(("binary_missing", + "ch-remote 二进制缺失或不可执行：%s" + % self.ch_remote)) + if not os.path.exists(self.kvm_path): + missing.append(("kvm_missing", + "/dev/kvm 不存在：%s，宿主无 KVM" + % self.kvm_path)) + else: + if not os.access(self.kvm_path, os.R_OK): + missing.append(("kvm_missing", + "/dev/kvm 不可读：%s" % self.kvm_path)) + if not os.access(self.kvm_path, os.W_OK): + missing.append(("kvm_missing", + "/dev/kvm 不可写：%s" % self.kvm_path)) + cpuinfo_ok = False + try: + with open(self.cpuinfo_path, "r") as fh: + for line in fh: + if line.startswith("flags"): + flags = line.split(":", 1)[1].split() + if "vmx" in flags or "svm" in flags: + cpuinfo_ok = True + break + except OSError: + missing.append(("cpu_flag_missing", + "CPU 信息文件缺失：%s" % self.cpuinfo_path)) + else: + if not cpuinfo_ok: + missing.append(("cpu_flag_missing", + "CPU 无虚拟化标志（无 vmx/svm）：%s" + % self.cpuinfo_path)) + if not self.kernel: + missing.append(("kernel_missing", "CHV_KERNEL 未设")) + elif not os.path.isfile(self.kernel): + missing.append(("kernel_missing", + "内核文件不存在：%s" % self.kernel)) + elif not os.access(self.kernel, os.R_OK): + missing.append(("kernel_missing", + "内核文件不可读：%s" % self.kernel)) + if not self.rootfs: + missing.append(("image_missing", "CHV_ROOTFS 未设")) + elif not os.path.isfile(self.rootfs): + missing.append(("image_missing", + "磁盘镜像文件不存在：%s" % self.rootfs)) + elif not os.access(self.rootfs, os.R_OK): + missing.append(("image_missing", + "磁盘镜像文件不可读：%s" % self.rootfs)) + if not self.ssh_key: + missing.append(("ssh_key_missing", "CHV_SSH_KEY 未设")) + elif not os.path.isfile(self.ssh_key): + missing.append(("ssh_key_missing", + "SSH key 文件不存在：%s" % self.ssh_key)) + elif not os.access(self.ssh_key, os.R_OK): + missing.append(("ssh_key_missing", + "SSH key 文件不可读：%s" % self.ssh_key)) + if not self.egress_proxy and not self.allow_direct_egress: + missing.append(("egress_not_configured", + "CHV_EGRESS_PROXY 未设且未显式允许直连：" + "VM 出口必须过白名单代理，默认不给直连外网")) + if missing: + kind = missing[0][0] + detail = "; ".join(text for _, text in missing) + raise ChvFailure( + kind, "cloud-hypervisor preflight 失败：%s" % detail) + return True + + def _validate(self): + """旧名保留：即 preflight。""" + return self.preflight() + + # -- 审计 ---------------------------------------------------------- + def _emit(self, event, task, detail="", kind=None): + """VM 生命周期审计事件：有 audit_callback 时调它 + （worker 接线后转 AuthStore 审计，即任务 timeline 的 + 审计来源），无 callback 时只记日志，绝不因审计失败炸 + 任务。event 取值：vm_create/vm_boot/vm_ready/vm_pause/ + vm_resume/vm_snapshot/vm_restore/vm_shutdown/vm_destroy/ + vm_failed/vm_recycled/warm_pool_acquire/warm_pool_release。""" + try: + if self.audit_callback is not None: + self.audit_callback(event, task, detail, kind) + else: + _CHV_LOGGER.info("chv_audit event=%s kind=%s %s", + event, kind, detail) + except Exception: + pass + + # -- 命令构造（纯函数，不做文件校验） ------------------------------- + def build_vm_command(self, task): + """构造 cloud-hypervisor 启动命令（干跑可测）。 + + 顺序固定：<二进制> --api-socket --cpus boot=N + --memory size=M --kernel <内核> --cmdline <命令行> + --disk path=<镜像> [--net <运维透传值>]。 + --net 不给时不带：tap/桥接是运维侧配置，本类不编造 + 一个不存在的 tap 名。 + """ + cmd = [self.binary, + "--api-socket", self.api_socket(task), + "--cpus", "boot=%s" % self.cpus, + "--memory", "size=%s" % self.memory, + "--kernel", self.kernel, + "--cmdline", self.cmdline, + "--disk", "path=%s" % self.rootfs] + if self.net: + cmd += ["--net", self.net] + return cmd + + def build_ssh_command(self, task, inner_cmd): + """构造 SSH 命令：ssh -p <端口> -i + 安全选项 + + user@127.0.0.1 + 客内命令。argv 里只有 key 文件路径， + 绝无任何 key 值。egress_proxy 已设时客内命令前加 env + 前缀（HTTPS_PROXY/HTTP_PROXY/NO_PROXY，代理地址本身 + 不是密钥；客机出口只走该白名单代理，见类 docstring）。""" + inner = list(inner_cmd) + if self.egress_proxy: + inner = ["env", + "HTTPS_PROXY=%s" % self.egress_proxy, + "HTTP_PROXY=%s" % self.egress_proxy, + "NO_PROXY=localhost,127.0.0.1"] + inner + return ["ssh", "-p", str(self.ssh_port), + "-i", self.ssh_key, + "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null", + "-o", "BatchMode=yes", + "%s@127.0.0.1" % self.ssh_user] + inner + + # -- API 模式：VmConfig / API 调用 / 就绪轮询 ----------------------- + def build_vm_config(self, task_disk=None): + """API 模式的 VmConfig（vm.create 请求体，纯函数）： + cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、 + kernel{path}、cmdline{args}、disks[{path: 每任务磁盘}]、 + rng{/dev/urandom}，以及 CHV_NET_JSON 给出的 net 列表 + （不给则不带 net——客机无网卡时 SSH 就绪会超时失败， + 这是 fail-fast 前置条件，见类 docstring）。 + task_disk 不给时用基础镜像路径（仅 lab/reuse 模式）。""" + try: + cpus = int(self.cpus) + except ValueError: + raise ValueError("CHV_CPU 必须是整数：%r" % (self.cpus,)) + config = { + "cpus": {"boot_vcpus": cpus, "max_vcpus": cpus}, + "memory": {"size": _parse_memory_bytes(self.memory)}, + "kernel": {"path": self.kernel}, + "cmdline": {"args": self.cmdline}, + "disks": [{"path": task_disk or self.rootfs}], + "rng": {"src": "/dev/urandom"}, + } + if self.net_json: + try: + net = json.loads(self.net_json) + except ValueError: + raise ValueError("CHV_NET_JSON 不是合法 JSON") + config["net"] = net if isinstance(net, list) else [net] + return config + + def _api_call(self, socket_path, method, path, payload=None): + """API 调用归一化：fake api_client 可返回 (status, body) + 或 {"status":..., "body":...} 或带 .status 的对象，本 + 方法统一成 (status:int, body)。""" + result = self.api_client(socket_path, method, path, payload) + if isinstance(result, tuple) and len(result) == 2: + return int(result[0]), result[1] + if isinstance(result, dict): + status = result.get("status", result.get("status_code")) + body = result.get("body", result) + return (int(status) if status is not None else 0), body + status = getattr(result, "status", None) \ + or getattr(result, "status_code", None) + return (int(status) if status is not None else 0), result + + def _call_runner(self, cmd, timeout=None): + """runner 调用：timeout 给定时优先传 timeout，假 runner + 只接受 cmd 时 TypeError 回退到只传 cmd（超时护栏丢了， + 但绝不因此静默换执行器；worker 层 task_timeout 仍是 + 外层硬护栏）。""" + if timeout is None: + return self.runner(cmd) + try: + return self.runner(cmd, timeout=timeout) + except TypeError: + return self.runner(cmd) + + def _kill_proc(self, proc): + """终止登记进程：terminate → 等 5 秒 → kill，全部 + best-effort，绝不抛异常。""" + if proc is None: + return + try: + proc.terminate() + except Exception: + pass + try: + proc.wait(timeout=5) + except Exception: + try: + proc.kill() + except Exception: + pass + + def _active_count(self): + with self._vms_lock: + return len(self._vms) + + def _check_capacity(self): + """宿主容量门：已登记 VM 数达到 max_vms 即明确拒绝 + （kind=host_capacity_exceeded，调用方应排队/拒绝， + 绝不降级 Docker/Local）。""" + if self._active_count() >= self.max_vms: + raise ChvFailure( + "host_capacity_exceeded", + "宿主 CHV 容量已满：已登记 %d 台，上限 %d，" + "请排队或拒绝本任务" % (self._active_count(), + self.max_vms)) + + def _check_disk_quota(self): + """磁盘配额门：CHV_DISK_GB 已设且基础镜像超过该 GB 数 + 即拒绝（kind=disk_quota_exceeded）。未设则不检查。""" + if self.disk_gb is None or not self.rootfs: + return + try: + size = os.path.getsize(self.rootfs) + except OSError: + return + if size > self.disk_gb * 1024 ** 3: + raise ChvFailure( + "disk_quota_exceeded", + "磁盘镜像 %d 字节超过配额 %.1f GB" + % (size, self.disk_gb)) + + def _provision_task_disk(self, task): + """每任务磁盘供给（单任务单 VM 的文件系统独立性）： + - copy（默认）：把基础镜像复制到 + state_dir//disk.raw，VmConfig 用该路径；复制 + 慢但两个任务绝不写同一文件，测试用小文件可直接验证。 + - reuse（仅单租户实验）：直接返回基础镜像路径，磁盘 + 不独立，生产禁用（preflight 不拦它，但类 docstring + 与就绪清单里明写 reuse 不可用于生产）。 + 返回每任务磁盘路径。""" + if self.disk_mode == "reuse": + return self.rootfs + task_dir = os.path.join(self.state_dir, self.vm_name(task)) + os.makedirs(task_dir, exist_ok=True) + task_disk = os.path.join(task_dir, "disk.raw") + if not os.path.exists(task_disk): + shutil.copyfile(self.rootfs, task_disk) + return task_disk + + def wait_api_ready(self, task): + """等 vmm.ping 通：按次数轮询（api_timeout/1 秒取整， + 至少 1 次），不用墙钟死循环——fake sleeper 不推进墙钟 + 时墙钟循环会空转，次数循环在单测里是确定性的。 + 超时 raise ChvFailure(kind=api_timeout)。""" + socket_path = self.api_socket(task) + attempts = max(1, int(self.api_timeout)) + for attempt in range(attempts): + try: + status, _ = self._api_call( + socket_path, "GET", "/vmm.ping") + if status == 200: + return True + except Exception: + pass + if attempt + 1 < attempts: + self.sleeper(1) + raise ChvFailure( + "api_timeout", + "vmm.ping 超时：%d 次内 API 未通，socket=%s" + % (attempts, socket_path)) + + def wait_guest_ready(self, task): + """等 guest SSH 就绪（只选 SSH，见类 docstring）： + 轮询 ssh … "true"，returncode 0 即就绪；次数 = + ready_timeout/ready_interval 取整，至少 1 次。 + 超时时先关 VM 再 raise ChvFailure(kind=guest_not_ready)， + 绝不在未就绪时派任务。""" + attempts = max(1, int(self.ready_timeout / self.ready_interval)) + for attempt in range(attempts): + try: + result = self._call_runner( + self.build_ssh_command(task, ["true"])) + if getattr(result, "returncode", 1) == 0: + return True + except Exception: + pass + if attempt + 1 < attempts: + self.sleeper(self.ready_interval) + try: + self.stop_vm(task) + except Exception: + pass + raise ChvFailure( + "guest_not_ready", + "guest SSH 就绪超时：%d 次轮询内 sshd 未通，已关 VM" + % attempts) + + def _guest_host_command(self, *args, model=None, effort=None, + approval=None, sandbox=None, + model_provider=None): + """客内 host.py 命令：python3 /app/host.py [顶层全局参数] + [子命令...]，顶层参数顺序与 Local/Docker 同款（必须在 + 子命令之前）。guest_host_py 默认 /app/host.py：客机镜像 + 须把 host.py 放在该路径（与 Docker 的 /app:ro 同款约定， + 只是载体从容器换成 VM 镜像）。""" + cmd = ["python3", self.guest_host_py] + effective = model or self.model + if effective: + cmd += ["--model", effective] + if model_provider: + cmd += ["--model-provider", model_provider] + if effort: + cmd += ["--effort", effort] + cmd += _approval_cli_args(approval) + if sandbox: + cmd += ["--sandbox", sandbox] + cmd += list(args) + return cmd + + # -- 生命周期：API 唤醒 / 暂停恢复 / 快照 / 关闭 -------------------- + def _launch_api_only(self, task): + """只启动 VMM 进程（API 模式第一步）：CLI 上只带 + --api-socket，不带内核/磁盘配置；VM 配置随后走 + vm.create 请求体。返回登记的进程。""" + socket_path = self.api_socket(task) + cmd = [self.binary, "--api-socket", socket_path] + proc = self.launcher(cmd) + with self._vms_lock: + self._vms[self.vm_name(task)] = proc + return proc + + def _ch_remote(self, task_or_name, *args): + """ch-remote 调用：[ch-remote --api-socket= args...]。 + returncode 非 0 即 raise RuntimeError（fail-fast）， + runner 本身抛异常时原样传播。返回 runner 结果。""" + name = self.vm_name(task_or_name) + socket_path = os.path.join(self.state_dir, name + ".sock") + result = self.runner([self.ch_remote, + "--api-socket=%s" % socket_path] + + list(args)) + if getattr(result, "returncode", 0) not in (0, None): + raise RuntimeError( + "ch-remote %s 失败：returncode=%s" + % (" ".join(args), getattr(result, "returncode"))) + return result + + def wake_vm(self, task): + """主唤醒路径（API 模式，dispatch/run/goal 的前置步骤）： + preflight → 容量门 → 磁盘配额门 → 每任务磁盘供给 → + launcher 只起 VMM（--api-socket）→ 等 vmm.ping 通 → + PUT vm.create（VmConfig）→ PUT vm.boot → 等 guest + SSH 就绪。任一步失败都关 VM（stop_vm）再抛 + ChvFailure，绝不带着半醒的 VM 去跑任务。 + 成功返回 {vm, api_socket, state: "running", ready: True, + mode: "cold"}，并审计 vm_create/vm_boot/vm_ready。 + 单任务单 VM：同一任务重复 wake 时先关旧 VM 再起新 + 的，绝不两个 VM 共用一个任务名。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + self.preflight() + self._check_capacity() + self._check_disk_quota() + name = self.vm_name(task) + old = None + with self._vms_lock: + if name in self._vms: + old = self._vms.pop(name, None) + if old is not None: + self._kill_proc(old) + task_disk = self._provision_task_disk(task) + socket_path = self.api_socket(task) + try: + self._launch_api_only(task) + self.wait_api_ready(task) + status, _ = self._api_call( + socket_path, "PUT", "/vm.create", + self.build_vm_config(task_disk)) + if status not in (200, 204): + raise ChvFailure( + "vm_create_failed", + "vm.create 失败：status=%s" % status) + self._emit("vm_create", task, + "vm=%s disk=%s" % (name, task_disk)) + status, _ = self._api_call(socket_path, "PUT", "/vm.boot") + if status not in (200, 204): + raise ChvFailure( + "vm_boot_failed", + "vm.boot 失败：status=%s" % status) + self._emit("vm_boot", task, "vm=%s" % name) + self.wait_guest_ready(task) + except Exception: + try: + self.stop_vm(task) + except Exception: + pass + if isinstance(task, dict): + self._emit("vm_failed", task, + "vm=%s wake 失败" % name) + raise + self._emit("vm_ready", task, "vm=%s" % name) + return {"vm": name, "api_socket": socket_path, + "state": "running", "ready": True, "mode": "cold"} + + def start_vm(self, task): + """旧名保留：即 wake_vm（API 模式）。直接启动命令 + build_vm_command() 只作手动等价写法，不再是执行器 + 的启动路径——两条启动路径并存是上一版的半吊子， + 本版只认 API 模式。""" + return self.wake_vm(task) + + def pause_vm(self, task_or_name): + """暂停 VM：ch-remote pause。成功审计 vm_pause。""" + self._ch_remote(task_or_name, "pause") + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + self._emit("vm_pause", task, + "vm=%s" % self.vm_name(task_or_name)) + return True + + def resume_vm(self, task_or_name): + """唤醒已暂停的 VM：ch-remote resume。成功审计 vm_resume。 + 注意：这是暂停唤醒的唯一路径，恢复后的 VM 也走它 + （见 restore_vm），绝不能只 restore 就算醒。""" + self._ch_remote(task_or_name, "resume") + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + self._emit("vm_resume", task, + "vm=%s" % self.vm_name(task_or_name)) + return True + + def snapshot_vm(self, task_or_name, snapshot_url): + """给（已暂停的）VM 做快照：ch-remote snapshot + 。snapshot_url 为空即 ValueError；成功 + 审计 vm_snapshot。快照本身不改变 VM 状态（仍 paused， + 需 resume 才醒）。""" + if not snapshot_url: + raise ValueError("snapshot_url 不能为空") + self._ch_remote(task_or_name, "snapshot", + "source_url=%s" % snapshot_url) + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + self._emit("vm_snapshot", task, + "vm=%s snapshot=%s" + % (self.vm_name(task_or_name), snapshot_url)) + return snapshot_url + + def restore_vm(self, task, snapshot_url): + """快照恢复唤醒：preflight → 只起 VMM（--api-socket） + → 等 vmm.ping → ch-remote restore → 显式 resume → + 等 guest SSH 就绪。恢复后的 VM 默认是 paused，不 + resume 就不算醒——本方法绝不省略 resume 那一步。 + 任一步失败都关 VM 再抛。成功返回 {vm, api_socket, + state: "running", ready: True, mode: "restored"}。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not snapshot_url: + raise ValueError("snapshot_url 不能为空") + if "://" not in snapshot_url: + snapshot_url = "file://%s" % snapshot_url + self.preflight() + self._check_capacity() + name = self.vm_name(task) + try: + self._launch_api_only(task) + self.wait_api_ready(task) + self._ch_remote(task, "restore", + "source_url=%s" % snapshot_url) + self._emit("vm_restore", task, + "vm=%s snapshot=%s" % (name, snapshot_url)) + self.resume_vm(task) + self.wait_guest_ready(task) + except Exception: + try: + self.stop_vm(task) + except Exception: + pass + raise + return {"vm": name, "api_socket": self.api_socket(task), + "state": "running", "ready": True, "mode": "restored"} + + def hibernate_vm(self, task_or_name, snapshot_url): + """休眠（CHV 无原生 hibernate 端点，等价实现）： + pause + snapshot + shutdown。休眠后文件与 worktree + 保留在快照里（快照是暂停时刻的完整状态），临时文件 + 不在恢复保证内（见 recycle_vm 的说明）。 + 成功返回 {vm, state: "hibernated", snapshot: ...}。 + 本方法明写“等价实现”，绝不声称接了原生 hibernate。""" + self.pause_vm(task_or_name) + self.snapshot_vm(task_or_name, snapshot_url) + self.stop_vm(task_or_name) + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + return {"vm": self.vm_name(task_or_name), + "state": "hibernated", "snapshot": snapshot_url} + + def wake_from_hibernate(self, task, snapshot_url): + """从休眠唤醒：即 restore_vm（restore + 显式 resume + + SSH 就绪），语义与 restore_vm 完全一致，只是名字 + 说明来源是休眠快照。""" + return self.restore_vm(task, snapshot_url) + + def stop_vm(self, task_or_name): + """停止 VM（尽力而为，不抛异常；三种情况）： + - 已死进程：登记进程 poll() 非 None，跳过 terminate + 等待，直接删 socket、清登记，返回 True。 + - 已关 VM：无登记进程、无 socket，ch-remote 仍尽力 + 调过一次，返回 False（本来就没有什么可清）。 + - socket 残留：有 socket、无登记进程，删 socket 并 + 返回 True。 + ch-remote shutdown 失败不炸（VM 可能已死），但 + ch-remote 命令本身仍会发出——测试据此断言关闭路径 + 确实用了 ch-remote shutdown，而不是只杀进程。""" + name = self.vm_name(task_or_name) + socket_path = os.path.join(self.state_dir, name + ".sock") + with self._vms_lock: + had_proc = name in self._vms + proc = self._vms.get(name) + had = had_proc or os.path.exists(socket_path) + try: + self.runner([self.ch_remote, + "--api-socket=%s" % socket_path, "shutdown"]) + except Exception: + pass + if proc is not None: + with self._vms_lock: + self._vms.pop(name, None) + try: + alive = proc.poll() is None + except Exception: + alive = False + if alive: + self._kill_proc(proc) + try: + if os.path.exists(socket_path): + os.remove(socket_path) + except Exception: + pass + task = task_or_name if isinstance(task_or_name, dict) \ + else {"id": task_or_name} + if had: + self._emit("vm_shutdown", task, "vm=%s" % name) + self._emit("vm_destroy", task, "vm=%s" % name) + _CHV_LOGGER.info("chv_vm_stop vm=%s had=%s", name, had) + return had + + def shutdown_vm(self, task_or_name): + """关闭入口的正式名：即 stop_vm（ch-remote shutdown）。 + 任务结束/失败/取消时调它。""" + return self.stop_vm(task_or_name) + + def cancel_vm(self, task_or_name): + """取消入口：即 shutdown_vm。worker 的取消路径对 + Docker 调 kill_container，对 CHV 调本方法——取消绝不 + 只改数据库状态，必须真关 VM。""" + return self.shutdown_vm(task_or_name) + + def recycle_vm(self, task): + """回收 VM（任务结束后的销毁语义）： + 先 shutdown，再返回保留/丢失清单。保留的只有宿主侧 + 的 task_checkpoints / task_attempts（数据库里的断点与 + 尝试账本，VM 销毁不影响它们，goal 断点续跑据此接续）； + 丢失的是客机内的临时文件与未拷出的 worktree 临时物 + ——recycle 绝不声称“恢复临时文件”。需要拷出的工件 + 必须在 shutdown 之前由调用方经 SSH 拷出，本方法不 + 假装替调用方拷了。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + self.shutdown_vm(task) + self._emit("vm_recycled", task, + "vm=%s" % self.vm_name(task)) + return { + "vm": self.vm_name(task), + "state": "recycled", + "preserved": ["task_checkpoints", "task_attempts"], + "lost": ["guest /tmp", "guest worktree temp files"], + } + + def cleanup_stale_vms(self): + """清理残留 VM：只清 socket 残留与已死进程对应的 VM。 + + 扫描 state_dir 下 *.sock：登记进程仍活（poll() 为 None） + 的跳过；其余一律走 stop_vm。返回已清理的 VM 名列表； + state_dir 不存在时返回 []，绝不抛异常。基础设施或 + 其它执行器的 socket 不在 state_dir 里，不会被误删； + 同目录下非 chv- 前缀的 socket 也跳过（只认本执行器 + 的 chv- 命名）。 + """ + cleaned = [] + try: + entries = os.listdir(self.state_dir) + except Exception: + return [] + for entry in entries: + if not entry.endswith(".sock"): + continue + name = entry[:-len(".sock")] + if not name.startswith("chv-"): + continue + proc = self._vms.get(name) + if proc is not None: + try: + alive = proc.poll() is None + except Exception: + alive = False + if alive: + continue + if self.stop_vm(name): + cleaned.append(name) + # 登记进程已死但 socket 文件已被删的情况：登记项本身 + # 也是残留，顺手清掉登记，不计入 cleaned（没有 socket + # 可删，不算一次清理）。 + for name in list(self._vms.keys()): + proc = self._vms.get(name) + try: + dead = proc is not None and proc.poll() is not None + except Exception: + dead = True + if dead: + self._vms.pop(name, None) + return cleaned + + # -- 客内执行 ------------------------------------------------------ + def _run_guest(self, task, inner_args, **kwargs): + """SSH 调客内 host.py，返回 Local/Docker 同款结果 dict + （executor 字段改写为 cloud-hypervisor，并附 vm 名与 + api-socket）。runner 异常即原样传播，由 worker 侧统一 + 写回 failed——本类不在这里把异常吞成假成功。""" + self._validate() + ssh_cmd = self.build_ssh_command( + task, self._guest_host_command(*inner_args, **kwargs)) + result = self._call_runner(ssh_cmd, timeout=self.task_timeout) + parsed = _parse_host_result(task["id"], + SimpleNamespaceShim(result)) + parsed["executor"] = "cloud-hypervisor" + parsed["vm"] = self.vm_name(task) + parsed["api_socket"] = self.api_socket(task) + return parsed + + def _goal_result(self, task_id, result): + """客内 host.py goal 输出解析：结论在末尾，解析规则与 + LocalExecutor._goal_result_from_runner 同款语义（met + True=达标 completed，False=未达标 failed，interrupted + 标记/无结论异常退出=interrupted 保住断点续跑活路， + 永久错误无结论=failed 死信），executor 字段为 + cloud-hypervisor。""" + returncode = getattr(result, "returncode", 1) + stdout_full = getattr(result, "stdout", "") or "" + stderr = getattr(result, "stderr", "") or "" + met, goal_error = _parse_goal_output(stdout_full) + interrupted_reason = _goal_interrupted_reason(stdout_full) + refusal = _goal_resume_refusal(stdout_full) + perm_kind, perm_snippet = _classify_goal_error( + stdout_full, stderr) + if refusal is not None and met is None: + status, goal_error = "failed", refusal + elif interrupted_reason is not None: + status = "interrupted" + goal_error = (interrupted_reason + or "host goal 异常中断，断点可续跑") + elif met is None and returncode != 0 \ + and perm_kind == "permanent": + status, goal_error = "failed", perm_snippet or "goal 永久错误" + elif returncode != 0 and met is None: + status = "interrupted" + goal_error = ((stderr.strip()[-500:] if stderr.strip() + else stdout_full.strip()[-500:]) + or "host goal 异常退出（无结论），断点可续跑") + elif met is True: + status = "completed" + elif met is False: + status = "failed" + else: + status = "failed" + goal_error = goal_error or "host goal 输出无结论" + return { + "executor": "cloud-hypervisor", + "task_id": task_id, + "status": status, + "stdout": _goal_stdout_tail(stdout_full), + "stderr": stderr, + "returncode": returncode, + "host_status": ("goal-met" if met is True else + "goal-unmet" if met is False else None), + "goal_met": met, + "goal_error": goal_error, + "usage": _parse_goal_usage(stdout_full), + } + + def _run_guest_goal(self, task, inner_args, **kwargs): + self._validate() + ssh_cmd = self.build_ssh_command( + task, self._guest_host_command(*inner_args, **kwargs)) + return self._goal_result( + task["id"], + self._call_runner(ssh_cmd, timeout=self.task_timeout)) + + def _run_with_vm(self, task, inner_args, goal=False, **kwargs): + """每任务 VM 生命周期：wake_vm（API 模式 + SSH 就绪） + → 跑客内命令 → finally shutdown_vm。wake 失败时不跑 + 任务、直接抛分类失败；跑任务抛异常时 finally 仍关 VM， + 异常原样传播（worker 侧统一写回 failed）。本方法不做 + 内部重试：重试是 worker 的 max_retries 语义，执行器 + 内部重试会把一次失败放大成多次计费与多次 VM 启动。""" + self.wake_vm(task) + try: + if goal: + return self._run_guest_goal(task, inner_args, **kwargs) + return self._run_guest(task, inner_args, **kwargs) + finally: + try: + self.shutdown_vm(task) + except Exception: + pass + + def execute(self, task) -> dict: + """只读执行：客内 host.py status，不直接跑模型回合 + （与 Local/Docker 的 execute 同款语义）。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + return self._run_guest( + task, ("status", "--task", self._host_name(task))) + + def dispatch_task(self, task) -> dict: + """客内 host.py dispatch：建 host 任务并跑一轮 turn。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not task.get("prompt"): + raise ValueError("任务缺少 prompt") + args = ["dispatch", "--name", self._host_name(task), + "--task", task["prompt"]] + repo = task.get("repo") + if repo: + args += ["--repo", repo] + return self._run_with_vm( + task, args, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval"), + sandbox=("read-only" if task.get("plan") else None)) + + def run_task(self, task) -> dict: + """客内 host.py run：在已有 host 任务上再跑一轮 turn。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not task.get("prompt"): + raise ValueError("任务缺少 prompt") + return self._run_with_vm( + task, ("run", "--task", self._host_name(task), + "--prompt", task["prompt"]), + model=task.get("model"), effort=task.get("effort"), + approval=task.get("approval")) + + def goal_task(self, task) -> dict: + """客内 host.py goal：目标循环，轮次规则与 Local 同款 + （缺省 3 轮，>100 轮 loud fail，旧验收形态 0 轮禁用）。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + goal_text = task.get("goal") or task.get("prompt") + if not goal_text: + raise ValueError("任务缺少 prompt") + verify = task.get("verify_command") + rounds = task.get("goal_max_rounds") + if rounds is None: + rounds = 3 + if rounds is not None and rounds > 100: + raise ValueError("goal_max_rounds 超过上限 100 轮") + if rounds == 0 and verify and not task.get("goal_mode"): + raise ValueError( + "旧验收形态 goal_max_rounds=0 表示不限轮次，已禁用") + args = ["goal", "--name", self._host_name(task), + "--task", goal_text] + if verify: + args += ["--verify", verify] + args += ["--max-rounds", str(rounds)] + repo = task.get("repo") + if repo: + args += ["--repo", repo] + return self._run_with_vm( + task, args, goal=True, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval")) + + def goal_resume_task(self, task, goal_text=None, + max_rounds=None) -> dict: + """客内 host.py goal-resume：goal_text=None 为断点续跑 + （不带 --goal-text，沿用原 goal+verify），给 goal_text + 为开新目标（带 --goal-text 与 --max-rounds，默认 0）。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + args = ["goal-resume", "--task", self._host_name(task)] + if goal_text: + args += ["--goal-text", goal_text, "--max-rounds", + str(max_rounds if max_rounds is not None else 0)] + elif max_rounds is not None: + args += ["--max-rounds", str(max_rounds)] + return self._run_with_vm( + task, args, goal=True, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval")) + + def goal_resume_breakpoint_task(self, task) -> dict: + """客内 host.py goal-resume 断点续跑：无 --goal-text、 + 无 --max-rounds，host 沿用原任务从断点轮次续跑。 + 注意 worker 层对非 Local 执行器的断点续跑目前仍明确 + 失败，本方法供直接调用与后续 worker 接线（见类 + docstring），不要理解成 worker 路径已通。""" + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + return self._run_with_vm( + task, ("goal-resume", "--task", self._host_name(task)), + goal=True, model=task.get("model"), + effort=task.get("effort"), approval=task.get("approval")) + + +class WarmPool: + """快照预热池（快速启动设计，接口与状态机在代码里）。 + + 目标：冷启动（preflight + 起 VMM + create + boot + SSH + 就绪）是慢路径；预热池提前把干净 VM 起好并暂停，任务 + 来时 acquire 直接 resume，省掉冷启动的大头。 + 隔离红线（绝不妥协）：预热条目是一次性的干净快照 VM， + acquire 消耗掉它，release 一律销毁，绝不把一个任务用 + 过的 VM（带任务文件、带 agent 痕迹）翻出来给下一个 + 任务/租户复用；池子空了就回落冷启动（wake_vm），绝不 + 因为池空就降级 Docker/Local。 + 状态机：cold（未预热）→ prewarming（wake 中）→ warm + （已暂停待分配）→ assigned（已分配给某任务）→ + running（已 resume）→ recycled/destroyed（release 后）。 + prewarm 的实现：对每个条目调 executor.wake_vm（冷启动） + 再 pause_vm，得到 paused 的干净 VM；mock 单测里 wake/ + pause 均可注入假实现，真机上这一步需要 KVM 宿主（见 + 就绪清单“需 KVM 宿主实测”）。snapshot_url 未设时 + prewarm 返回 []（无基准快照可预热，不是静默造快照）。 + """ + + def __init__(self, executor, size=None, snapshot_url=None): + self.executor = executor + self.size = int(size if size is not None + else getattr(executor, "warm_pool_size", 0)) + self.snapshot_url = snapshot_url \ + if snapshot_url is not None \ + else getattr(executor, "snapshot_url", None) + self._entries = {} # vm 名 -> {"state": ..., "task": ...} + + def warm_count(self): + return sum(1 for e in self._entries.values() + if e["state"] == "warm") + + def prewarm(self, count=None): + """预热到目标数：已有 warm 条目不重复建；每个新条目 + 用一个合成任务名 wake + pause（合成任务 id 为 + warm-<序号>，绝不与真实任务 id 撞名）。返回新增的 + warm VM 名列表。""" + if self.snapshot_url is None: + return [] + target = int(count) if count is not None else self.size + added = [] + i = 0 + while self.warm_count() < target: + i += 1 + name = "warm-%d" % i + if name in self._entries: + continue + task = {"id": name} + self._entries[name] = {"state": "prewarming", + "task": task} + try: + self.executor.wake_vm(task) + self.executor.pause_vm(task) + except Exception: + self._entries.pop(name, None) + raise + self._entries[name] = {"state": "warm", "task": task} + added.append(name) + return added + + def acquire(self, task): + """分配一个预热 VM 给任务：有 warm 条目时消耗一个 + （状态 assigned，返回 mode=warm + 该条目的合成任务 + 名，供调用方 resume 后使用）；无 warm 条目时返回 + mode=cold（调用方回落 wake_vm 冷启动）。""" + for name, entry in list(self._entries.items()): + if entry["state"] == "warm": + entry["state"] = "assigned" + entry["task"] = task + self.executor._emit( + "warm_pool_acquire", task, + "vm=%s" % name) + return {"mode": "warm", "vm": name, + "pool_task": entry["task"]} + return {"mode": "cold", "vm": None, "pool_task": None} + + def release(self, task, pool_vm=None): + """释放：无论 warm 还是 cold 路径，VM 一律销毁 + （executor.shutdown_vm），池条目清掉；绝不把用过 + 的 VM 放回 warm 池。返回 True 表示确实销毁了一个 + 条目或 VM。""" + destroyed = False + if pool_vm and pool_vm in self._entries: + self._entries.pop(pool_vm, None) + destroyed = True + try: + if self.executor.shutdown_vm(task): + destroyed = True + except Exception: + pass + self.executor._emit( + "warm_pool_release", task, + "vm=%s destroyed=%s" % (pool_vm, destroyed)) + return destroyed diff --git a/cloud/server.py b/cloud/server.py index 3d15da0..a2289d3 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -494,6 +494,9 @@ def _task_worktree_for_files(server, task): """ if task.get("executor") == "cloud": return None, (409, "cloud task worktrees are not on this host") + if task.get("executor") == "cloud-hypervisor": + return None, (409, "cloud-hypervisor task worktrees are " + "inside the VM, not on this host") worktree_host = task.get("worktree_host") if worktree_host and worktree_host != socket.gethostname(): return None, (409, "task worktree is on host %s;" @@ -1550,7 +1553,7 @@ def _default_executor(): raw = raw.strip() if raw == "": return None - if raw in ("local", "docker", "cloud"): + if raw in ("local", "docker", "cloud", "cloud-hypervisor"): return raw return "docker" @@ -2492,9 +2495,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, {"error": "name too long"}) return executor = body.get("executor") - if executor is not None and executor not in ("local", "docker"): + if executor is not None and executor not in ( + "local", "docker", "cloud-hypervisor"): self._send_json(400, { - "error": "executor must be local or docker"}) + "error": "executor must be local, docker or " + "cloud-hypervisor"}) return try: branch = _validate_env_branch(body.get("branch")) @@ -4248,12 +4253,15 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, {"error": "priority must be an integer"}) return # executor 可选：None 表示任意 worker 可认领；指定时只能是 - # local/docker/cloud，非法值 400 且不建任务。 + # local/docker/cloud/cloud-hypervisor，非法值 400 且不建任务。 + # cloud-hypervisor 是创建层可见的调度选择（任务内 LLM + # 绝不能自选它，见 sandbox.py 该类 docstring）。 executor = body.get("executor") if executor is not None and executor not in ( - "local", "docker", "cloud"): + "local", "docker", "cloud", "cloud-hypervisor"): self._send_json(400, { - "error": "executor must be local, docker or cloud"}) + "error": "executor must be local, docker, cloud or " + "cloud-hypervisor"}) return # 任务级选项：model 不做白名单（中转目录会变，下拉只是 # 建议值）；effort 必须是 codex 的 effort 档位；approval @@ -4481,6 +4489,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, { "error": "附件不支持 cloud 执行器"}) return + if executor == "cloud-hypervisor": + self._send_json(400, { + "error": "附件不支持 cloud-hypervisor 执行器"}) + return conn0 = self._db() try: brow = conn0.execute( @@ -5893,6 +5905,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if executor == "cloud": return (409, "thread ops are local-executor only;" " cloud task threads are not on this host") + if executor == "cloud-hypervisor": + return (409, "thread ops are local-executor only;" + " cloud-hypervisor task threads live inside" + " the VM, not on this host") worktree_host = task.get("worktree_host") if worktree_host and worktree_host != socket.gethostname(): return (409, "task worktree is on host %s;" diff --git a/cloud/tests/test_cloud_hypervisor.py b/cloud/tests/test_cloud_hypervisor.py new file mode 100644 index 0000000..044b9fb --- /dev/null +++ b/cloud/tests/test_cloud_hypervisor.py @@ -0,0 +1,665 @@ +"""Cloud Hypervisor 执行器 v1 + 生产级唤醒/生命周期测试。 + +只用标准库与 mock runner/launcher/api_client，绝不启动真实 +VM：当前环境没有可跑真 VM 的宿主（生产主机是 CVM、无 +/dev/kvm、无嵌套虚拟化；本 VM 容器内同样无 /dev/kvm）， +所以本文件验证的是：preflight 分类、API 唤醒顺序、VmConfig、 +SSH 就绪、暂停/快照/休眠/关闭/清理、预热池、单任务单 VM +隔离、容量门、磁盘门、worker/server/autoscaler 接线。 +真实 VM 启动、快照恢复、压测一律不在本文件声称已验证。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + WarmPool, + _parse_memory_bytes, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.sandbox import ( + ChvFailure, + CloudHypervisorExecutor, + WarmPool, + _parse_memory_bytes, + ) + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class FakeProc: + """假 VMM 进程：poll() 返回 None=活着，数字=已死。""" + + def __init__(self, alive=True): + self._alive = alive + self.terminated = False + self.killed = False + + def poll(self): + return None if self._alive else 1 + + def terminate(self): + self.terminated = True + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self.killed = True + self._alive = False + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +class ChvCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) + self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) + self.kernel = _write(os.path.join(self.dir, "vmlinux")) + self.rootfs = _write(os.path.join(self.dir, "disk.raw"), + b"R" * 100) + self.ssh_key = _write(os.path.join(self.dir, "id_chv")) + self.kvm = _write(os.path.join(self.dir, "kvm")) + self.cpuinfo = os.path.join(self.dir, "cpuinfo") + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + self.state_dir = os.path.join(self.dir, "state") + self.calls = [] # runner 调用记录 + self.api_calls = [] # api_client 调用记录 + self.launches = [] # launcher 调用记录 + self.audit = [] # audit 事件记录 + + def tearDown(self): + self.tmp.cleanup() + + def make_exec(self, runner=None, api_client=None, launcher=None, + env_extra=None, **overrides): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"} + if env_extra: + env.update(env_extra) + if runner is None: + runner = self._default_runner + if api_client is None: + api_client = self._default_api + if launcher is None: + launcher = self._default_launcher + kwargs = dict( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=runner, launcher=launcher, api_client=api_client, + sleeper=lambda s: None, env=env) + kwargs.update(overrides) + ex = CloudHypervisorExecutor(**kwargs) + ex.audit_callback = lambda e, t, d, k=None: self.audit.append( + (e, t.get("id") if isinstance(t, dict) else t, d, k)) + return ex + + def _default_runner(self, cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd[0] == "ssh": + if cmd[-1] == "true": + return FakeResult(0, "", "") + if "dispatch" in cmd: + return FakeResult(0, json.dumps( + {"status": "completed", "text": "ok-dispatch"}), "") + if "goal" in cmd or "goal-resume" in cmd: + return FakeResult(0, "goal met after rounds\n", "") + if "status" in cmd: + return FakeResult(0, json.dumps( + {"status": "completed", "text": "ok-status"}), "") + return FakeResult(0, "", "") + return FakeResult(0, "", "") + + def _default_api(self, socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + return (200, {}) + + def _default_launcher(self, cmd): + self.launches.append(list(cmd)) + return FakeProc(alive=True) + + def _task(self, task_id="abc123"): + return {"id": task_id, "prompt": "做一个加法函数", + "project_id": "proj1"} + + +# -- preflight --------------------------------------------------------- +class PreflightTest(ChvCase): + def test_all_present_passes(self): + self.assertTrue(self.make_exec().preflight()) + + def test_missing_binary_reports_it(self): + ex = self.make_exec(binary="/tmp/no-such-chv-binary") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "binary_missing") + self.assertIn("cloud-hypervisor", str(ctx.exception)) + + def test_missing_ch_remote_reports_it(self): + ex = self.make_exec(ch_remote="/tmp/no-such-ch-remote") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "binary_missing") + self.assertIn("ch-remote", str(ctx.exception)) + + def test_missing_kvm_reports_it(self): + ex = self.make_exec(kvm_path="/tmp/no-such-kvm") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "kvm_missing") + self.assertIn("/tmp/no-such-kvm", str(ctx.exception)) + + def test_missing_cpu_flag_reports_it(self): + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu sse\n") + with self.assertRaises(ChvFailure) as ctx: + self.make_exec().preflight() + self.assertEqual(ctx.exception.kind, "cpu_flag_missing") + self.assertIn("vmx", str(ctx.exception)) + + def test_missing_kernel_reports_it(self): + ex = self.make_exec(kernel="/tmp/no-such-vmlinux") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "kernel_missing") + + def test_missing_image_reports_it(self): + ex = self.make_exec(rootfs="/tmp/no-such-disk.raw") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "image_missing") + + def test_missing_ssh_key_reports_it(self): + ex = self.make_exec(ssh_key="/tmp/no-such-key") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "ssh_key_missing") + + def test_missing_egress_is_fail_closed(self): + ex = self.make_exec(env_extra={"CHV_EGRESS_PROXY": ""}) + # env_extra 空值会被 __init__ 的 or 链忽略，故改用显式 env： + ex.egress_proxy = None + ex.allow_direct_egress = False + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + self.assertEqual(ctx.exception.kind, "egress_not_configured") + + def test_multiple_missing_lists_all(self): + ex = self.make_exec(binary="/tmp/no-chv", kernel="/tmp/no-kern") + with self.assertRaises(ChvFailure) as ctx: + ex.preflight() + text = str(ctx.exception) + self.assertIn("cloud-hypervisor", text) + self.assertIn("内核", text) + + def test_dispatch_missing_dependency_does_not_call_runner(self): + ex = self.make_exec(binary="/tmp/no-chv") + with self.assertRaises(ChvFailure): + ex.dispatch_task(self._task()) + self.assertEqual(self.calls, []) + + +# -- 配置与命令构造 ----------------------------------------------------- +class ConfigCommandTest(ChvCase): + def test_vm_config_shape_and_memory_bytes(self): + ex = self.make_exec() + cfg = ex.build_vm_config("/tmp/x/disk.raw") + self.assertEqual(cfg["cpus"], {"boot_vcpus": 2, + "max_vcpus": 2}) + self.assertEqual(cfg["memory"]["size"], 2 * 1024 ** 3) + self.assertEqual(cfg["kernel"]["path"], self.kernel) + self.assertEqual(cfg["disks"], [{"path": "/tmp/x/disk.raw"}]) + self.assertEqual(cfg["cmdline"]["args"], + "console=hvc0 root=/dev/vda1 rw") + + def test_memory_parsing(self): + self.assertEqual(_parse_memory_bytes("512M"), 512 * 1024 ** 2) + self.assertEqual(_parse_memory_bytes("1024"), 1024) + self.assertEqual(_parse_memory_bytes("1K"), 1024) + with self.assertRaises(ValueError): + _parse_memory_bytes("abc") + + def test_rootfs_image_alias(self): + ex = self.make_exec(rootfs=None, env_extra={}) + # rootfs=None 时 __init__ 仍从 env 取不到（env 无 CHV_*）， + # 故用 env 含 CHV_IMAGE 的构造验证别名： + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_IMAGE": self.rootfs} + ex2 = CloudHypervisorExecutor( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, ssh_key=self.ssh_key, + kvm_path=self.kvm, cpuinfo_path=self.cpuinfo, + state_dir=self.state_dir, env=env, + sleeper=lambda s: None) + self.assertEqual(ex2.rootfs, self.rootfs) + + def test_manual_vm_command_shape(self): + ex = self.make_exec() + cmd = ex.build_vm_command(self._task()) + self.assertEqual(cmd[0], self.binary) + self.assertIn("--api-socket", cmd) + self.assertIn("--cpus", cmd) + self.assertIn("boot=2", cmd) + self.assertIn("size=2G", cmd) + self.assertIn("--kernel", cmd) + self.assertIn("--disk", cmd) + + def test_ssh_command_has_key_path_not_secret_value(self): + ex = self.make_exec() + cmd = ex.build_ssh_command(self._task(), ["true"]) + joined = " ".join(cmd) + self.assertIn(self.ssh_key, joined) + self.assertIn("BatchMode=yes", joined) + self.assertNotIn("SECRET", joined) + # egress 代理地址在 env 前缀里（不是密钥），且只在代理已设时出现： + self.assertIn("HTTPS_PROXY=http://10.0.0.1:8888", joined) + + def test_isolation_names_and_disks_differ(self): + ex = self.make_exec() + a, b = {"id": "task-a"}, {"id": "task-b"} + self.assertNotEqual(ex.vm_name(a), ex.vm_name(b)) + self.assertNotEqual(ex.api_socket(a), ex.api_socket(b)) + self.assertNotEqual(ex._provision_task_disk(a), + ex._provision_task_disk(b)) + # copy 模式：每任务磁盘是基础镜像的真实副本，内容一致、 + # 路径不同（两个任务绝不写同一文件）： + with open(ex._provision_task_disk(a), "rb") as fh: + self.assertEqual(fh.read(), b"R" * 100) + + def test_disk_quota_gate(self): + ex = self.make_exec(env_extra={"CHV_DISK_GB": "0.00000001"}) + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm(self._task()) + self.assertEqual(ctx.exception.kind, "disk_quota_exceeded") + + def test_capacity_gate_rejects_not_fallback(self): + ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"}) + ex._vms["chv-other"] = FakeProc(alive=True) + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm(self._task()) + self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") + self.assertEqual(self.launches, []) + + +# -- API 唤醒 ----------------------------------------------------------- +class WakeApiTest(ChvCase): + def test_wake_uses_api_mode_not_direct_command(self): + ex = self.make_exec() + out = ex.wake_vm(self._task()) + self.assertEqual(out["state"], "running") + self.assertTrue(out["ready"]) + self.assertEqual(out["mode"], "cold") + # launcher 只带 --api-socket，不带内核/磁盘（API 模式证明）： + self.assertEqual(len(self.launches), 1) + launch = self.launches[0] + self.assertIn("--api-socket", launch) + self.assertNotIn("--kernel", launch) + self.assertNotIn("--disk", launch) + # API 调用顺序：ping → create → boot： + kinds = [(m, p) for m, p, _ in self.api_calls] + self.assertEqual(kinds[0], ("GET", "/vmm.ping")) + self.assertEqual(kinds[1], ("PUT", "/vm.create")) + self.assertEqual(kinds[2], ("PUT", "/vm.boot")) + create_payload = self.api_calls[1][2] + self.assertEqual(create_payload["kernel"]["path"], self.kernel) + self.assertTrue(create_payload["disks"][0]["path"].endswith( + "disk.raw")) + events = [e for e, _, _, _ in self.audit] + self.assertIn("vm_create", events) + self.assertIn("vm_boot", events) + self.assertIn("vm_ready", events) + + def test_wake_api_timeout_kills_proc(self): + def never_ok(socket_path, method, path, payload=None): + self.api_calls.append((method, path, payload)) + return (500, {}) + ex = self.make_exec(api_client=never_ok, api_timeout=3) + proc_holder = [] + + def launcher(cmd): + proc = FakeProc(alive=True) + proc_holder.append(proc) + self.launches.append(list(cmd)) + return proc + ex.launcher = launcher + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm(self._task()) + self.assertEqual(ctx.exception.kind, "api_timeout") + self.assertTrue(proc_holder[0].terminated) + self.assertNotIn("chv-abc123", ex._vms) + + def test_guest_ready_timeout_shuts_down(self): + def ssh_never(cmd, timeout=None): + self.calls.append(list(cmd)) + if cmd[0] == "ssh": + return FakeResult(1, "", "") + return FakeResult(0, "", "") + ex = self.make_exec(runner=ssh_never, ready_timeout=10, + ready_interval=5) + with self.assertRaises(ChvFailure) as ctx: + ex.wake_vm(self._task()) + self.assertEqual(ctx.exception.kind, "guest_not_ready") + # 超时后关 VM：runner 里有 ch-remote shutdown： + self.assertTrue(any("shutdown" in c for c in self.calls)) + + +# -- 暂停 / 快照 / 休眠 / 关闭 / 清理 ----------------------------------- +class LifecycleTest(ChvCase): + def test_pause_resume(self): + ex = self.make_exec() + self.assertTrue(ex.pause_vm(self._task())) + self.assertTrue(ex.resume_vm(self._task())) + kinds = [c[-1] for c in self.calls if c[0] == self.ch_remote] + self.assertEqual(kinds, ["pause", "resume"]) + + def test_restore_then_explicit_resume_order(self): + ex = self.make_exec() + out = ex.restore_vm(self._task(), "file:///snap/1") + self.assertEqual(out["mode"], "restored") + self.assertTrue(out["ready"]) + kinds = [c[-1] if c[-1] != "restore" else c[-1] + for c in self.calls if c[0] == self.ch_remote] + # ch-remote 调用里 restore 必须在 resume 之前： + flat = [] + for c in self.calls: + if c[0] == self.ch_remote: + flat.append(c[-1] if c[-1].startswith("source_url") + else c[-1]) + self.assertIn("resume", flat) + restore_idx = next(i for i, c in enumerate(self.calls) + if c[0] == self.ch_remote + and any(a.startswith("source_url") + for a in c)) + resume_idx = next(i for i, c in enumerate(self.calls) + if c[0] == self.ch_remote and c[-1] == "resume") + self.assertLess(restore_idx, resume_idx) + + def test_hibernate_is_pause_snapshot_shutdown(self): + ex = self.make_exec() + out = ex.hibernate_vm(self._task(), "file:///snap/h1") + self.assertEqual(out["state"], "hibernated") + self.assertEqual(out["snapshot"], "file:///snap/h1") + kinds = [c[-1] for c in self.calls if c[0] == self.ch_remote] + self.assertEqual(kinds[0], "pause") + self.assertTrue(any(a.startswith("source_url") + for c in self.calls for a in c)) + self.assertIn("shutdown", kinds) + # 从休眠唤醒 = restore + resume： + ex2 = self.make_exec() + out2 = ex2.wake_from_hibernate(self._task(), "file:///snap/h1") + self.assertEqual(out2["mode"], "restored") + + def test_stop_dead_proc(self): + ex = self.make_exec() + proc = FakeProc(alive=False) + ex._vms["chv-abc123"] = proc + sock = os.path.join(self.state_dir, "chv-abc123.sock") + os.makedirs(self.state_dir, exist_ok=True) + with open(sock, "w") as _fh: + _fh.write("") + self.assertTrue(ex.stop_vm(self._task())) + self.assertFalse(os.path.exists(sock)) + self.assertNotIn("chv-abc123", ex._vms) + + def test_stop_already_closed(self): + ex = self.make_exec() + self.assertFalse(ex.stop_vm(self._task())) + # 已关仍尽力调过一次 ch-remote shutdown： + self.assertTrue(any("shutdown" in c for c in self.calls)) + + def test_stop_socket_residual(self): + ex = self.make_exec() + sock = os.path.join(self.state_dir, "chv-abc123.sock") + os.makedirs(self.state_dir, exist_ok=True) + with open(sock, "w") as _fh: + _fh.write("") + self.assertTrue(ex.stop_vm(self._task())) + self.assertFalse(os.path.exists(sock)) + + def test_cleanup_skips_live_and_non_chv(self): + ex = self.make_exec() + os.makedirs(self.state_dir, exist_ok=True) + live = os.path.join(self.state_dir, "chv-live.sock") + with open(live, "w") as _fh: + _fh.write("") + ex._vms["chv-live"] = FakeProc(alive=True) + dead = os.path.join(self.state_dir, "chv-dead.sock") + with open(dead, "w") as _fh: + _fh.write("") + other = os.path.join(self.state_dir, "other.sock") + with open(other, "w") as _fh: + _fh.write("") + cleaned = ex.cleanup_stale_vms() + self.assertEqual(cleaned, ["chv-dead"]) + self.assertTrue(os.path.exists(live)) + self.assertTrue(os.path.exists(other)) + self.assertFalse(os.path.exists(dead)) + + def test_cleanup_missing_state_dir(self): + ex = self.make_exec() + self.assertEqual(ex.cleanup_stale_vms(), []) + + def test_recycle_preserves_checkpoints_not_temp(self): + ex = self.make_exec() + out = ex.recycle_vm(self._task()) + self.assertEqual(out["state"], "recycled") + self.assertIn("task_checkpoints", out["preserved"]) + self.assertIn("task_attempts", out["preserved"]) + self.assertIn("guest /tmp", out["lost"]) + + +# -- 预热池 -------------------------------------------------------------- +class WarmPoolTest(ChvCase): + def test_prewarm_acquire_release(self): + ex = self.make_exec() + pool = WarmPool(ex, size=2, snapshot_url="file:///snap/base") + added = pool.prewarm() + self.assertEqual(len(added), 2) + self.assertEqual(pool.warm_count(), 2) + first = pool.acquire(self._task("t1")) + self.assertEqual(first["mode"], "warm") + second = pool.acquire(self._task("t2")) + self.assertEqual(second["mode"], "warm") + third = pool.acquire(self._task("t3")) + self.assertEqual(third["mode"], "cold") + # release 一律销毁，不放回池： + self.assertTrue(pool.release(self._task("t1"), + pool_vm=added[0])) + self.assertEqual(pool.warm_count(), 0) + + def test_prewarm_without_snapshot_is_empty(self): + ex = self.make_exec() + pool = WarmPool(ex, size=2, snapshot_url=None) + self.assertEqual(pool.prewarm(), []) + self.assertEqual(pool.acquire(self._task())["mode"], "cold") + + +# -- dispatch / goal（wake → 跑 → finally 关） --------------------------- +class DispatchTest(ChvCase): + def test_dispatch_wakes_runs_shuts_down(self): + ex = self.make_exec() + out = ex.dispatch_task(self._task()) + self.assertEqual(out["executor"], "cloud-hypervisor") + self.assertEqual(out["status"], "completed") + self.assertIn("ok-dispatch", out["stdout"]) + self.assertEqual(out["vm"], "chv-abc123") + # finally 关 VM：ch-remote shutdown 在 dispatch 之后出现： + shutdown_idx = next(i for i, c in enumerate(self.calls) + if c[0] == self.ch_remote + and c[-1] == "shutdown") + ssh_idx = next(i for i, c in enumerate(self.calls) + if c[0] == "ssh" and "dispatch" in c) + self.assertLess(ssh_idx, shutdown_idx) + self.assertNotIn("chv-abc123", ex._vms) + + def test_goal_met(self): + ex = self.make_exec() + out = ex.goal_task({"id": "g1", "prompt": "做目标", + "goal_mode": True}) + self.assertEqual(out["status"], "completed") + self.assertTrue(out["goal_met"]) + + def test_goal_resume_breakpoint(self): + ex = self.make_exec() + out = ex.goal_resume_breakpoint_task({"id": "g1", + "prompt": "续跑"}) + self.assertEqual(out["status"], "completed") + self.assertTrue(out["goal_met"]) + + def test_execute_is_ssh_status_only(self): + ex = self.make_exec() + out = ex.execute(self._task()) + self.assertEqual(out["executor"], "cloud-hypervisor") + self.assertIn("ok-status", out["stdout"]) + # execute 不 wake（launcher 未调用）： + self.assertEqual(self.launches, []) + + +# -- worker / autoscaler / server 接线 ----------------------------------- +class WiringTest(ChvCase): + def test_worker_executor_name_and_cli_builder(self): + from cloud import worker as worker_mod + from cloud.sandbox import CloudHypervisorExecutor as CHV + ex = self.make_exec() + w = worker_mod.Worker.__new__(worker_mod.Worker) + w.executor = ex + self.assertEqual(w._executor_name(), "cloud-hypervisor") + + class Args: + executor = "cloud-hypervisor" + built = worker_mod._build_cli_executor(Args(), None) + self.assertIsInstance(built, CHV) + + def test_autoscaler_passthrough(self): + from cloud.autoscaler import Autoscaler + scaler = Autoscaler("/tmp/no.db", [], executor="cloud-hypervisor") + cmd = scaler._worker_command("w1") + self.assertIn("--executor", cmd) + self.assertEqual(cmd[cmd.index("--executor") + 1], + "cloud-hypervisor") + scaler2 = Autoscaler("/tmp/no.db", + ["--executor", "docker"], + executor="cloud-hypervisor") + cmd2 = scaler2._worker_command("w1") + self.assertEqual(cmd2.count("--executor"), 1) + self.assertEqual(cmd2[cmd2.index("--executor") + 1], "docker") + + def test_server_default_executor(self): + from cloud.server import _default_executor + with mock.patch.dict(os.environ, + {"CLOUD_DEFAULT_EXECUTOR": + "cloud-hypervisor"}): + self.assertEqual(_default_executor(), "cloud-hypervisor") + with mock.patch.dict(os.environ, + {"CLOUD_DEFAULT_EXECUTOR": "bogus"}): + self.assertEqual(_default_executor(), "docker") + + +class ServerWhitelistTest(unittest.TestCase): + """server 白名单端到端：真实 HTTP + 临时库 + 临时端口。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + try: + from cloud.auth import AuthStore + from cloud.server import create_server + except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.server import create_server + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + old = os.environ.get("CODEX_GLOBAL_ADMINS") + self._old_admins = old + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201, self.project) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def test_task_executor_cloud_hypervisor_accepted(self): + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x", + "executor": "cloud-hypervisor"}, self.owner_key) + self.assertEqual(status, 201, task) + self.assertEqual(task["executor"], "cloud-hypervisor") + + def test_task_executor_bogus_rejected(self): + status, body = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x", + "executor": "bogus"}, self.owner_key) + self.assertEqual(status, 400) + self.assertIn("cloud-hypervisor", body["error"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 64a9824..83f8d2e 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -195,7 +195,8 @@ from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job, mark_pr_job_needs_local_worker) from cloud.queue import budget_exceeded, ready_tasks from cloud.repos import RepoError, checkout_dir, resolve_repo -from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor +from cloud.sandbox import CloudExecutor, CloudHypervisorExecutor, \ + DockerExecutor, LocalExecutor from cloud.uploads import upload_root as _upload_root from cloud.sandbox import _default_runner as _sandbox_default_runner from cloud.caps import effective_capacity, resolve_capacity @@ -506,6 +507,24 @@ class Worker: self.executor.ignore_host_budget = self.ignore_host_budget except Exception: pass + # CHV 审计转发：VM 生命周期事件（启动/暂停/恢复/快照/ + # 销毁/失败）经本回调进 AuthStore 审计（_audit 本身 + # best-effort，失败不影响任务）。时间线展示 VM 事件时 + # 读审计表即可，不另起一套 VM 事件表；成本/配额仍走 + # worker 现有 quota 语义，本回调绝不记账。 + if isinstance(self.executor, CloudHypervisorExecutor): + def _chv_audit(event, task, detail, kind=None): + try: + project_id = task.get("project_id") \ + if isinstance(task, dict) else None + self._audit(project_id, "chv_vm_" + event, + detail or "") + except Exception: + pass + try: + self.executor.audit_callback = _chv_audit + except Exception: + pass self._cost_store = cost_store self._cancel_requested = set() # 真取消用：process_factory 可注入（单测用它返回一个假长跑 @@ -2409,12 +2428,15 @@ class Worker: # -- 缺 key 门 ------------------------------------------------------ def _executor_name(self): """执行器名：DockerExecutor -> docker，CloudExecutor -> cloud， - 其它 -> local。cloud 绝不能回落成 local，否则 cloud worker + CloudHypervisorExecutor -> cloud-hypervisor，其它 -> local。 + cloud/cloud-hypervisor 绝不能回落成 local，否则该 worker 会认领 local 池的任务，池过滤就失效了。""" if isinstance(self.executor, DockerExecutor): return "docker" if isinstance(self.executor, CloudExecutor): return "cloud" + if isinstance(self.executor, CloudHypervisorExecutor): + return "cloud-hypervisor" return "local" def key_available(self): """只查存在性：名在环境里且值非空即 True；值本身绝不返回。""" @@ -6756,7 +6778,12 @@ def _build_cli_executor(args, host_py): docker 时返回 DockerExecutor（env_key 名用 Worker 的解析规则， 值本身绝不从这里传——DockerExecutor 只查存在性，-e 只传名）， cloud 时返回 CloudExecutor（通用 HTTP runner，未配 endpoint - 时执行即明确失败，绝不降级）。""" + 时执行即明确失败，绝不降级），cloud-hypervisor 时返回 + CloudHypervisorExecutor（配置走 CHV_* 环境变量，缺依赖时 + 执行即明确失败，绝不降级；任务内 LLM 不可自选它，见 + sandbox.py 该类 docstring）。""" + if getattr(args, "executor", "local") == "cloud-hypervisor": + return CloudHypervisorExecutor() if getattr(args, "executor", "local") == "cloud": return CloudExecutor() if getattr(args, "executor", "local") != "docker": @@ -6834,11 +6861,15 @@ def main(argv=None): "的 max_daily_turns 预算照常生效（双重门）") parser.add_argument("--max-tasks", type=int, default=None) parser.add_argument("--executor", default="local", - choices=["local", "docker", "cloud"], + choices=["local", "docker", "cloud", + "cloud-hypervisor"], help="执行器：local 调 host.py 子进程；docker" " 调 DockerExecutor（无 docker 时任务" "明确失败，不静默降级）；cloud 调通用" - " HTTP runner（需 CLOUD_RUNNER_ENDPOINT）") + " HTTP runner（需 CLOUD_RUNNER_ENDPOINT）；" + "cloud-hypervisor 调 CloudHypervisorExecutor" + "（VM 内经 SSH 调 host.py，配置走 CHV_*" + " 环境变量，缺依赖时任务明确失败）") parser.add_argument("--worker-id", default=None, help="多 worker 池身份（默认：主机名-进程号；" "生产部署应配稳定 id，否则重启后旧行变 stale）") diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md new file mode 100644 index 0000000..afa8a64 --- /dev/null +++ b/docs/chv-production-readiness.md @@ -0,0 +1,173 @@ +# Cloud Hypervisor 执行器生产就绪清单 + +> 状态图例：✅ 代码已完成 · 🧪 mock 测试已覆盖（无真 VM） · +> ⏳ 需 KVM 宿主实测 · 🏗️ 需镜像构建 · 📈 需压测 · 🔒 需安全审计 +> +> 本清单是 PR 的一部分，不是事后补的口号。凡是标 ⏳/🏗️/📈/🔒 +> 的项，在本轨内一律不声称已验证、已建成、已通过。 + +## 0. 交付点卡在哪一关（必读，先于逐项清单） + +- **CI 门禁（未过，交付点 = PR OPEN）**：CI 必须通过才能合并。 + 此前全仓 CI 因 GitHub 账单问题 0 步骤失败；若账单未恢复、 + CI 仍失败，本 PR 只到 OPEN，不合并，更不为变绿改 CI 配置 + 或绕过检查。 +- **真 VM 门禁（未过，执行器默认关闭）**：生产启用（把真实任务 + 路由到 cloud-hypervisor、部署到生产主机）必须先有带 /dev/kvm + 的宿主并完成真实 VM 启动、快照恢复、任务端到端验证。 + 当前事实：生产主机 124.222.165.170 本身就是 CVM + （systemd-detect-virt=kvm，DMI product=CVM），/dev/kvm 不存在、 + 无 vmx/svm 标志、无 kvm 模块、未装 cloud-hypervisor，无嵌套 + 虚拟化透传；本 VM 容器内同样无 /dev/kvm。 + 故真 VM、快照恢复、压测在本轨内一律不声称已验证；执行器 + 默认关闭（server 默认仍是 docker，worker 默认仍是 local， + CHV 只在任务创建时显式选 `executor=cloud-hypervisor` 且有 + CHV worker 认领时才跑），不进入生产任务路由。 +- **还差什么才能真正上生产**（按顺序）：① 用户侧开通带 /dev/kvm + 的裸金属或开启嵌套虚拟化的实例；② 在该宿主上构建并发布 + chv-guest 镜像（见 §4）；③ 在该宿主上跑通真实 VM 启动、 + 快照恢复、任务端到端、预热池与压测；④ 安全审计通过（§6）； + ⑤ CI 账单恢复且本 PR CI 变绿后，按当时 main 头复核 diff 再合并、 + 再部署。缺任何一项，都不把 CHV 推上生产。 + +## 1. 单任务单 VM 隔离 + +- ✅🧪 一个任务一个 VM：dispatch/run/goal 均走 wake → 跑 → finally + shutdown，同一任务重复 wake 先关旧 VM；VM 名、api-socket、 + 每任务磁盘路径均按任务派生，互不相同（测试断言）。 +- ✅🧪 每任务磁盘独立：copy 模式复制基础镜像为每任务磁盘； + reuse 模式仅单租户实验，生产禁用（代码与文档双写）。 +- ✅🧪 租户无共享路径：执行器不挂载宿主根、宿主代码目录或含 + 密钥目录；客机代码在镜像内 `/app`，不在宿主共享挂载上。 +- ✅🧪 密钥不进镜像/代码/prompt：SSH argv 只有 key 文件路径； + 模型 key 由客机 cloud-init 预置的环境变量名注入；测试断言 + 命令中不含哨兵密钥值。 +- 🔒 需安全审计：VM 逃逸面（virtio 设备、tap 桥接、SSH 配置）、 + 客机 root 提权面、租户间东西向隔离的宿主网络配置，须由 + 安全审计逐项确认，代码自查不能代替审计。 + +## 2. 快速启动：预热快照池 + 镜像缓存 + +- ✅🧪 快照/恢复接口：snapshot_vm / restore_vm（restore 后显式 + resume，见 §3），WarmPool 接口与状态机（cold → prewarming → + warm → assigned → running → recycled/destroyed）。 +- ✅🧪 冷启动回落：池空时 acquire 返回 cold，调用方回落 wake_vm + 冷启动，绝不降级 Docker/Local。 +- ✅🧪 预热条目一次性：acquire 消耗、release 一律销毁，绝不跨 + 任务复用脏 VM。 +- ⏳ 需 KVM 宿主实测：预热池的真实冷启动耗时、resume 耗时、 + 池大小与命中率、快照来源一致性，均需真机数据，mock 数字 + 不能当性能结论。 +- 📈 需压测：多 VM 并行预热/分配/销毁的宿主 CPU/内存/磁盘 + 压力、快照存储增长，均未测。 + +## 3. 完整生命周期 + +- ✅🧪 create / boot：API 模式 wake_vm（vmm.ping → vm.create → + vm.boot → SSH 就绪）。 +- ✅🧪 pause-resume（sleep-wake）：pause_vm / resume_vm（ch-remote）。 +- ✅🧪 snapshot-restore：restore_vm（restore + 显式 resume + + SSH 就绪），恢复后默认 paused 的坑已在代码与测试里钉住。 +- ✅🧪 hibernate：CHV 无原生 hibernate 端点，本轨用 pause + + snapshot + shutdown 等价实现，文档与代码双写“等价实现”， + 不声称接了原生端点；唤醒走 restore_vm。 +- ✅🧪 shutdown-destroy / recycle：shutdown_vm（ch-remote + shutdown + 杀进程 + 删 socket）、recycle_vm（保留 + task_checkpoints/task_attempts，丢失 guest /tmp 与未拷出 + 的 worktree 临时物，绝不声称恢复临时文件）。 +- ⏳ 需 KVM 宿主实测：上述全部生命周期在真机上的状态转换、 + 快照恢复后 guest 状态一致性、休眠后文件/worktree 保留， + 均未在真机验证。 + +## 4. 完整开发环境（客机镜像） + +- ✅ 客机镜像内容清单：`chv-guest/README.md` 按 + `cloud/docker/Dockerfile` 能力对齐（Python 3.12、git、 + pandoc/poppler/latex、python-docx/openpyxl/pptx/PyMuPDF/ + Pillow、codex 0.164.0-alpha.9 成套、Playwright + Chromium、 + openssh-server、cloud-init）。 +- ✅ 构建脚本骨架：`chv-guest/build.sh`（参数校验、版本 tag + 格式、KVM 宿主检查、TODO 步骤清单）。 +- 🏗️ 需镜像构建：镜像本身未构建、未发布；BUILDINFO 版本核对、 + 浏览器版本钉死、中文 LaTeX 单独一版，均待构建时做。 +- ⏳ 需 KVM 宿主实测：镜像在真 VM 里启动、codex/Playwright/ + git/文档工具链逐项跑通，均未验证。 + +## 5. 资源与成本 + +- ✅🧪 每 VM 配额：CPU/内存进 VmConfig（CHV_CPU/CHV_MEM）， + 磁盘配额门（CHV_DISK_GB，超限拒绝），宿主容量门 + （CHV_MAX_VMS，超限明确拒绝排队，不降级）。 +- ✅ 超时与最大轮次：SSH 单次超时（CHV_TASK_TIMEOUT）+ + worker task_timeout 外层护栏；goal 最大轮次 100（与 Local + 同款），执行器内部不重试（重试是 worker max_retries 语义）。 +- ✅ 成本复用：不另起一套计费；worker 现有 quota/成本语义 + （per-user USD 配额、quota 看门狗、成本写库）仍是唯一口径， + 执行器只把 host.py 用量原样返回。 +- ✅ 空闲休眠省成本：hibernate_vm（pause + snapshot + + shutdown）+ WarmPool 释放即销毁，空闲 VM 不占 vCPU/内存 + （快照只占磁盘，见 §2 压测项）。 +- ✅ autoscaler：按 executor 过滤队列深度（已有）+ 本轨透传 + `--executor` 给拉起的 worker（否则 CHV 池永远饿死）。 +- 📈 需压测：配额下的真实资源占用、超限拒绝的排队行为、 + 成本与配额在长任务下的实际表现，均未测。 + +## 6. 网络与治理 + +- ✅🧪 受控 egress：默认 fail-closed（CHV_EGRESS_PROXY 未设且 + 未显式允许直连时 preflight 直接失败）；客机命令加 env + 代理前缀，出口只走白名单代理（Tinyproxy 思路复用）。 +- ⏳🔒 需宿主网络配置 + 安全审计：tap 桥接是否真的只路由到 + 白名单代理、VM 能否绕过代理直连外网，是宿主网络配置问题， + 执行器代码不能自证，必须在 KVM 宿主上实测并审计。 +- ✅ 审批门：任务 approval 经 `_approval_cli_args` 进客内 host.py + 顶层参数，VM 内 agent 不能自改；delete/deploy/spend 类动作 + 仍走 server 的 approval/merge_gate，不由 VM 内 agent 自批 + （本轨不改 approval/merge_gate 语义，只复用）。 +- ✅ 工具与系统调用范围收敛：VM 边界本身就是系统调用边界 + （客机独立内核）；plan 任务走 read-only 沙箱参数；最大步数 + （goal 100 轮）、时间（task_timeout）、重试（worker 1 次） + 超限即停并审计（vm_failed 带 kind）。 + +## 7. 可观测与审计 + +- ✅🧪 VM 事件审计：create/boot/ready/pause/resume/snapshot/ + restore/shutdown/destroy/failed/recycled/warm_pool_acquire/ + release 均经 audit_callback 发出；worker 接线后转 AuthStore + 审计（_audit 本身 best-effort）。 +- ✅🧪 失败分类：ChvFailure 带 kind（kvm_missing/binary_missing/ + kernel_missing/image_missing/ssh_key_missing/cpu_flag_missing/ + egress_not_configured/api_timeout/guest_not_ready/vm_create_failed/ + vm_boot_failed/host_capacity_exceeded/disk_quota_exceeded）， + 不笼统 failed。 +- ✅ 断点对齐：VM 快照/休眠与现有 task_checkpoints/task_attempts + 语义对齐（recycle 保留两者，丢失临时文件）；goal 断点续跑 + 方法（goal_resume_breakpoint_task）已实现，worker 层对非 + Local 执行器的断点续跑目前仍明确失败（见执行器 docstring）， + 后续 worker 接线是单独一轨，不在本轨假装已通。 +- ⏳ 需 KVM 宿主实测：审计事件在真机任务 timeline 里的完整 + 呈现、失败分类与真实故障的对应关系，均未验证。 + +## 8. 多 VM 并发与容量 + +- ✅🧪 worker 注册：`_executor_name()` 返回 cloud-hypervisor， + worker register/heartbeat 已上报 executor + capacity（现有 + 机制复用，无需新表）。 +- ✅🧪 多 VM 并行：`_vms` 登记 + 锁保护，wake 按任务名登记， + 不同任务互不覆盖；容量门在 wake 前检查。 +- ✅🧪 残留 VM 清理：cleanup_stale_vms 处理已死/已关/socket + 残留三种情况，只清 chv- 前缀 socket。 +- ✅🧪 容量不足明确拒绝：kind=host_capacity_exceeded，调用方 + 排队/拒绝，绝不降级 Docker/Local（测试断言 launcher 未调用）。 +- 📈 需压测：多 VM 并行下的宿主容量、VM 启动并发上限、残留 + 清理在高压下的表现，均未测。 + +## 9. 默认关闭与生产启用条件（重申） + +- ✅ 默认关闭：server 默认执行器仍是 docker，worker 默认 local； + CHV 只在显式选 executor 且有 CHV worker 时才跑；无 CHV_* + 配置时 preflight 直接失败。 +- ⏳ 生产启用前置（全部未完成）：带 /dev/kvm 的宿主开通 → + chv-guest 镜像构建发布 → 真实 VM 启动/快照恢复/任务端到端 + 验证 → 压测 → 安全审计 → CI 变绿后按当时 main 头复核 diff + 再合并部署。缺一不可。