feat: CHV 执行器优化（拆模块/去重复/SSH 钉死/池持久化/中途死亡分类）

- 拆模块：CHV 从 cloud/sandbox.py 巨类拆到 cloud/chv/（executor/
  pool/config/api_client/ssh_transport/disk/errors），sandbox 只留
  兼容 re-export，旧导入路径与新路径是同一对象
- 去重复：goal 解析、host 顶层参数、runner 超时抽成共享 helper，
  Local 与 CHV 在矩阵输入下结论全等（Docker goal-resume 变体
  有意保留内联版，见 PR 描述）
- 安全：SSH host key 按 VM 钉死（state_dir/<vm>/known_hosts，
  ssh-keyscan 首次捕获 TOFU），废除 no + /dev/null；CHV_SSH_USER
  空串与 CHV_SSH_PORT 越界直接拒绝，不静默 fallback
- execute 语义：只读已运行 VM，未运行报 vm_not_running，不为
  status 起 VM
- 预热池：registry 落盘、重启载入只当孤儿、reconcile 清理、
  损坏 quarantine；条目仍一次性，绝不复用脏 VM
- 长任务：VM 中途死亡判 interrupted（断点可续跑）；中途心跳
  经评估不做（无客内进度信号可采）
- #215 未合并：只列复用清单，不复制模块

测试：原 42 项 + 新 24 项全过；相关旧套件与 cloud 全量 1496 项
全过（skipped=4）。验证仍是 mock + 干跑 + 单元测试，未做真机、
压测、安全审计；堆叠于 #216 之上，base 为 feat-cloud-hypervisor。

cloud/README.md

@@ -1288,8 +1288,9 @@ 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` 缩写）：
+- CloudHypervisorExecutor（生产级 v1，`cloud/chv/`，旧路径
+  `cloud/sandbox.py` 只留兼容 re-export，执行器名
+  `cloud-hypervisor`，绝不用 `cloud` 缩写）：
   单任务单 VM：一个任务一个 VM、独立内核/文件系统/网络，
   任务结束销毁，租户之间无共享路径、无东西向互访；VM 内
   agent 可有 root，但只能影响本 VM。密钥按现有惯例走运行
@@ -1307,7 +1308,24 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   用 pause + snapshot + shutdown 等价实现，恢复走 restore。
   预热池 `WarmPool`：预热条目是一次性干净快照 VM，acquire
   消耗、release 一律销毁，绝不跨任务复用脏 VM；池空回落
-  冷启动，绝不降级 Docker/Local。关闭与清理：任务结束/
+  冷启动，绝不降级 Docker/Local；池状态另落
+  `state_dir/warm-pool.json` registry，worker 重启后载入的
+  条目一律当孤儿、只能 `reconcile_orphans()` 清理，绝不
+  重新发给任务，registry 损坏时 quarantine 成 `.corrupt` 并
+  从空池开始。SSH host key 按 VM 钉死（#216 的
+  `StrictHostKeyChecking=no` + `/dev/null` 已废除）：每 VM
+  在 `state_dir/<vm>/known_hosts` 有独立文件，首次启动经
+  ssh-keyscan 在本机回环端口捕获 key 写入该文件（首次信任
+  TOFU，只防首次之后的替换，不验证首次本身），此后 SSH
+  一律 `StrictHostKeyChecking=yes` + 该文件校验；客机用户
+  （`CHV_SSH_USER`，默认 root，显式空串直接拒绝）与端口
+  （`CHV_SSH_PORT`，默认 2222，1-65535）为显式配置。
+  `execute()` 只读已运行 VM：未运行报 `vm_not_running`
+  分类失败，不为一次 status 起 VM。长任务中途死亡
+  （传输异常、或结果已 failed 且登记 VMM 进程已死）判
+  `interrupted` 而非笼统 failed；中途心跳经评估未做
+  （host.py 单次 SSH 长回合无客内进度信号可采，另起轮询
+  只能证明 VM 活着）。关闭与清理：任务结束/
   失败/取消时 `shutdown_vm`（ch-remote shutdown + 杀进程 +
   删 socket），`cleanup_stale_vms` 处理已死/已关/socket 残留
   三种情况。审计：VM 的 create/boot/ready/pause/resume/

cloud/chv/__init__.py

@@ -0,0 +1,18 @@
+"""Cloud Hypervisor 执行器包（从 cloud/sandbox.py 巨类拆出）。
+
+公开名与 #216 一致，cloud/sandbox.py 会 re-export 它们，
+保证旧导入路径可用：
+- CloudHypervisorExecutor：执行器本体（executor.py）
+- WarmPool：预热池 + registry 持久化与孤儿清理（pool.py）
+- ChvFailure：分类失败（errors.py）
+
+子模块分工见 executor.py 头注释。本包顶层 import 不触
+cloud.sandbox（执行器方法内按需 import 共享 helper），
+所以任何导入顺序都不会循环。
+"""
+
+from cloud.chv.errors import ChvFailure
+from cloud.chv.executor import CloudHypervisorExecutor
+from cloud.chv.pool import WarmPool
+
+__all__ = ["ChvFailure", "CloudHypervisorExecutor", "WarmPool"]

cloud/chv/api_client.py

@@ -0,0 +1,74 @@
+"""Cloud Hypervisor REST API 传输（叶子模块，不依赖
+cloud.sandbox，避免循环导入）。
+
+只用标准库 raw HTTP over Unix socket，无第三方依赖。
+"""
+
+import json
+import os
+import socket
+
+
+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
+
+
+def normalize_api_result(result):
+    """API 调用归一化：fake api_client 可返回 (status, body)
+    或 {"status":..., "body":...} 或带 .status 的对象，本
+    函数统一成 (status:int, body)。"""
+    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

cloud/chv/config.py

@@ -0,0 +1,146 @@
+"""Cloud Hypervisor 执行器配置解析（叶子模块，不依赖
+cloud.sandbox，避免循环导入）。
+
+职责只有两件：内存字符串解析，以及把构造参数 + 环境变量
+解析成一份显式配置。参数优先于环境变量、环境变量优先于
+默认值——与 #216 原 __init__ 的 `or` 链完全一致，只收紧了
+两处原来会静默 fallback 的 nonsense 输入（见下文），收紧点
+在 PR 描述里明写、单测钉住：
+
+- CHV_SSH_USER 显式传空字符串：原来 `or` 链会静默 fallback
+  成 root（调用方以为没给用户、实际跑了 root），现在直接
+  ValueError。
+- CHV_SSH_PORT 显式传 0、或解析后超出 1-65535：原来 0 会
+  静默 fallback 成 2222、其它越界值会原样进 ssh 命令，现在
+  直接 ValueError。
+
+与 #215 的关系（只列清单，不复制模块）：#215 未合并，
+本轨不假装复用它。#215 合并后应再复用的点：超时类
+（api_timeout/ready_timeout/ready_interval/task_timeout →
+timeouts 模块）、门禁类（preflight fail-fast、容量门、磁盘
+配额门、egress fail-closed → gates 模块）、配额类
+（max_vms、disk_gb、任务成本走 worker quota → quota 模块）、
+隔离类（单任务单 VM、磁盘 copy 模式、egress 代理、SSH 钉死
+→ sandbox_policy 模块）。见 PR 描述的复用清单。
+"""
+
+import re
+
+
+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()]
+
+
+class ChvConfig:
+    """解析后的 CHV 配置：纯数据对象，字段名与
+    CloudHypervisorExecutor 原 __init__ 的属性名一一对应，
+    执行器 __init__ 直接把字段抄到 self 上，旧测试读属性
+    的方式全部不变。"""
+
+    def __init__(self, **fields):
+        for key, value in fields.items():
+            setattr(self, key, value)
+
+
+def build_chv_config(params, env):
+    """把显式构造参数（params，None=未给）与环境变量（env，
+    默认为 os.environ 的调用方传入 dict）解析成 ChvConfig。
+
+    显式参数只有 __init__ 本来就有的那些；max_vms、磁盘模式、
+    磁盘配额、egress、快照、预热池目标数本来就没有显式参数，
+    仍只走环境变量（与 #216 一致，不悄悄加新参数面）。"""
+    env = env if env is not None else {}
+    binary = params.get("binary") or env.get("CHV_BINARY") \
+        or "cloud-hypervisor"
+    kernel = params.get("kernel") or env.get("CHV_KERNEL")
+    rootfs = params.get("rootfs") or params.get("image") \
+        or env.get("CHV_ROOTFS") or env.get("CHV_IMAGE")
+    cpus = str(params.get("cpus") or env.get("CHV_CPU") or "2")
+    memory = params.get("memory") or env.get("CHV_MEM") or "2G"
+    state_dir = params.get("state_dir") or env.get("CHV_STATE_DIR") \
+        or "/tmp/chv-vms"
+    ssh_key = params.get("ssh_key") or env.get("CHV_SSH_KEY")
+    if params.get("ssh_user") is not None \
+            and params.get("ssh_user") == "":
+        raise ValueError("CHV_SSH_USER 不能为空字符串")
+    ssh_user = params.get("ssh_user") or env.get("CHV_SSH_USER") \
+        or "root"
+    if params.get("ssh_port") is not None:
+        raw_port = params.get("ssh_port")
+    else:
+        raw_port = env.get("CHV_SSH_PORT") or 2222
+    try:
+        ssh_port = int(raw_port)
+    except (TypeError, ValueError):
+        raise ValueError("CHV_SSH_PORT 必须是整数")
+    if not 1 <= ssh_port <= 65535:
+        raise ValueError("CHV_SSH_PORT 必须在 1-65535 之间")
+    guest_host_py = params.get("guest_host_py") or "/app/host.py"
+    cmdline = params.get("cmdline") or env.get("CHV_CMDLINE") \
+        or "console=hvc0 root=/dev/vda1 rw"
+    net = params.get("net") or env.get("CHV_NET")
+    net_json = params.get("net_json") or env.get("CHV_NET_JSON")
+    kvm_path = params.get("kvm_path") \
+        if params.get("kvm_path") is not None else "/dev/kvm"
+    cpuinfo_path = params.get("cpuinfo_path") \
+        if params.get("cpuinfo_path") is not None else "/proc/cpuinfo"
+    ch_remote = params.get("ch_remote") or env.get("CHV_REMOTE") \
+        or "ch-remote"
+    model = params.get("model")
+    try:
+        max_vms = int(env.get("CHV_MAX_VMS") or 4)
+    except ValueError:
+        raise ValueError("CHV_MAX_VMS 必须是整数")
+    disk_mode = env.get("CHV_DISK_MODE") or "copy"
+    if disk_mode not in ("copy", "reuse"):
+        raise ValueError(
+            "CHV_DISK_MODE 只能是 copy 或 reuse：%s" % disk_mode)
+    disk_gb_raw = env.get("CHV_DISK_GB")
+    disk_gb = float(disk_gb_raw) if disk_gb_raw else None
+    egress_proxy = env.get("CHV_EGRESS_PROXY")
+    allow_direct_egress = env.get("CHV_ALLOW_DIRECT_EGRESS") == "1"
+    task_timeout_raw = env.get("CHV_TASK_TIMEOUT")
+    task_timeout = float(task_timeout_raw) \
+        if task_timeout_raw else None
+    snapshot_url = env.get("CHV_SNAPSHOT_URL")
+    try:
+        warm_pool_size = int(env.get("CHV_WARM_POOL_SIZE") or 0)
+    except ValueError:
+        raise ValueError("CHV_WARM_POOL_SIZE 必须是整数")
+    try:
+        api_timeout = float(
+            params.get("api_timeout")
+            if params.get("api_timeout") is not None
+            else env.get("CHV_API_TIMEOUT") or 30)
+        ready_timeout = float(
+            params.get("ready_timeout")
+            if params.get("ready_timeout") is not None
+            else env.get("CHV_READY_TIMEOUT") or 120)
+        ready_interval = float(
+            params.get("ready_interval")
+            if params.get("ready_interval") is not None
+            else env.get("CHV_READY_INTERVAL") or 5)
+    except ValueError:
+        raise ValueError("CHV_*_TIMEOUT/INTERVAL 必须是数字")
+    if ready_interval <= 0:
+        ready_interval = 5
+    return ChvConfig(
+        binary=binary, kernel=kernel, rootfs=rootfs, cpus=cpus,
+        memory=memory, state_dir=state_dir, ssh_key=ssh_key,
+        ssh_user=ssh_user, ssh_port=ssh_port,
+        guest_host_py=guest_host_py, cmdline=cmdline, net=net,
+        net_json=net_json, kvm_path=kvm_path,
+        cpuinfo_path=cpuinfo_path, ch_remote=ch_remote, model=model,
+        max_vms=max_vms, disk_mode=disk_mode, disk_gb=disk_gb,
+        egress_proxy=egress_proxy,
+        allow_direct_egress=allow_direct_egress,
+        task_timeout=task_timeout, snapshot_url=snapshot_url,
+        warm_pool_size=warm_pool_size, api_timeout=api_timeout,
+        ready_timeout=ready_timeout, ready_interval=ready_interval)

cloud/chv/disk.py

@@ -0,0 +1,43 @@
+"""Cloud Hypervisor 每任务磁盘供给（叶子模块）。
+
+单任务单 VM 的文件系统独立性就靠这里：copy 模式每任务
+复制一份基础镜像，两个任务绝不写同一文件；reuse 模式
+直接返回基础镜像路径（仅单租户实验，生产禁用，见
+CloudHypervisorExecutor 的类 docstring）。
+"""
+
+import os
+import shutil
+
+from cloud.chv.errors import ChvFailure
+
+
+def check_disk_quota(rootfs, disk_gb):
+    """磁盘配额门：CHV_DISK_GB 已设且基础镜像超过该 GB 数
+    即拒绝（kind=disk_quota_exceeded）。未设则不检查；
+    镜像读不到时不拦（preflight 自会报 image_missing，
+    这里不抢它的错）。"""
+    if disk_gb is None or not rootfs:
+        return
+    try:
+        size = os.path.getsize(rootfs)
+    except OSError:
+        return
+    if size > disk_gb * 1024 ** 3:
+        raise ChvFailure(
+            "disk_quota_exceeded",
+            "磁盘镜像 %d 字节超过配额 %.1f GB" % (size, disk_gb))
+
+
+def provision_task_disk(state_dir, vm_name, rootfs, disk_mode):
+    """每任务磁盘路径：copy 模式复制到
+    state_dir/<vm>/disk.raw（已存在则复用，不重复复制）；
+    reuse 模式直接返回基础镜像路径。"""
+    if disk_mode == "reuse":
+        return rootfs
+    task_dir = os.path.join(state_dir, vm_name)
+    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(rootfs, task_disk)
+    return task_disk

cloud/chv/errors.py

@@ -0,0 +1,12 @@
+"""Cloud Hypervisor 执行器的分类失败（叶子模块，不依赖
+cloud.sandbox，避免循环导入）。"""
+
+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

cloud/chv/pool.py

@@ -0,0 +1,257 @@
+"""快照预热池 WarmPool（含 state_dir registry 持久化与孤儿清理）。
+
+隔离红线（与 #216 一致，绝不妥协）：预热条目是一次性的
+干净快照 VM，acquire 消耗掉它，release 一律销毁，绝不把
+一个任务用过的 VM（带任务文件、带 agent 痕迹）翻出来给
+下一个任务/租户复用；池子空了就回落冷启动（wake_vm），
+绝不因为池空就降级 Docker/Local。
+
+Registry（本轨新增）：
+- 文件：state_dir/warm-pool.json（registry_path 可覆盖），
+  内容为 {<合成名>: {"state", "vm", "task_id", "snapshot_url"}}。
+  每次 prewarm/acquire/release 后立即落盘（临时文件 + 原子
+  rename），worker 崩溃后重启仍能发现上一世的条目。
+- 载入时绝不复用：新池实例从 registry 读到的条目一律标为
+  orphan（不计入 warm_count，acquire 绝不发给任务），因为
+  旧 VMM 进程已随旧 worker 消失，无法验证它还是当初那个
+  干净快照 VM。它们只能走 reconcile_orphans() 清理。
+- reconcile_orphans()：worker 启动/清理时的显式入口，对
+  每个 orphan/assigned 条目尽力调 executor.shutdown_vm
+  （清 socket 与残留登记），条目清掉并落盘，返回已清理的
+  VM 名列表。registry 里没有的残留 socket 仍由执行器的
+  cleanup_stale_vms() 扫 state_dir 处理，两者互补，不互相
+  替代；reconcile 绝不在构造时自动执行（同一 state_dir 只
+  能有一个活池，构造时自动清理会误杀另一个活池的 VM）。
+- 损坏 fail-safe：registry 不是合法 JSON / 顶层不是 dict 时，
+  整个文件 quarantine 成 <path>.corrupt（尽力而为），池子
+  从空开始，绝不用损坏数据复用 VM；单条目结构非法时只
+  丢该条（记 registry_corrupt_entries），不连坐其它条目。
+  registry 写失败时只记日志警告、不炸任务流（池子内存态
+  仍可用，重启后孤儿改由 socket 扫描兜底）。
+"""
+
+import json
+import logging
+import os
+
+_LOGGER = logging.getLogger("cloud.cloud_hypervisor")
+
+_REGISTRY_FILENAME = "warm-pool.json"
+
+
+class WarmPool:
+    """快照预热池（快速启动设计，接口与状态机在代码里）。
+
+    目标：冷启动（preflight + 起 VMM + create + boot + SSH
+    就绪）是慢路径；预热池提前把干净 VM 起好并暂停，任务
+    来时 acquire 直接 resume，省掉冷启动的大头。
+    状态机：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,
+                 registry_path=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 = {}  # 合成名 -> {"state", "vm", "task", ...}
+        self.registry_path = registry_path
+        if self.registry_path is None:
+            state_dir = getattr(executor, "state_dir", None)
+            if state_dir:
+                self.registry_path = os.path.join(
+                    state_dir, _REGISTRY_FILENAME)
+        self.registry_corrupt = False
+        self.registry_corrupt_entries = 0
+        self._load_registry()
+
+    # -- registry ------------------------------------------------------
+    def _load_registry(self):
+        """载入 registry：读到的条目一律标 orphan（见模块
+        docstring，绝不直接当 warm 发给任务）。无文件时为空
+        池；损坏时 quarantine 并从空开始。"""
+        if not self.registry_path:
+            return
+        try:
+            with open(self.registry_path, "r") as fh:
+                data = json.load(fh)
+        except FileNotFoundError:
+            return
+        except (OSError, ValueError):
+            self.registry_corrupt = True
+            self._quarantine_registry()
+            return
+        if not isinstance(data, dict):
+            self.registry_corrupt = True
+            self._quarantine_registry()
+            return
+        for name, entry in data.items():
+            if not isinstance(entry, dict) \
+                    or not entry.get("vm") \
+                    or entry.get("state") not in ("warm", "assigned"):
+                self.registry_corrupt_entries += 1
+                continue
+            self._entries[name] = {
+                "state": "orphan",
+                "vm": entry["vm"],
+                "task": {"id": entry.get("task_id") or name},
+                "snapshot_url": entry.get("snapshot_url"),
+            }
+
+    def _quarantine_registry(self):
+        """把损坏的 registry 文件挪到 <path>.corrupt 旁（尽力
+        而为），原路径不再留损坏数据给下一次载入。"""
+        if not self.registry_path:
+            return
+        try:
+            os.replace(self.registry_path,
+                       self.registry_path + ".corrupt")
+        except OSError:
+            _LOGGER.warning("chv_pool_registry_quarantine_failed "
+                            "path=%s", self.registry_path)
+
+    def _save_registry(self):
+        """落盘当前条目（只写 warm/assigned 态；orphan 是
+        待清理的上一世条目，reconcile 后即删，不写回）。
+        写失败只记日志，不炸任务流。"""
+        if not self.registry_path:
+            return
+        data = {}
+        for name, entry in self._entries.items():
+            if entry["state"] not in ("warm", "assigned"):
+                continue
+            task = entry.get("task") or {}
+            task_id = task.get("id") if isinstance(task, dict) \
+                else None
+            data[name] = {
+                "state": entry["state"],
+                "vm": entry.get("vm"),
+                "task_id": task_id,
+                "snapshot_url": entry.get("snapshot_url"),
+            }
+        try:
+            directory = os.path.dirname(self.registry_path)
+            if directory:
+                os.makedirs(directory, exist_ok=True)
+            tmp_path = self.registry_path + ".tmp"
+            with open(tmp_path, "w") as fh:
+                json.dump(data, fh)
+            os.replace(tmp_path, self.registry_path)
+        except OSError:
+            _LOGGER.warning("chv_pool_registry_save_failed path=%s",
+                            self.registry_path)
+
+    def reconcile_orphans(self):
+        """清理上一世残留条目：对每个 orphan/assigned 条目
+        尽力 shutdown_vm（清 socket 与残留登记），条目清掉
+        并落盘，返回已清理的 VM 名列表。worker 启动/清理时
+        显式调它（见模块 docstring 的单活池假设）。"""
+        cleaned = []
+        for name, entry in list(self._entries.items()):
+            if entry["state"] not in ("orphan", "assigned"):
+                continue
+            vm = entry.get("vm") or name
+            try:
+                self.executor.shutdown_vm(vm)
+            except Exception:
+                pass
+            self._entries.pop(name, None)
+            cleaned.append(vm)
+        if cleaned:
+            self._save_registry()
+            try:
+                self.executor._emit(
+                    "warm_pool_reconcile", {"id": None},
+                    "cleaned=%s" % ",".join(cleaned))
+            except Exception:
+                pass
+        return cleaned
+
+    # -- 池操作 ----------------------------------------------------------
+    def warm_count(self):
+        return sum(1 for e in self._entries.values()
+                   if e["state"] == "warm")
+
+    def orphan_count(self):
+        """上一世残留条目数（待 reconcile，不可分配）。"""
+        return sum(1 for e in self._entries.values()
+                   if e["state"] in ("orphan", "assigned"))
+
+    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
+            vm = self.executor.vm_name(task)
+            self._entries[name] = {
+                "state": "warm", "task": task, "vm": vm,
+                "snapshot_url": self.snapshot_url}
+            added.append(name)
+        if added:
+            self._save_registry()
+        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._save_registry()
+                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._save_registry()
+        self.executor._emit(
+            "warm_pool_release", task,
+            "vm=%s destroyed=%s" % (pool_vm, destroyed))
+        return destroyed

cloud/chv/ssh_transport.py

@@ -0,0 +1,93 @@
+"""Cloud Hypervisor 客机 SSH 传输（叶子模块，不依赖
+cloud.sandbox，避免循环导入）。
+
+安全模型（本轨的核心变化，替代 #216 的
+StrictHostKeyChecking=no + UserKnownHostsFile=/dev/null）：
+
+- 每 VM 一个独立 known_hosts 文件：
+  state_dir/<vm 名>/known_hosts（与每任务磁盘同目录，
+  任务销毁时不自动删——known_hosts 留在宿主 state_dir 里，
+  供事后审计该 VM 当时钉死的 key；VM 名本身已含任务 id，
+  不会串到别的 VM 上）。
+- 首次信任（TOFU，边界必须写清，不假装是强验证）：
+  首次启动时客机 host key 未知，执行器用 ssh-keyscan 经
+  宿主转发端口（127.0.0.1:<port>，只走本机回环）抓一次
+  当时呈现的 key，写入该 VM 的 known_hosts 文件；此后
+  同一 VM 的每次 SSH 都用 StrictHostKeyChecking=yes +
+  该文件校验，VM 被替换 / 中间人换 key 时 SSH 直接拒绝，
+  不再静默接受。首次捕获本身不验证“这台客机就是我们要
+  的那台”——它只防首次之后的替换；更强的方案是镜像构建
+  /cloud-init 时把客机 host key 预置进 known_hosts（未来
+  工作，见 docs/chv-production-readiness.md 的待办），本轨
+  在 mock 测试里只能验证“捕获→写文件→后续命令带钉死
+  选项”的管线，绝不声称 mock 验证了首次密钥的真实性。
+- 密钥红线不变：argv 里只有 key 文件路径，绝无 key 值；
+  known_hosts 文件只含 host key 公钥行，不含任何私钥。
+"""
+
+import os
+
+
+def known_hosts_path(state_dir, vm_name):
+    """该 VM 的 known_hosts 文件路径：state_dir/<vm>/known_hosts。"""
+    return os.path.join(state_dir, vm_name, "known_hosts")
+
+
+def build_ssh_command(ssh_port, ssh_key, ssh_user, egress_proxy,
+                      known_hosts_file, inner_cmd):
+    """构造 SSH 命令：ssh -p <端口> -i <key 文件> + 钉死选项
+    + user@127.0.0.1 + 客内命令。
+
+    与 #216 的区别只有 host key 选项：StrictHostKeyChecking=yes
+    + UserKnownHostsFile=<每 VM 文件>，不再是 no + /dev/null。
+    BatchMode=yes 不变（绝不交互问 host key / 口令）。
+    egress_proxy 已设时客内命令前加 env 前缀（代理地址本身
+    不是密钥；客机出口只走该白名单代理）。"""
+    inner = list(inner_cmd)
+    if egress_proxy:
+        inner = ["env",
+                 "HTTPS_PROXY=%s" % egress_proxy,
+                 "HTTP_PROXY=%s" % egress_proxy,
+                 "NO_PROXY=localhost,127.0.0.1"] + inner
+    return ["ssh", "-p", str(ssh_port),
+            "-i", ssh_key,
+            "-o", "StrictHostKeyChecking=yes",
+            "-o", "UserKnownHostsFile=%s" % known_hosts_file,
+            "-o", "BatchMode=yes",
+            "%s@127.0.0.1" % ssh_user] + inner
+
+
+def parse_keyscan_output(text):
+    """从 ssh-keyscan 的 stdout 里挑出本机回环端口的 host key
+    行：形如 "[127.0.0.1]:<port> ssh-ed25519 AAAA..."（-p 时
+    的标准写法）或 "127.0.0.1 ssh-rsa AAAA..."。其它主机行、
+    注释行、空行一律丢掉，绝不把陌生主机的 key 写进文件。"""
+    kept = []
+    for line in (text or "").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        parts = line.split()
+        if len(parts) < 3:
+            continue
+        host, keytype = parts[0], parts[1]
+        if host != "127.0.0.1" \
+                and not host.startswith("[127.0.0.1]:"):
+            continue
+        if not keytype.startswith("ssh-"):
+            continue
+        kept.append(line)
+    return kept
+
+
+def write_known_hosts(path, key_lines):
+    """原子写 known_hosts：先写同目录临时文件再 rename，
+    文件权限 0600（只含公钥行，仍按敏感文件对待）。"""
+    directory = os.path.dirname(path)
+    if directory:
+        os.makedirs(directory, exist_ok=True)
+    tmp_path = path + ".tmp"
+    with open(tmp_path, "w") as fh:
+        fh.write("\n".join(key_lines) + "\n")
+    os.chmod(tmp_path, 0o600)
+    os.replace(tmp_path, path)

cloud/executor_base.py

@@ -0,0 +1,30 @@
+"""执行器基类（叶子模块，不依赖 cloud.sandbox 与 cloud.chv，
+避免循环导入）。
+
+原来在 cloud/sandbox.py 里；拆分后 Local/Docker/Cloud
+执行器（sandbox.py 内）与 CloudHypervisorExecutor
+（cloud/chv/executor.py）共用这一个类对象，isinstance
+判断与旧路径 `from cloud.sandbox import Executor` 都不变
+（sandbox.py 会 re-export 它）。
+"""
+
+
+class Executor:
+    """执行器基类。
+
+    task 是 dict，含 id、project_id、repo、prompt、branch（可选）。
+    execute(task) -> dict，至少含 executor、task_id、status、stdout。
+    """
+
+    def execute(self, task) -> dict:
+        raise NotImplementedError("子类必须实现 execute(task)")
+
+    def goal_task(self, task) -> dict:
+        """目标循环（host.py goal）：turn→验测→喂回重试。
+
+        子类实现；默认不支持（CloudExecutor 的通用 HTTP 协议里
+        没有 goal/验测概念，调它会如实失败，不静默降级成普通
+        dispatch）。结果 dict 除 execute() 的字段外另含：
+        goal_met（True/False/None=无结论）、goal_error。
+        """
+        raise NotImplementedError("该执行器不支持 goal 模式")

cloud/tests/test_chv_optimize.py

@@ -0,0 +1,586 @@
+"""Cloud Hypervisor 优化轨（feat-chv-optimize，堆叠于 #216 之上）测试。
+
+只用标准库与 mock runner/launcher/api_client，绝不启动真实
+VM（与 test_cloud_hypervisor.py 同款边界）。覆盖：
+- 拆模块：cloud/chv/* 新路径与 cloud/sandbox 旧路径是同一对象
+- 去重复：Local 与 CHV 的 goal 解析在矩阵输入下结论全等；
+  三执行器 host 顶层参数顺序全等；runner 超时 helper
+- 安全：每 VM known_hosts 钉死、ssh-keyscan 首次捕获（TOFU）、
+  非 root 用户与端口显式配置与收紧校验
+- execute 语义：只读已运行 VM，未运行报 vm_not_running
+- 预热池：registry 持久化、重启载入只当孤儿、reconcile 清理、
+  损坏 quarantine、一 次性不复用
+- 长任务：VM 中途死亡判 interrupted 而非 failed/抛异常
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv import WarmPool
+    from cloud.chv.api_client import _unix_socket_api as _chv_api
+    from cloud.chv.config import _parse_memory_bytes as _chv_mem
+    from cloud.sandbox import ChvFailure as _sb_failure
+    from cloud.sandbox import CloudHypervisorExecutor as _sb_exec
+    from cloud.sandbox import DockerExecutor
+    from cloud.sandbox import LocalExecutor
+    from cloud.sandbox import WarmPool as _sb_pool
+    from cloud.sandbox import _call_runner_with_timeout
+    from cloud.sandbox import _parse_memory_bytes as _sb_mem
+    from cloud.sandbox import _unix_socket_api as _sb_api
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv import WarmPool
+    from cloud.chv.api_client import _unix_socket_api as _chv_api
+    from cloud.chv.config import _parse_memory_bytes as _chv_mem
+    from cloud.sandbox import ChvFailure as _sb_failure
+    from cloud.sandbox import CloudHypervisorExecutor as _sb_exec
+    from cloud.sandbox import DockerExecutor
+    from cloud.sandbox import LocalExecutor
+    from cloud.sandbox import WarmPool as _sb_pool
+    from cloud.sandbox import _call_runner_with_timeout
+    from cloud.sandbox import _parse_memory_bytes as _sb_mem
+    from cloud.sandbox import _unix_socket_api as _sb_api
+
+
+class FakeResult:
+    def __init__(self, returncode=0, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class FakeProc:
+    def __init__(self, alive=True):
+        self._alive = alive
+        self.terminated = 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._alive = False
+
+
+def _write(path, content=b"x"):
+    with open(path, "wb") as fh:
+        fh.write(content)
+    os.chmod(path, 0o755)
+    return path
+
+
+class ChvOptCase(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 = []
+        self.api_calls = []
+        self.launches = []
+
+    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)
+        return CloudHypervisorExecutor(**kwargs)
+
+    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:
+                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"}
+
+
+# -- 拆模块 ---------------------------------------------------------------
+class SplitTest(unittest.TestCase):
+    def test_new_and_old_import_paths_are_same_objects(self):
+        self.assertIs(CloudHypervisorExecutor, _sb_exec)
+        self.assertIs(WarmPool, _sb_pool)
+        self.assertIs(ChvFailure, _sb_failure)
+        self.assertIs(_chv_mem, _sb_mem)
+        self.assertIs(_chv_api, _sb_api)
+
+    def test_classes_live_in_chv_modules(self):
+        self.assertEqual(CloudHypervisorExecutor.__module__,
+                         "cloud.chv.executor")
+        self.assertEqual(WarmPool.__module__, "cloud.chv.pool")
+        self.assertEqual(ChvFailure.__module__, "cloud.chv.errors")
+
+
+# -- 去重复：goal 解析与 host 命令 ------------------------------------------
+class SharedGoalTest(unittest.TestCase):
+    CASES = [
+        # (returncode, stdout, stderr)
+        (0, "goal met after rounds\n", ""),
+        (0, "round 1 verify_out=bad\ngoal unmet after rounds\n", ""),
+        (1, "goal interrupted after rounds: boom\n", ""),
+        (1, "KeyError: 'x'\n", ""),
+        (1, "some odd failure\n", ""),
+        (0, "nothing conclusive\n", ""),
+        (1, "goal-resume 拒绝：断点丢失\n", ""),
+        (1, "任务已存在\n", ""),
+        (1, "goal met after rounds\n", ""),
+        (1, "goal unmet after rounds\n", ""),
+        (1, "", "HTTP 503 Service Unavailable\n"),
+    ]
+
+    def test_local_and_chv_goal_conclusions_match(self):
+        local = LocalExecutor("/tmp/host.py")
+        chv = ChvOptCase()
+        chv.setUp()
+        try:
+            ex = chv.make_exec()
+            for rc, out, err in self.CASES:
+                res = FakeResult(rc, out, err)
+                a = local._goal_result_from_runner("t1", res)
+                b = ex._goal_result("t1", res)
+                self.assertEqual(
+                    (a["status"], a["goal_met"]),
+                    (b["status"], b["goal_met"]),
+                    "rc=%s stdout=%r" % (rc, out))
+                self.assertEqual(a["executor"], "local")
+                self.assertEqual(b["executor"], "cloud-hypervisor")
+        finally:
+            chv.tearDown()
+
+    def test_already_exists_message_points_to_continue(self):
+        chv = ChvOptCase()
+        chv.setUp()
+        try:
+            ex = chv.make_exec()
+            out = ex._goal_result(
+                "t1", FakeResult(1, "任务已存在\n", ""))
+            self.assertEqual(out["status"], "failed")
+            self.assertIn("继续", out["goal_error"])
+        finally:
+            chv.tearDown()
+
+    def test_host_command_options_identical_across_executors(self):
+        chv = ChvOptCase()
+        chv.setUp()
+        try:
+            ex = chv.make_exec()
+            local = LocalExecutor("/tmp/host.py")
+            docker = DockerExecutor()
+            for kwargs in (
+                    {"model": "m1", "effort": "low", "approval": "ask"},
+                    {"model": "m1", "effort": "low", "approval": "full",
+                     "sandbox": "read-only",
+                     "model_provider": "sub2api"},
+            ):
+                lcmd = local._base_command(**kwargs)
+                dcmd = docker._inner_host_command("status", **kwargs)
+                ccmd = ex._guest_host_command("status", **kwargs)
+                self.assertEqual(lcmd[2:], dcmd[2:-1])
+                self.assertEqual(lcmd[2:], ccmd[2:-1])
+        finally:
+            chv.tearDown()
+
+    def test_runner_timeout_helper(self):
+        seen = []
+
+        def only_cmd(cmd):
+            seen.append(("cmd-only", list(cmd)))
+            return FakeResult(0, "", "")
+
+        out = _call_runner_with_timeout(only_cmd, ["x"], timeout=5)
+        self.assertEqual(out.returncode, 0)
+        self.assertEqual(seen, [("cmd-only", ["x"])])
+
+        def with_timeout(cmd, timeout=None):
+            seen.append(("with-timeout", timeout))
+            return FakeResult(0, "", "")
+
+        _call_runner_with_timeout(with_timeout, ["x"], timeout=7)
+        self.assertEqual(seen[-1], ("with-timeout", 7))
+        _call_runner_with_timeout(with_timeout, ["x"], timeout=None)
+        self.assertEqual(seen[-1], ("with-timeout", None))
+
+
+# -- 安全：known_hosts 钉死 -------------------------------------------------
+class SshPinningTest(ChvOptCase):
+    def test_known_hosts_path_is_per_vm(self):
+        ex = self.make_exec()
+        a = ex.known_hosts_path({"id": "t-a"})
+        b = ex.known_hosts_path({"id": "t-b"})
+        self.assertNotEqual(a, b)
+        self.assertEqual(a, os.path.join(self.state_dir, "chv-t-a",
+                                         "known_hosts"))
+
+    def test_capture_host_key_writes_pinned_file_0600(self):
+        def keyscan(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            return FakeResult(
+                0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n"
+                "evil.example.com ssh-rsa AAAAEvil\n"
+                "# comment\n", "")
+        ex = self.make_exec(runner=keyscan)
+        self.assertTrue(ex.capture_host_key(self._task()))
+        path = ex.known_hosts_path(self._task())
+        with open(path) as fh:
+            content = fh.read()
+        self.assertIn("[127.0.0.1]:2222 ssh-ed25519 AAAAB3test",
+                      content)
+        self.assertNotIn("evil.example.com", content)
+        self.assertEqual(os.stat(path).st_mode & 0o777, 0o600)
+
+    def test_capture_host_key_empty_is_false_not_error(self):
+        ex = self.make_exec(
+            runner=lambda cmd, timeout=None: FakeResult(0, "", ""))
+        self.assertFalse(ex.capture_host_key(self._task()))
+        self.assertFalse(os.path.exists(
+            ex.known_hosts_path(self._task())))
+
+    def test_wait_guest_ready_captures_before_ssh(self):
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh-keyscan":
+                return FakeResult(
+                    0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n",
+                    "")
+            if cmd[0] == "ssh" and cmd[-1] == "true":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+        ex = self.make_exec(runner=runner)
+        self.assertTrue(ex.wait_guest_ready(self._task()))
+        self.assertTrue(os.path.exists(
+            ex.known_hosts_path(self._task())))
+        keyscan_idx = next(i for i, c in enumerate(self.calls)
+                           if c[0] == "ssh-keyscan")
+        ssh_idx = next(i for i, c in enumerate(self.calls)
+                       if c[0] == "ssh" and c[-1] == "true")
+        self.assertLess(keyscan_idx, ssh_idx)
+
+    def test_ssh_user_port_explicit_config(self):
+        ex = self.make_exec(ssh_user="codex", ssh_port=2201)
+        cmd = ex.build_ssh_command(self._task(), ["true"])
+        joined = " ".join(cmd)
+        self.assertIn("codex@127.0.0.1", joined)
+        self.assertNotIn("root@127.0.0.1", joined)
+        self.assertIn("2201", joined)
+
+    def test_ssh_user_port_tightened_validation(self):
+        with self.assertRaises(ValueError):
+            self.make_exec(ssh_user="")
+        with self.assertRaises(ValueError):
+            self.make_exec(ssh_port=0)
+        with self.assertRaises(ValueError):
+            self.make_exec(ssh_port=70000)
+        with self.assertRaises(ValueError):
+            self.make_exec(ssh_port="abc")
+        with self.assertRaises(ValueError):
+            self.make_exec(env_extra={"CHV_SSH_PORT": "0"})
+
+
+# -- execute 语义 -------------------------------------------------------------
+class ExecuteGateTest(ChvOptCase):
+    def test_execute_without_running_vm_is_classified(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.execute(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_not_running")
+        self.assertEqual(self.launches, [])
+        self.assertFalse(any(c[0] == "ssh" for c in self.calls))
+
+    def test_execute_with_dead_proc_and_no_socket_is_classified(self):
+        ex = self.make_exec()
+        ex._vms["chv-abc123"] = FakeProc(alive=False)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.execute(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_not_running")
+
+    def test_execute_with_socket_but_no_registry_proceeds(self):
+        ex = self.make_exec()
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir,
+                               "chv-abc123.sock"), "w") as fh:
+            fh.write("")
+        out = ex.execute(self._task())
+        self.assertIn("ok-status", out["stdout"])
+        self.assertEqual(self.launches, [])
+
+
+# -- 预热池持久化与孤儿清理 ----------------------------------------------------
+class StubExecutor:
+    """WarmPool 测试替身：只实现池子调用的鸭子接口。"""
+
+    def __init__(self, state_dir):
+        self.state_dir = state_dir
+        self.warm_pool_size = 2
+        self.snapshot_url = "file:///snap/base"
+        self.wake_calls = []
+        self.pause_calls = []
+        self.shutdown_calls = []
+        self.audit = []
+
+    def vm_name(self, task):
+        raw = task.get("id") if isinstance(task, dict) else str(task)
+        return raw if raw.startswith("chv-") else "chv-" + raw
+
+    def wake_vm(self, task):
+        self.wake_calls.append(task["id"])
+        return {"vm": self.vm_name(task), "state": "running"}
+
+    def pause_vm(self, task):
+        self.pause_calls.append(task["id"])
+        return True
+
+    def shutdown_vm(self, task_or_name):
+        name = task_or_name if isinstance(task_or_name, str) \
+            else self.vm_name(task_or_name)
+        self.shutdown_calls.append(name)
+        return True
+
+    def _emit(self, event, task, detail="", kind=None):
+        self.audit.append(event)
+
+
+class PoolRegistryTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state_dir = os.path.join(self.tmp.name, "state")
+        os.makedirs(self.state_dir, exist_ok=True)
+        self.registry = os.path.join(self.state_dir, "warm-pool.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_persist_load_orphan_reconcile(self):
+        stub = StubExecutor(self.state_dir)
+        pool = WarmPool(stub, size=2, snapshot_url="file:///snap/base")
+        added = pool.prewarm()
+        self.assertEqual(len(added), 2)
+        self.assertTrue(os.path.exists(self.registry))
+        with open(self.registry) as fh:
+            data = json.load(fh)
+        self.assertEqual(len(data), 2)
+        # 新池实例载入：上一世条目只当孤儿，绝不发给任务：
+        stub2 = StubExecutor(self.state_dir)
+        pool2 = WarmPool(stub2, size=2,
+                         snapshot_url="file:///snap/base")
+        self.assertEqual(pool2.warm_count(), 0)
+        self.assertEqual(pool2.orphan_count(), 2)
+        self.assertEqual(pool2.acquire({"id": "t1"})["mode"], "cold")
+        cleaned = pool2.reconcile_orphans()
+        self.assertEqual(len(cleaned), 2)
+        self.assertEqual(sorted(stub2.shutdown_calls), sorted(cleaned))
+        self.assertEqual(pool2.orphan_count(), 0)
+        with open(self.registry) as fh:
+            self.assertEqual(json.load(fh), {})
+        # 第三实例：无残留：
+        pool3 = WarmPool(StubExecutor(self.state_dir), size=2,
+                         snapshot_url="file:///snap/base")
+        self.assertEqual(pool3.orphan_count(), 0)
+
+    def test_corrupt_registry_quarantines_and_starts_empty(self):
+        with open(self.registry, "w") as fh:
+            fh.write("{not json")
+        pool = WarmPool(StubExecutor(self.state_dir), size=2,
+                        snapshot_url="file:///snap/base")
+        self.assertTrue(pool.registry_corrupt)
+        self.assertEqual(pool.warm_count(), 0)
+        self.assertEqual(pool.orphan_count(), 0)
+        self.assertTrue(os.path.exists(self.registry + ".corrupt"))
+        self.assertFalse(os.path.exists(self.registry))
+        # 损坏后仍可正常预热（registry 重建）：
+        self.assertEqual(len(pool.prewarm()), 2)
+        self.assertTrue(os.path.exists(self.registry))
+
+    def test_assigned_entry_on_load_is_never_served(self):
+        with open(self.registry, "w") as fh:
+            json.dump({"warm-1": {"state": "assigned",
+                                  "vm": "chv-warm-1",
+                                  "task_id": "old-task",
+                                  "snapshot_url": "file:///snap/base"}},
+                      fh)
+        pool = WarmPool(StubExecutor(self.state_dir), size=2,
+                        snapshot_url="file:///snap/base")
+        self.assertEqual(pool.warm_count(), 0)
+        self.assertEqual(pool.acquire({"id": "t1"})["mode"], "cold")
+        self.assertEqual(pool.reconcile_orphans(), ["chv-warm-1"])
+
+    def test_release_never_rewarms_and_registry_clears(self):
+        stub = StubExecutor(self.state_dir)
+        pool = WarmPool(stub, size=1, snapshot_url="file:///snap/base")
+        added = pool.prewarm()
+        got = pool.acquire({"id": "t1"})
+        self.assertEqual(got["mode"], "warm")
+        self.assertTrue(pool.release({"id": "t1"}, pool_vm=added[0]))
+        self.assertEqual(pool.warm_count(), 0)
+        with open(self.registry) as fh:
+            self.assertEqual(json.load(fh), {})
+
+
+# -- 长任务中途死亡 --------------------------------------------------------------
+class MidDeathTest(ChvOptCase):
+    def _dispatch_with(self, runner):
+        ex = self.make_exec(runner=runner)
+        return ex.dispatch_task(self._task())
+
+    def test_failed_result_with_dead_vm_is_interrupted(self):
+        proc_holder = []
+
+        def launcher(cmd):
+            proc = FakeProc(alive=True)
+            proc_holder.append(proc)
+            self.launches.append(list(cmd))
+            return proc
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                proc_holder[0]._alive = False
+                return FakeResult(1, "",
+                                  "Connection closed by remote host")
+            if cmd[0] == "ssh":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner, launcher=launcher)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "interrupted")
+        self.assertTrue(out["vm_dead"])
+
+    def test_failed_result_with_live_vm_stays_failed(self):
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                return FakeResult(1, "", "host dispatch failed")
+            if cmd[0] == "ssh":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "failed")
+        self.assertNotIn("vm_dead", out)
+
+    def test_exception_with_dead_vm_is_interrupted_not_raised(self):
+        proc_holder = []
+
+        def launcher(cmd):
+            proc = FakeProc(alive=True)
+            proc_holder.append(proc)
+            self.launches.append(list(cmd))
+            return proc
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                proc_holder[0]._alive = False
+                raise RuntimeError("ssh transport broken")
+            if cmd[0] == "ssh":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner, launcher=launcher)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "interrupted")
+        self.assertTrue(out["vm_dead"])
+
+    def test_exception_with_live_vm_propagates(self):
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                raise RuntimeError("ssh transport broken")
+            if cmd[0] == "ssh":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner)
+        with self.assertRaises(RuntimeError):
+            ex.dispatch_task(self._task())
+
+    def test_goal_exception_with_dead_vm_is_interrupted(self):
+        proc_holder = []
+
+        def launcher(cmd):
+            proc = FakeProc(alive=True)
+            proc_holder.append(proc)
+            self.launches.append(list(cmd))
+            return proc
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "goal" in cmd:
+                proc_holder[0]._alive = False
+                raise RuntimeError("ssh transport broken")
+            if cmd[0] == "ssh":
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner, launcher=launcher)
+        out = ex.goal_task({"id": "g1", "prompt": "做目标",
+                            "goal_mode": True})
+        self.assertEqual(out["status"], "interrupted")
+        self.assertIsNone(out["goal_met"])
+        self.assertTrue(out["vm_dead"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_cloud_hypervisor.py

@@ -279,6 +279,14 @@ class ConfigCommandTest(ChvCase):
         self.assertNotIn("SECRET", joined)
         # egress 代理地址在 env 前缀里（不是密钥），且只在代理已设时出现：
         self.assertIn("HTTPS_PROXY=http://10.0.0.1:8888", joined)
+        # 本轨安全默认变化（#216 的 no + /dev/null 已废除）：
+        # host key 钉死在每 VM 的 known_hosts 文件上：
+        self.assertIn("StrictHostKeyChecking=yes", joined)
+        self.assertNotIn("StrictHostKeyChecking=no", joined)
+        self.assertNotIn("/dev/null", joined)
+        self.assertIn(os.path.join(self.state_dir,
+                                   "chv-abc123", "known_hosts"),
+                      joined)
 
     def test_isolation_names_and_disks_differ(self):
         ex = self.make_exec()
@@ -534,6 +542,10 @@ class DispatchTest(ChvCase):
 
     def test_execute_is_ssh_status_only(self):
         ex = self.make_exec()
+        # 本轨 execute 语义门禁：只读已运行 VM。先登记一个活
+        # 的 VMM 进程表示 VM 已醒（未登记时 execute 报
+        # vm_not_running，见 test_chv_optimize.py）：
+        ex._vms["chv-abc123"] = FakeProc(alive=True)
         out = ex.execute(self._task())
         self.assertEqual(out["executor"], "cloud-hypervisor")
         self.assertIn("ok-status", out["stdout"])

docs/chv-production-readiness.md

@@ -42,6 +42,17 @@
 - ✅🧪 密钥不进镜像/代码/prompt：SSH argv 只有 key 文件路径；
   模型 key 由客机 cloud-init 预置的环境变量名注入；测试断言
   命令中不含哨兵密钥值。
+- ✅🧪 SSH host key 按 VM 钉死（优化轨，#216 的 no + /dev/null
+  已废除）：每 VM 在 state_dir/<vm>/known_hosts 有独立文件，
+  首次启动经 ssh-keyscan 在本机回环端口捕获 key 写入该文件，
+  此后 SSH 一律 StrictHostKeyChecking=yes + 该文件校验；
+  非 root 客机用户（CHV_SSH_USER，显式空串直接拒绝）与端口
+  （CHV_SSH_PORT，1-65535）为显式配置。mock 只验证捕获→
+  写文件→钉死选项的管线，不验证首次密钥真实性。
+- 🔒 首次信任边界未闭合：首次捕获是 TOFU，只防首次之后的
+  替换，不验证首次本身；更强的方案是镜像构建/cloud-init 时
+  把客机 host key 预置进 known_hosts，仍是待办，安全审计时
+  须单独立项确认。
 - 🔒 需安全审计：VM 逃逸面（virtio 设备、tap 桥接、SSH 配置）、
   客机 root 提权面、租户间东西向隔离的宿主网络配置，须由
   安全审计逐项确认，代码自查不能代替审计。
@@ -55,6 +66,12 @@
   冷启动，绝不降级 Docker/Local。
 - ✅🧪 预热条目一次性：acquire 消耗、release 一律销毁，绝不跨
   任务复用脏 VM。
+- ✅🧪 预热池 registry 与孤儿清理（优化轨）：池状态落
+  state_dir/warm-pool.json（原子写），worker 重启后载入的
+  条目一律当孤儿、只能 reconcile_orphans() 清理，绝不重新
+  发给任务；registry 损坏时 quarantine 成 .corrupt 并从空池
+  开始；registry 之外的残留 socket 仍由 cleanup_stale_vms()
+  扫描处理。
 - ⏳ 需 KVM 宿主实测：预热池的真实冷启动耗时、resume 耗时、
   池大小与命中率、快照来源一致性，均需真机数据，mock 数字
   不能当性能结论。
@@ -75,6 +92,17 @@
   shutdown + 杀进程 + 删 socket）、recycle_vm（保留
   task_checkpoints/task_attempts，丢失 guest /tmp 与未拷出
   的 worktree 临时物，绝不声称恢复临时文件）。
+- ✅🧪 execute 只读已运行 VM（优化轨）：execute() 先查 VM
+  是否在运行（登记活进程或 api-socket 存在），未运行报
+  ChvFailure（kind=vm_not_running），不为一次 status 起 VM，
+  也不向不存在的端口发 SSH 再拿连接失败冒充任务状态。
+- ✅🧪 长任务中途死亡判 interrupted（优化轨）：客内命令抛
+  传输异常、或非 goal 结果已 failed 且登记 VMM 进程已死时，
+  _run_with_vm 改判 interrupted（断点可续跑）而非笼统
+  failed；VM 还活着时的异常/失败原样处理。中途心跳未做：
+  host.py 单次 SSH 长回合没有客内进度信号可采，另起一路
+  SSH 轮询只能证明 VM 活着、不能证明任务在进展，评估后
+  认为风险大于收益，宁可只做死亡分类。
 - ⏳ 需 KVM 宿主实测：上述全部生命周期在真机上的状态转换、
   快照恢复后 guest 状态一致性、休眠后文件/worktree 保留，
   均未在真机验证。