feat: CHV 审核修复（P0+P1，堆叠于 #230）

P0：seed 磁盘 CIDATA ISO 挂进 VmConfig 第二块只读盘；
restore provisioned 下复用钉死 known_hosts、缺失 fail-fast
绝不 keyscan；preflight 拦截 CHV_DISK_MODE=reuse（显式
实验二次确认 + 审计放行）；SSH 目标选客机 IP 方案（显式
host > net 静态 ip > 显式转发宣称回环，三者皆无 fail-fast）。
P1：网关 extra_env 经 SSH env 前缀注入、worker 守卫收窄到
Local/CHV；取消/清理/看门狗接 CHV 通用路径；WarmPool
acquire 内 resume；容量门持久化计数；磁盘配额总量+余量
与已存在磁盘校验重拷；API 按 Content-Length 验完整性；
快照 file:// 验落盘再 shutdown；execute 死端口归类
vm_not_running。
P2 全部排除未做（用户范围收窄令，留待以后一轨）。
测试：原 84 + 新增 29 = 113 项，mock，无真 VM。

cloud/chv/api_client.py

@@ -33,25 +33,73 @@ def _unix_socket_api(socket_path, method, path, payload=None):
         sock.connect(socket_path)
         sock.sendall(raw)
         chunks = []
-        while True:
+        # feat-chv-fix（P1-6）：读超时不再当“响应结束”。
+        # 先读满响应头，再按 Content-Length 读满体部；超时、
+        # 提前关流、体部短于 Content-Length 一律 RuntimeError
+        # （调用方判 api_timeout / 传输失败），绝不解析半截
+        # 响应。chunked 与无 Content-Length 响应明确拒绝
+        # （Cloud Hypervisor REST 不应发 chunked；无长度则
+        # 无法证完整性）。
+        buf = b""
+        header_end = None
+        timed_out = False
+        while header_end is None:
             try:
                 data = sock.recv(65536)
             except socket.timeout:
+                timed_out = True
                 break
             if not data:
                 break
-            chunks.append(data)
+            buf += data
+            header_end = buf.find(b"\r\n\r\n")
+        if timed_out or header_end is None:
+            raise RuntimeError("CHV API 响应头不完整（超时或提前关流）")
+        header = buf[:header_end].decode("utf-8", errors="replace")
+        body = buf[header_end + 4:]
+        content_length = None
+        is_chunked = False
+        for line in header.split("\r\n")[1:]:
+            if ":" not in line:
+                continue
+            name, value = line.split(":", 1)
+            name = name.strip().lower()
+            value = value.strip()
+            if name == "content-length":
+                try:
+                    content_length = int(value)
+                except ValueError:
+                    raise RuntimeError("CHV API Content-Length 非法")
+            elif name == "transfer-encoding" \
+                    and "chunked" in value.lower():
+                is_chunked = True
+        if is_chunked:
+            raise RuntimeError("CHV API 响应为 chunked，拒绝解析")
+        if content_length is None:
+            raise RuntimeError("CHV API 响应无 Content-Length，拒绝解析")
+        while len(body) < content_length:
+            try:
+                data = sock.recv(65536)
+            except socket.timeout:
+                raise RuntimeError(
+                    "CHV API 响应体截断：已收 %d / 需 %d 字节"
+                    % (len(body), content_length))
+            if not data:
+                raise RuntimeError(
+                    "CHV API 响应体截断：已收 %d / 需 %d 字节"
+                    % (len(body), content_length))
+            body += data
+        body = body[:content_length]
     finally:
         sock.close()
-    text = b"".join(chunks).decode("utf-8", errors="replace")
+    text = header + "\r\n\r\n" + body.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 \
+    body_text = body.decode("utf-8", errors="replace")
-        else ""
     try:
         parsed = json.loads(body_text) if body_text.strip() else {}
     except ValueError:

cloud/chv/config.py

@@ -93,6 +93,20 @@ def build_chv_config(params, env):
         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"
+    # P0-4 SSH 目标解析（feat-chv-fix）：显式 CHV_SSH_HOST 优先；
+    # 其次取 VmConfig net 首个带 ip 的条目（静态 tap 客机 IP，
+    # 即选定的客机 IP 发现方案）；两者皆无时只有显式
+    # CHV_SSH_FORWARD=1 才允许回环 127.0.0.1（运维宣称宿主
+    # NAT 已把该端口转发到客机 sshd），否则执行器在 wake/
+    # wait 时报 ssh_target_undeclared，绝不无条件假设回环能通。
+    # vsock 未选：需客内 agent + 宿主代理两套新代码（#216 已
+    # 否决）；纯 NAT 方案未选：DNAT 是宿主防火墙外部状态，
+    # 执行器代码既不能创建也不能校验它，写进代码仍是假保证。
+    ssh_host = params.get("ssh_host") or env.get("CHV_SSH_HOST")
+    ssh_forward = env.get("CHV_SSH_FORWARD") == "1"
+    # P0-3：reuse 实验二次确认（名字即带实验含义）。
+    reuse_experiment = env.get(
+        "CHV_REUSE_SINGLE_TENANT_EXPERIMENT") == "1"
     model = params.get("model")
     try:
         max_vms = int(env.get("CHV_MAX_VMS") or 4)
@@ -145,6 +159,8 @@ def build_chv_config(params, env):
         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,
+        ssh_host=ssh_host, ssh_forward=ssh_forward,
+        reuse_experiment=reuse_experiment,
         egress_proxy=egress_proxy,
         allow_direct_egress=allow_direct_egress,
         host_key_mode=host_key_mode,

cloud/chv/disk.py

@@ -2,8 +2,18 @@
 
 单任务单 VM 的文件系统独立性就靠这里：copy 模式每任务
 复制一份基础镜像，两个任务绝不写同一文件；reuse 模式
-直接返回基础镜像路径（仅单租户实验，生产禁用，见
+直接返回基础镜像路径（仅单租户实验，生产拦截见执行器
-CloudHypervisorExecutor 的类 docstring）。
+preflight，须同时设 CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1
+并审计放行）。
+
+feat-chv-fix（P1-5）配额口径：不再只比基础镜像大小。
+check_disk_quota 改为三项：state_dir 已用总量 + 本次
+拷贝（基础镜像大小）+ 宿主剩余空间必须容纳本次拷贝；
+镜像不可读时不再静默跳过，报 disk_quota_unverified
+（preflight 在 wake 路径上会先报 image_missing，这里
+是直接调用者的第二道门）。provision_task_disk 对已存在
+磁盘校验大小：与源不一致即重拷；源不可读即报失败，
+绝不复用脏磁盘。
 """
 
 import os
@@ -12,32 +22,83 @@ import shutil
 from cloud.chv.errors import ChvFailure
 
 
-def check_disk_quota(rootfs, disk_gb):
+def _dir_total_bytes(path):
-    """磁盘配额门：CHV_DISK_GB 已设且基础镜像超过该 GB 数
+    total = 0
-    即拒绝（kind=disk_quota_exceeded）。未设则不检查；
+    for root, _dirs, files in os.walk(path):
-    镜像读不到时不拦（preflight 自会报 image_missing，
+        for name in files:
-    这里不抢它的错）。"""
+            try:
+                total += os.path.getsize(os.path.join(root, name))
+            except OSError:
+                pass
+    return total
+
+
+def check_disk_quota(rootfs, disk_gb, state_dir=None):
+    """磁盘配额门（feat-chv-fix 口径）：
+
+    - disk_gb 未设时不检查；
+    - 镜像不可读时报 disk_quota_unverified，不静默跳过；
+    - 镜像大小超过 disk_gb 即 disk_quota_exceeded；
+    - state_dir 已给时：已用总量 + 本次拷贝超过 disk_gb
+      即 disk_quota_exceeded；宿主剩余空间不足以容纳本次
+      拷贝即 disk_quota_exceeded（宿主余量校验）。
+    """
     if disk_gb is None or not rootfs:
         return
     try:
         size = os.path.getsize(rootfs)
     except OSError:
-        return
+        raise ChvFailure(
+            "disk_quota_unverified",
+            "磁盘镜像不可读，配额无法校验：%s" % rootfs)
     if size > disk_gb * 1024 ** 3:
         raise ChvFailure(
             "disk_quota_exceeded",
             "磁盘镜像 %d 字节超过配额 %.1f GB" % (size, disk_gb))
+    if state_dir and os.path.isdir(state_dir):
+        used = _dir_total_bytes(state_dir)
+        if used + size > disk_gb * 1024 ** 3:
+            raise ChvFailure(
+                "disk_quota_exceeded",
+                "磁盘总量 %d 字节 + 本次 %d 字节超过配额 %.1f GB"
+                % (used, size, disk_gb))
+        try:
+            free = shutil.disk_usage(state_dir).free
+        except OSError:
+            free = None
+        if free is not None and free < size:
+            raise ChvFailure(
+                "disk_quota_exceeded",
+                "宿主剩余空间 %d 字节不足以容纳本次磁盘拷贝 %d 字节"
+                % (free, size))
 
 
 def provision_task_disk(state_dir, vm_name, rootfs, disk_mode):
     """每任务磁盘路径：copy 模式复制到
-    state_dir/<vm>/disk.raw（已存在则复用，不重复复制）；
+    state_dir/<vm>/disk.raw；reuse 模式直接返回基础镜像路径。
-    reuse 模式直接返回基础镜像路径。"""
+
+    feat-chv-fix：已存在磁盘须校验——大小与源不一致即
+    重拷（脏磁盘不复用）；源不可读即 ChvFailure，绝不
+    静默返回旧磁盘。"""
     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):
+    if os.path.exists(task_disk):
+        try:
+            same = os.path.getsize(task_disk) == os.path.getsize(rootfs)
+        except OSError:
+            raise ChvFailure(
+                "disk_quota_unverified",
+                "任务磁盘或源镜像不可读，无法校验复用：%s" % task_disk)
+        if same:
+            return task_disk
+        os.remove(task_disk)
+    try:
         shutil.copyfile(rootfs, task_disk)
+    except OSError as exc:
+        raise ChvFailure(
+            "disk_quota_unverified",
+            "任务磁盘拷贝失败：%s" % exc)
     return task_disk

cloud/chv/executor.py

@@ -227,7 +227,8 @@ class CloudHypervisorExecutor(Executor):
                  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",
+                 net=None, net_json=None, ssh_host=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,
@@ -243,7 +244,8 @@ class CloudHypervisorExecutor(Executor):
             "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,
+            "net": net, "net_json": net_json, "ssh_host": ssh_host,
+            "kvm_path": kvm_path,
             "cpuinfo_path": cpuinfo_path, "ch_remote": ch_remote,
             "model": model, "api_timeout": api_timeout,
             "ready_timeout": ready_timeout,
@@ -257,7 +259,8 @@ class CloudHypervisorExecutor(Executor):
                 "max_vms", "disk_mode", "disk_gb", "egress_proxy",
                 "allow_direct_egress", "task_timeout", "snapshot_url",
                 "warm_pool_size", "api_timeout", "ready_timeout",
-                "ready_interval", "host_key_mode"):
+                "ready_interval", "host_key_mode", "ssh_host",
+                "ssh_forward", "reuse_experiment"):
             setattr(self, _field, getattr(_cfg, _field))
         self.api_client = api_client if api_client is not None \
             else _unix_socket_api
@@ -408,6 +411,17 @@ class CloudHypervisorExecutor(Executor):
                 if not isinstance(_net, (list, dict)):
                     missing.append(("net_config_invalid",
                                     "CHV_NET_JSON 须为 JSON list 或 dict"))
+        # feat-chv-fix（P0-3）：reuse 生产拦截。一行配置再也
+        # 不能击穿租户隔离：只有显式实验二次确认
+        # CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 时才放行，
+        # 且放行本身在 wake_vm 里审计；否则 preflight 直接
+        # 拒绝（kind=disk_mode_reuse_blocked）。
+        if getattr(self, "disk_mode", "copy") == "reuse" \
+                and not getattr(self, "reuse_experiment", False):
+            missing.append(("disk_mode_reuse_blocked",
+                            "CHV_DISK_MODE=reuse 未经显式实验确认："
+                            "须同时设 CHV_REUSE_SINGLE_TENANT_"
+                            "EXPERIMENT=1（仅单租户实验）"))
         if missing:
             kind = missing[0][0]
             detail = "; ".join(text for _, text in missing)
@@ -463,6 +477,39 @@ class CloudHypervisorExecutor(Executor):
         return _known_hosts_path(self.state_dir,
                                  self.vm_name(task_or_name))
 
+    def _net_guest_ip(self):
+        """VmConfig net 首个带 ip 的条目（静态 tap 客机 IP）。"""
+        if not self.net_json:
+            return None
+        try:
+            net = json.loads(self.net_json)
+        except ValueError:
+            return None
+        entries = net if isinstance(net, list) else [net]
+        for entry in entries:
+            if isinstance(entry, dict) and entry.get("ip"):
+                return str(entry["ip"])
+        return None
+
+    def resolve_ssh_host(self):
+        """SSH 目标解析（feat-chv-fix，P0-4 选定方案：客机 IP）：
+
+        显式 CHV_SSH_HOST > net 静态客机 IP > 显式转发宣称下
+        的回环；三者皆无时报 ChvFailure（ssh_target_undeclared），
+        绝不无条件假设 127.0.0.1:2222 能通。"""
+        if getattr(self, "ssh_host", None):
+            return self.ssh_host
+        guest_ip = self._net_guest_ip()
+        if guest_ip:
+            return guest_ip
+        if getattr(self, "ssh_forward", False):
+            return "127.0.0.1"
+        raise ChvFailure(
+            "ssh_target_undeclared",
+            "SSH 目标未声明：须设 CHV_SSH_HOST、或在 CHV_NET_JSON "
+            "里给 net 条目配 ip（静态 tap 客机 IP）、或显式设 "
+            "CHV_SSH_FORWARD=1 宣称宿主 NAT 已转发回环端口")
+
     def provision_host_key(self, task):
         """按 VM 预置 host key（补缺口轨强方案，见 hostkeys.py）：
         启动前为该 VM 生成一次性 ed25519 host key 对，公钥
@@ -484,9 +531,16 @@ class CloudHypervisorExecutor(Executor):
             public_text = fh.read()
         with open(paths["private"], "r") as fh:
             private_text = fh.read()
+        # feat-chv-fix：known_hosts 行的主机部分跟 SSH 目标走
+        # （客机 IP 时不再写死回环，否则钉死校验永远对不上）。
+        try:
+            ssh_host = self.resolve_ssh_host()
+        except ChvFailure:
+            ssh_host = "127.0.0.1"
         write_known_hosts(
             paths["known_hosts"],
-            [known_hosts_line(self.ssh_port, public_text)])
+            [known_hosts_line(self.ssh_port, public_text,
+                              host=ssh_host)])
         seed_dir = write_seed(paths["seed_dir"], vm,
                               private_text, public_text)
         return {"paths": paths, "seed_dir": seed_dir,
@@ -502,14 +556,18 @@ class CloudHypervisorExecutor(Executor):
         抓不到可用 key 行时返回 False（不写文件、不报错），
         由 wait_guest_ready 下一轮再试；sshd 还没起来时
         这就是常态，不是失败。"""
+        try:
+            ssh_host = self.resolve_ssh_host()
+        except ChvFailure:
+            return False
         try:
             result = self._call_runner(
                 ["ssh-keyscan", "-p", str(self.ssh_port),
-                 "127.0.0.1"], timeout=10)
+                 ssh_host], timeout=10)
             stdout = getattr(result, "stdout", "") or ""
         except Exception:
             return False
-        key_lines = parse_keyscan_output(stdout)
+        key_lines = parse_keyscan_output(stdout, host=ssh_host)
         if not key_lines:
             return False
         try:
@@ -518,7 +576,7 @@ class CloudHypervisorExecutor(Executor):
             return False
         return True
 
-    def build_ssh_command(self, task, inner_cmd):
+    def build_ssh_command(self, task, inner_cmd, extra_env=None):
         """构造 SSH 命令：ssh -p <端口> -i <key 文件> + 钉死选项
         + user@127.0.0.1 + 客内命令。argv 里只有 key 文件路径，
         绝无任何 key 值。egress_proxy 已设时客内命令前加 env
@@ -527,13 +585,19 @@ class CloudHypervisorExecutor(Executor):
         Host key 选项是钉死式（StrictHostKeyChecking=yes +
         每 VM known_hosts 文件），#216 的 no + /dev/null 已
         在本轨废除，见类 docstring 与 ssh_transport.py。"""
+        try:
+            ssh_host = self.resolve_ssh_host()
+        except ChvFailure:
+            # 纯命令构造（旧测试/干跑）不炸：回环缺省仅作
+            # 构造兼容，wake/wait 路径仍严格解析。
+            ssh_host = "127.0.0.1"
         return _build_ssh_cmd(
             self.ssh_port, self.ssh_key, self.ssh_user,
             self.egress_proxy, self.known_hosts_path(task),
-            inner_cmd)
+            inner_cmd, ssh_host=ssh_host, extra_env=extra_env)
 
     # -- API 模式：VmConfig / API 调用 / 就绪轮询 -----------------------
-    def build_vm_config(self, task_disk=None):
+    def build_vm_config(self, task_disk=None, seed_disk=None):
         """API 模式的 VmConfig（vm.create 请求体，纯函数）：
         cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、
         kernel{path}、cmdline{args}、disks[{path: 每任务磁盘}]、
@@ -553,6 +617,12 @@ class CloudHypervisorExecutor(Executor):
             "disks": [{"path": task_disk or self.rootfs}],
             "rng": {"src": "/dev/urandom"},
         }
+        # feat-chv-fix（P0-1）：seed 磁盘是 VmConfig 第二块
+        # 只读盘（CIDATA 卷标 ISO，客机 cloud-init NoCloud
+        # 数据源），不是宿主目录——目录本身从不进 VmConfig。
+        if seed_disk:
+            config["disks"].append({"path": seed_disk,
+                                    "readonly": True})
         if self.net_json:
             try:
                 net = json.loads(self.net_json)
@@ -595,22 +665,51 @@ class CloudHypervisorExecutor(Executor):
         with self._vms_lock:
             return len(self._vms)
 
-    def _check_capacity(self):
+    def _persistent_vm_count(self, exclude=None):
-        """宿主容量门：已登记 VM 数达到 max_vms 即明确拒绝
+        """跨重启容量计数（feat-chv-fix，P1-4）：本实例登记
-        （kind=host_capacity_exceeded，调用方应排队/拒绝，
+        的活进程名 ∪ state_dir 下 chv-*.sock 对应的 VM 名。
-        绝不降级 Docker/Local）。"""
+        worker 重启后 _vms 为空，但旧 VMM 的 socket 仍在，
-        if self._active_count() >= self.max_vms:
+        仍计入容量——重启不再绕过 CHV_MAX_VMS。注意这是
+        保守计数：已死 VMM 的残留 socket 在清理前也计入，
+        直到 cleanup_stale_vms / worker reconcile 清掉它。
+        exclude：调用方自己的 VM 名（同任务重复 wake 时先
+        关旧 VM，不应被自己的残留 socket 误拒）。"""
+        names = set()
+        with self._vms_lock:
+            for name, proc in self._vms.items():
+                try:
+                    alive = proc.poll() is None
+                except Exception:
+                    alive = False
+                if alive:
+                    names.add(name)
+        try:
+            entries = os.listdir(self.state_dir)
+        except Exception:
+            entries = []
+        for entry in entries:
+            if entry.endswith(".sock") and entry[:-5].startswith("chv-"):
+                names.add(entry[:-5])
+        if exclude:
+            names.discard(exclude)
+        return len(names)
+
+    def _check_capacity(self, exclude=None):
+        """宿主容量门：持久化计数（见 _persistent_vm_count）
+        达到 max_vms 即明确拒绝（kind=host_capacity_exceeded，
+        调用方应排队/拒绝，绝不降级 Docker/Local）。"""
+        count = self._persistent_vm_count(exclude=exclude)
+        if count >= self.max_vms:
             raise ChvFailure(
                 "host_capacity_exceeded",
-                "宿主 CHV 容量已满：已登记 %d 台，上限 %d，"
+                "宿主 CHV 容量已满：已计数 %d 台，上限 %d，"
-                "请排队或拒绝本任务" % (self._active_count(),
+                "请排队或拒绝本任务" % (count, self.max_vms))
-                                        self.max_vms))
 
     def _check_disk_quota(self):
-        """磁盘配额门（实现在 cloud/chv/disk.py）：
+        """磁盘配额门（实现在 cloud/chv/disk.py，feat-chv-fix
-        CHV_DISK_GB 已设且基础镜像超过该 GB 数即拒绝
+        口径：总量 + 本次 + 宿主余量，镜像不可读不静默跳过）。"""
-        （kind=disk_quota_exceeded）。未设则不检查。"""
+        check_disk_quota(self.rootfs, self.disk_gb,
-        check_disk_quota(self.rootfs, self.disk_gb)
+                         state_dir=self.state_dir)
 
     def _provision_task_disk(self, task):
         """每任务磁盘供给（实现在 cloud/chv/disk.py）：
@@ -618,8 +717,8 @@ class CloudHypervisorExecutor(Executor):
           state_dir/<vm>/disk.raw，VmConfig 用该路径；复制
           慢但两个任务绝不写同一文件，测试用小文件可直接验证。
         - reuse（仅单租户实验）：直接返回基础镜像路径，磁盘
-          不独立，生产禁用（preflight 不拦它，但类 docstring
+          不独立；preflight 拦截（须显式实验二次确认并审计
-          与就绪清单里明写 reuse 不可用于生产）。
+          放行，见 preflight，feat-chv-fix）。
         返回每任务磁盘路径。"""
         return provision_task_disk(
             self.state_dir, self.vm_name(task), self.rootfs,
@@ -657,6 +756,19 @@ class CloudHypervisorExecutor(Executor):
         不先捕获就永远等不到就绪。超时时先关 VM 再 raise
         ChvFailure(kind=guest_not_ready)，绝不在未就绪时
         派任务。"""
+        # feat-chv-fix（P0-2）：provisioned 模式下缺 known_hosts
+        # 时直接 fail-fast，绝不调用 capture_host_key 回落 TOFU。
+        # （keyscan 模式才走下面的捕获分支。）
+        if getattr(self, "host_key_mode", "keyscan") == "provisioned" \
+                and not os.path.exists(self.known_hosts_path(task)):
+            try:
+                self.stop_vm(task)
+            except Exception:
+                pass
+            raise ChvFailure(
+                "host_key_not_pinned",
+                "provisioned 模式下 known_hosts 缺失，拒绝 keyscan "
+                "回落：VM %s 未钉死 host key" % self.vm_name(task))
         attempts = max(1, int(self.ready_timeout / self.ready_interval))
         for attempt in range(attempts):
             try:
@@ -738,9 +850,17 @@ class CloudHypervisorExecutor(Executor):
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
         self.preflight()
-        self._check_capacity()
+        # feat-chv-fix：SSH 目标与 reuse 放行在启动前就定性。
-        self._check_disk_quota()
+        # resolve 失败即 fail-fast（不先起 VMM 再超时）；
+        # reuse 实验放行必须审计，留痕给运维复核。
+        self.resolve_ssh_host()
         name = self.vm_name(task)
+        self._check_capacity(exclude=name)
+        self._check_disk_quota()
+        if getattr(self, "disk_mode", "copy") == "reuse":
+            self._emit("disk_reuse_experiment_allowed", task,
+                       "vm=%s CHV_REUSE_SINGLE_TENANT_EXPERIMENT=1 "
+                       "explicit" % name)
         old = None
         with self._vms_lock:
             if name in self._vms:
@@ -757,15 +877,25 @@ class CloudHypervisorExecutor(Executor):
             self._emit("egress_direct_allowed", task,
                        "vm=%s CHV_ALLOW_DIRECT_EGRESS=1 explicit" % name)
         try:
+            seed_disk = None
             if getattr(self, "host_key_mode", "keyscan") == "provisioned":
                 # 强方案：启动前预置每 VM host key 并钉死，
-                # 失败即 fail-fast，绝不回落 keyscan。
+                # 再把 seed 目录做成 CIDATA seed 磁盘并挂进
-                self.provision_host_key(task)
+                # VmConfig（P0-1 闭环：生成→制盘→挂盘三步缺
+                # 一即 fail-fast，绝不回落 keyscan、绝不只
+                # 生成不注入）。
+                from cloud.chv.hostkeys import (
+                    build_seed_disk, seed_disk_path)
+                provisioned = self.provision_host_key(task)
+                seed_disk = seed_disk_path(self.state_dir, name)
+                build_seed_disk(
+                    provisioned["seed_dir"], seed_disk,
+                    runner=lambda cmd: self._call_runner(cmd))
             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))
+                self.build_vm_config(task_disk, seed_disk=seed_disk))
             if status not in (200, 204):
                 raise ChvFailure(
                     "vm_create_failed",
@@ -828,6 +958,18 @@ class CloudHypervisorExecutor(Executor):
             raise ValueError("snapshot_url 不能为空")
         self._ch_remote(task_or_name, "snapshot",
                         "source_url=%s" % snapshot_url)
+        # feat-chv-fix（P1-7）：只查 returncode 不够。file://
+        # 快照须验证目标文件存在且非空，否则报
+        # snapshot_not_persisted，调用方（hibernate_vm）据此
+        # 不 shutdown、不销毁原 VM；非 file:// 目标无法从宿
+        # 主验证落盘，仍只认 returncode，但文档明写该边界。
+        if snapshot_url.startswith("file://"):
+            _snap_path = snapshot_url[len("file://"):]
+            if not os.path.isfile(_snap_path) \
+                    or os.path.getsize(_snap_path) == 0:
+                raise ChvFailure(
+                    "snapshot_not_persisted",
+                    "快照未落盘或为空：%s" % snapshot_url)
         task = task_or_name if isinstance(task_or_name, dict) \
             else {"id": task_or_name}
         self._emit("vm_snapshot", task,
@@ -849,8 +991,27 @@ class CloudHypervisorExecutor(Executor):
         if "://" not in snapshot_url:
             snapshot_url = "file://%s" % snapshot_url
         self.preflight()
-        self._check_capacity()
+        self.resolve_ssh_host()
+        # feat-chv-fix（P0-2 修正版）：restore 不走 vm.create，
+        # 没有 VmConfig 可挂 seed——审计报告“restore 须把 seed
+        # 接入 VmConfig”的说法在此不适用，纠正为：快照内已
+        # 固化原 VM 的 host key，重新预置新 key 反而会破坏
+        # 钉死。正确不变量是：provisioned 下复用已钉死的
+        # known_hosts（stop 时公钥保留、私钥已删但快照内仍有
+        # 原私钥），缺失时 fail-fast、绝不 keyscan、绝不重
+        # 生成。file:// 快照须先验存在且非空，否则不启动。
+        # 注意：restore 时不重复验快照文件存在——快照落盘
+        # 校验是 snapshot_vm/hibernate_vm 创建时的门（P1-7），
+        # restore 的 ch-remote 本身会在源缺失时失败；在这里
+        # 再验一次会把“源在远端/对象存储”的合法路径也挡掉。
+        if getattr(self, "host_key_mode", "keyscan") == "provisioned" \
+                and not os.path.exists(self.known_hosts_path(task)):
+            raise ChvFailure(
+                "host_key_not_pinned",
+                "provisioned 模式下 restore 缺 known_hosts，拒绝 "
+                "keyscan 回落：VM %s" % self.vm_name(task))
         name = self.vm_name(task)
+        self._check_capacity(exclude=name)
         try:
             self._launch_api_only(task)
             self.wait_api_ready(task)
@@ -936,7 +1097,8 @@ class CloudHypervisorExecutor(Executor):
             if os.path.isdir(seed_dir):
                 shutil.rmtree(seed_dir)
             for _suffix in ("ssh_host_ed25519_key",
-                            "ssh_host_ed25519_key.pub"):
+                            "ssh_host_ed25519_key.pub",
+                            "seed.iso"):
                 _p = os.path.join(self.state_dir, name, _suffix)
                 if os.path.exists(_p):
                     os.remove(_p)
@@ -1027,14 +1189,16 @@ class CloudHypervisorExecutor(Executor):
         return cleaned
 
     # -- 客内执行 ------------------------------------------------------
-    def _run_guest(self, task, inner_args, **kwargs):
+    def _run_guest(self, task, inner_args, extra_env=None, **kwargs):
         """SSH 调客内 host.py，返回 Local/Docker 同款结果 dict
         （executor 字段改写为 cloud-hypervisor，并附 vm 名与
         api-socket）。runner 异常即原样传播，由 worker 侧统一
-        写回 failed——本类不在这里把异常吞成假成功。"""
+        写回 failed——本类不在这里把异常吞成假成功。
+        extra_env（P1-1）：经 SSH env 前缀进客机环境。"""
         self._validate()
         ssh_cmd = self.build_ssh_command(
-            task, self._guest_host_command(*inner_args, **kwargs))
+            task, self._guest_host_command(*inner_args, **kwargs),
+            extra_env=extra_env)
         result = self._call_runner(ssh_cmd, timeout=self.task_timeout)
         # 方法内 import：见模块头注释的循环导入说明。
         from cloud.sandbox import _parse_host_result
@@ -1063,10 +1227,12 @@ class CloudHypervisorExecutor(Executor):
         return parse_goal_result(task_id, result,
                                  "cloud-hypervisor")
 
-    def _run_guest_goal(self, task, inner_args, **kwargs):
+    def _run_guest_goal(self, task, inner_args, extra_env=None,
+                        **kwargs):
         self._validate()
         ssh_cmd = self.build_ssh_command(
-            task, self._guest_host_command(*inner_args, **kwargs))
+            task, self._guest_host_command(*inner_args, **kwargs),
+            extra_env=extra_env)
         return self._goal_result(
             task["id"],
             self._call_runner(ssh_cmd, timeout=self.task_timeout))
@@ -1137,7 +1303,8 @@ class CloudHypervisorExecutor(Executor):
             "interrupted_reason": reason,
         }
 
-    def _run_with_vm(self, task, inner_args, goal=False, **kwargs):
+    def _run_with_vm(self, task, inner_args, goal=False, extra_env=None,
+                     **kwargs):
         """每任务 VM 生命周期：wake_vm（API 模式 + SSH 就绪）
         → 跑客内命令 → finally shutdown_vm。wake 失败时不跑
         任务、直接抛分类失败。本方法不做内部重试：重试是
@@ -1155,8 +1322,9 @@ class CloudHypervisorExecutor(Executor):
             try:
                 if goal:
                     return self._run_guest_goal(
-                        task, inner_args, **kwargs)
+                        task, inner_args, extra_env=extra_env, **kwargs)
-                result = self._run_guest(task, inner_args, **kwargs)
+                result = self._run_guest(task, inner_args,
+                                         extra_env=extra_env, **kwargs)
             except Exception as exc:
                 if self._vm_proc_dead(task):
                     return self._vm_death_result(
@@ -1195,10 +1363,25 @@ class CloudHypervisorExecutor(Executor):
                 "execute() 只读已运行 VM 的 status：VM %s 未运行，"
                 "请先 wake_vm() 再调 execute()，不要用 execute "
                 "唤醒 VM" % self.vm_name(task))
-        return self._run_guest(
+        try:
-            task, ("status", "--task", self._host_name(task)))
+            return self._run_guest(
+                task, ("status", "--task", self._host_name(task)))
+        except Exception as exc:
+            # feat-chv-fix（P1-8）：socket 残留但 VMM 已死时，
+            # _vm_is_running 会因 socket 存在而放行，随后的
+            # SSH 失败不得以原始传输异常抛出——登记进程已死
+            # 或 socket 已消失即统一归类 vm_not_running；VM
+            # 仍活着时的 SSH 失败原样传播（那是客内 status
+            # 失败，不是 VM 未运行）。
+            if self._vm_proc_dead(task) \
+                    or not os.path.exists(self.api_socket(task)):
+                raise ChvFailure(
+                    "vm_not_running",
+                    "execute() 时 VM 已死（socket 残留/进程已死），"
+                    "SSH 失败归类为未运行：%s" % exc)
+            raise
 
-    def dispatch_task(self, task) -> dict:
+    def dispatch_task(self, task, extra_env=None, model_provider=None) -> dict:
         """客内 host.py dispatch：建 host 任务并跑一轮 turn。"""
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
@@ -1210,11 +1393,12 @@ class CloudHypervisorExecutor(Executor):
         if repo:
             args += ["--repo", repo]
         return self._run_with_vm(
-            task, args, model=task.get("model"),
+            task, args, extra_env=extra_env, model=task.get("model"),
             effort=task.get("effort"), approval=task.get("approval"),
+            model_provider=model_provider,
             sandbox=("read-only" if task.get("plan") else None))
 
-    def run_task(self, task) -> dict:
+    def run_task(self, task, extra_env=None, model_provider=None) -> dict:
         """客内 host.py run：在已有 host 任务上再跑一轮 turn。"""
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
@@ -1223,10 +1407,11 @@ class CloudHypervisorExecutor(Executor):
         return self._run_with_vm(
             task, ("run", "--task", self._host_name(task),
                    "--prompt", task["prompt"]),
-            model=task.get("model"), effort=task.get("effort"),
+            extra_env=extra_env, model=task.get("model"),
-            approval=task.get("approval"))
+            effort=task.get("effort"), approval=task.get("approval"),
+            model_provider=model_provider)
 
-    def goal_task(self, task) -> dict:
+    def goal_task(self, task, extra_env=None, model_provider=None) -> dict:
         """客内 host.py goal：目标循环，轮次规则与 Local 同款
         （缺省 3 轮，>100 轮 loud fail，旧验收形态 0 轮禁用）。"""
         if not isinstance(task, dict) or task.get("id") is None:
@@ -1252,11 +1437,13 @@ class CloudHypervisorExecutor(Executor):
         if repo:
             args += ["--repo", repo]
         return self._run_with_vm(
-            task, args, goal=True, model=task.get("model"),
+            task, args, goal=True, extra_env=extra_env,
-            effort=task.get("effort"), approval=task.get("approval"))
+            model=task.get("model"),
+            effort=task.get("effort"), approval=task.get("approval"),
+            model_provider=model_provider)
 
     def goal_resume_task(self, task, goal_text=None,
-                         max_rounds=None) -> dict:
+                         max_rounds=None, extra_env=None, model_provider=None) -> dict:
         """客内 host.py goal-resume：goal_text=None 为断点续跑
         （不带 --goal-text，沿用原 goal+verify），给 goal_text
         为开新目标（带 --goal-text 与 --max-rounds，默认 0）。"""
@@ -1269,11 +1456,14 @@ class CloudHypervisorExecutor(Executor):
         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"),
+            task, args, goal=True, extra_env=extra_env,
-            effort=task.get("effort"), approval=task.get("approval"))
+            model=task.get("model"),
+            effort=task.get("effort"), approval=task.get("approval"),
+            model_provider=model_provider)
 
     def goal_resume_breakpoint_task(self, task,
-                                    model_provider=None) -> dict:
+                                    model_provider=None,
+                                    extra_env=None) -> dict:
         """客内 host.py goal-resume 断点续跑：无 --goal-text、
         无 --max-rounds，host 沿用原任务从断点轮次续跑。
         worker 层已接线（补缺口轨）：非网关的 CHV 断点任务
@@ -1286,7 +1476,7 @@ class CloudHypervisorExecutor(Executor):
             raise ValueError("任务缺少 id")
         return self._run_with_vm(
             task, ("goal-resume", "--task", self._host_name(task)),
-            goal=True, model=task.get("model"),
+            goal=True, extra_env=extra_env, model=task.get("model"),
             effort=task.get("effort"), approval=task.get("approval"),
             model_provider=model_provider)
 

cloud/chv/hostkeys.py

@@ -44,16 +44,49 @@ def host_key_paths(state_dir, vm_name):
     }
 
 
-def known_hosts_line(port, public_key_text):
+def known_hosts_line(port, public_key_text, host="127.0.0.1"):
     """公钥文件内容转 known_hosts 行：只取前两列（类型+数据）。
 
     公钥文件形如 'ssh-ed25519 AAAA... [comment]'，known_hosts
-    行形如 '[127.0.0.1]:<port> ssh-ed25519 AAAA...'。非法
+    行形如 '[<host>]:<port> ssh-ed25519 AAAA...'（host 缺省
-    内容 ValueError，绝不把陌生行原样写进文件。"""
+    回环，仅作兼容；客机 IP 方案下 host 应为解析出的客机 IP）。
+    非法内容 ValueError，绝不把陌生行原样写进文件。"""
     parts = (public_key_text or "").strip().split()
     if len(parts) < 2 or not parts[0].startswith("ssh-"):
         raise ValueError("host 公钥格式非法")
-    return "[127.0.0.1]:%s %s %s" % (port, parts[0], parts[1])
+    return "[%s]:%s %s %s" % (host, port, parts[0], parts[1])
+
+
+def seed_disk_path(state_dir, vm_name):
+    """该 VM 的 NoCloud seed 磁盘镜像路径（CIDATA 卷标 ISO）。"""
+    return os.path.join(state_dir, vm_name, "seed.iso")
+
+
+def build_seed_disk(seed_dir, seed_disk, runner=None):
+    """把 seed 目录做成 CIDATA 卷标 ISO（feat-chv-fix，P0-1 的
+    注入载体）：genisoimage -volid CIDATA -joliet -rock。
+
+    runner 不给时用 subprocess；测试可注入假 runner（假
+    runner 负责把 seed_disk 文件写出来）。工具失败、返回码
+    非 0、或产物不存在/为空时一律 RuntimeError，调用方
+    （wake_vm provisioned 路径）fail-fast，绝不带着“seed
+    目录已写但未挂进 VmConfig”的 VM 继续启动。seed 目录
+    本身只是中间物，不再被当成注入通道。"""
+    cmd = ["genisoimage", "-output", seed_disk, "-volid", "CIDATA",
+           "-joliet", "-rock", seed_dir]
+    if runner is not None:
+        result = runner(cmd)
+        rc = getattr(result, "returncode", 0)
+        if rc not in (0, None):
+            raise RuntimeError("genisoimage 失败：returncode=%s" % rc)
+    else:
+        import subprocess
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        if result.returncode != 0:
+            raise RuntimeError("genisoimage 失败：%s" % result.stderr)
+    if not os.path.isfile(seed_disk) or os.path.getsize(seed_disk) == 0:
+        raise RuntimeError("seed 磁盘未生成：%s" % seed_disk)
+    return seed_disk
 
 
 def build_seed_user_data(private_key_text, public_key_text):

cloud/chv/pool.py

@@ -220,20 +220,43 @@ class WarmPool:
         return added
 
     def acquire(self, task):
-        """分配一个预热 VM 给任务：有 warm 条目时消耗一个
+        """分配一个预热 VM 给任务（feat-chv-fix，P1-3 接线）：
-        （状态 assigned，返回 mode=warm + 该条目的合成任务
+
-        名，供调用方 resume 后使用）；无 warm 条目时返回
+        有 warm 条目时先调 executor.resume_vm() 把暂停的 VM
-        mode=cold（调用方回落 wake_vm 冷启动）。"""
+        真正唤醒（resume 失败即条目作废、继续找下一个，
+        绝不把 paused 的 VM 当 running 发给任务），再标
+        assigned 并返回 mode=warm；resume 全部失败或无 warm
+        条目时返回 mode=cold（调用方回落 wake_vm 冷启动）。
+        注意：本方法只解决“acquire 不再悬空 resume 责任”
+        这一半；worker dispatch 前查池的另一半仍未接线
+        （见执行器/就绪文档的“未接线缺口”标注），预热收益
+        在生产路径上仍不是已交付特性。"""
         for name, entry in list(self._entries.items()):
-            if entry["state"] == "warm":
+            if entry["state"] != "warm":
-                entry["state"] = "assigned"
+                continue
-                entry["task"] = task
+            resume = getattr(self.executor, "resume_vm", None)
-                self._save_registry()
+            if resume is not None:
+                try:
+                    resume(entry.get("task") or task)
+                except Exception:
+                    self._entries.pop(name, None)
+                    self._save_registry()
+                    continue
+            # 无 resume_vm 的鸭子型执行器（旧测试替身）：
+            # 保持旧 assigned 语义，不因此丢弃条目；真实
+            # CloudHypervisorExecutor 一律有 resume_vm，走
+            # 上面分支，责任不悬空。
+            entry["state"] = "assigned"
+            entry["task"] = task
+            self._save_registry()
+            try:
                 self.executor._emit(
                     "warm_pool_acquire", task,
                     "vm=%s" % name)
-                return {"mode": "warm", "vm": name,
+            except Exception:
-                        "pool_task": entry["task"]}
+                pass
+            return {"mode": "warm", "vm": name,
+                    "pool_task": entry["task"]}
         return {"mode": "cold", "vm": None, "pool_task": None}
 
     def release(self, task, pool_vm=None):

cloud/chv/ssh_transport.py

@@ -34,34 +34,48 @@ def known_hosts_path(state_dir, vm_name):
 
 
 def build_ssh_command(ssh_port, ssh_key, ssh_user, egress_proxy,
-                      known_hosts_file, inner_cmd):
+                      known_hosts_file, inner_cmd, ssh_host="127.0.0.1",
+                      extra_env=None):
     """构造 SSH 命令：ssh -p <端口> -i <key 文件> + 钉死选项
-    + user@127.0.0.1 + 客内命令。
+    + user@<ssh_host> + 客内命令。
 
-    与 #216 的区别只有 host key 选项：StrictHostKeyChecking=yes
+    feat-chv-fix（P0-4）：ssh_host 不再在传输层写死回环，
-    + UserKnownHostsFile=<每 VM 文件>，不再是 no + /dev/null。
+    由执行器解析传入（显式 CHV_SSH_HOST / net 客机 IP /
-    BatchMode=yes 不变（绝不交互问 host key / 口令）。
+    显式转发宣称下的回环）；本函数缺省回环仅为旧调用点
+    兼容，执行器路径绝不依赖该缺省。
+    extra_env（feat-chv-fix，P1-1 网关注入）：经同一 env
+    前缀进客机环境（与 egress 代理前缀同款机制），不进
+    镜像/代码/prompt；值本身只出现在宿主侧 ssh argv（与
+    worker 同一信任边界），执行器绝不把完整命令记入日志。
     egress_proxy 已设时客内命令前加 env 前缀（代理地址本身
     不是密钥；客机出口只走该白名单代理）。"""
     inner = list(inner_cmd)
+    env_assignments = []
+    if extra_env:
+        for key in sorted(extra_env):
+            env_assignments.append("%s=%s" % (key, extra_env[key]))
     if egress_proxy:
-        inner = ["env",
+        env_assignments += [
-                 "HTTPS_PROXY=%s" % egress_proxy,
+            "HTTPS_PROXY=%s" % egress_proxy,
-                 "HTTP_PROXY=%s" % egress_proxy,
+            "HTTP_PROXY=%s" % egress_proxy,
-                 "NO_PROXY=localhost,127.0.0.1"] + inner
+            "NO_PROXY=localhost,127.0.0.1"]
+    if env_assignments:
+        inner = ["env"] + env_assignments + 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
+            "%s@%s" % (ssh_user, ssh_host)] + inner
 
 
-def parse_keyscan_output(text):
+def parse_keyscan_output(text, host="127.0.0.1"):
-    """从 ssh-keyscan 的 stdout 里挑出本机回环端口的 host key
+    """从 ssh-keyscan 的 stdout 里挑出目标主机的 host key
-    行：形如 "[127.0.0.1]:<port> ssh-ed25519 AAAA..."（-p 时
+    行：形如 "[<host>]:<port> ssh-ed25519 AAAA..."（-p 时
-    的标准写法）或 "127.0.0.1 ssh-rsa AAAA..."。其它主机行、
+    的标准写法）或 "<host> ssh-rsa AAAA..."。其它主机行、
-    注释行、空行一律丢掉，绝不把陌生主机的 key 写进文件。"""
+    注释行、空行一律丢掉，绝不把陌生主机的 key 写进文件。
+    host 缺省回环仅为旧调用点兼容；客机 IP 方案下应传
+    解析出的客机 IP。"""
     kept = []
     for line in (text or "").splitlines():
         line = line.strip()
@@ -70,9 +84,9 @@ def parse_keyscan_output(text):
         parts = line.split()
         if len(parts) < 3:
             continue
-        host, keytype = parts[0], parts[1]
+        line_host, keytype = parts[0], parts[1]
-        if host != "127.0.0.1" \
+        if line_host != host \
-                and not host.startswith("[127.0.0.1]:"):
+                and not line_host.startswith("[%s]:" % host):
             continue
         if not keytype.startswith("ssh-"):
             continue

cloud/tests/test_chv_fix.py

@@ -0,0 +1,554 @@
+"""CHV 审核修复轨测试（堆叠于 #230，base=feat-chv-gaps）。
+
+只用标准库与 mock runner/launcher/api_client，绝不启动真实
+VM：当前环境没有可跑真 VM 的宿主。真实 VM 启动、seed 盘
+在真机上的 cloud-init 应用、SSH 可达实测一律不在本文件
+声称已验证；本文件只证明代码层面的接线（VmConfig 含 seed
+盘、restore 不回落、reuse 拦截、SSH 目标解析、配额/快照/
+API 完整性等）成立。
+"""
+
+import json
+import os
+import socket
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.chv.api_client import _unix_socket_api
+    from cloud.chv.disk import check_disk_quota, provision_task_disk
+    from cloud.chv.hostkeys import build_seed_disk, seed_disk_path
+    from cloud.chv.pool import WarmPool
+    from cloud.sandbox import (
+        ChvFailure,
+        CloudHypervisorExecutor,
+    )
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.api_client import _unix_socket_api
+    from cloud.chv.disk import check_disk_quota, provision_task_disk
+    from cloud.chv.hostkeys import build_seed_disk, seed_disk_path
+    from cloud.chv.pool import WarmPool
+    from cloud.sandbox import (
+        ChvFailure,
+        CloudHypervisorExecutor,
+    )
+    from cloud.worker import Worker
+
+
+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
+
+    def poll(self):
+        return None if self._alive else 1
+
+    def terminate(self):
+        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 ChvFixCase(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.api_payloads = {}
+        self.audit = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _default_runner(self, cmd, timeout=None):
+        self.calls.append(list(cmd))
+        if cmd and cmd[0] == "ssh":
+            if cmd[-1] == "true":
+                return FakeResult(0, "", "")
+            if "goal" in cmd:
+                return FakeResult(0, "goal met after rounds\n", "")
+            return FakeResult(0, "", "")
+        if cmd and cmd[0] == "ssh-keygen":
+            priv = cmd[cmd.index("-f") + 1]
+            with open(priv, "w") as fh:
+                fh.write("-----BEGIN OPENSSH PRIVATE KEY-----\nFAKE\n")
+            os.chmod(priv, 0o600)
+            with open(priv + ".pub", "w") as fh:
+                fh.write("ssh-ed25519 AAAAfake1 vm-key\n")
+            return FakeResult(0, "", "")
+        if cmd and cmd[0] == "genisoimage":
+            out = cmd[cmd.index("-output") + 1]
+            with open(out, "wb") as fh:
+                fh.write(b"CIDATA" * 16)
+            return FakeResult(0, "", "")
+        return FakeResult(0, "", "")
+
+    def _default_api(self, socket_path, method, path, payload=None):
+        self.api_calls.append((method, path))
+        if path == "/vm.create":
+            self.api_payloads["vm.create"] = payload
+        return (200, {})
+
+    def make_exec(self, runner=None, env_extra=None, **overrides):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_SSH_FORWARD": "1"}
+        if env_extra:
+            env.update(env_extra)
+        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 or self._default_runner,
+            launcher=lambda cmd: FakeProc(alive=True),
+            api_client=self._default_api,
+            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
+
+
+# -- P0-1 seed 磁盘接入 -------------------------------------------------
+class SeedDiskTest(ChvFixCase):
+    def test_vm_config_carries_seed_disk_readonly(self):
+        ex = self.make_exec()
+        cfg = ex.build_vm_config("/tmp/x/disk.raw",
+                                 seed_disk="/tmp/x/seed.iso")
+        self.assertEqual(len(cfg["disks"]), 2)
+        self.assertEqual(cfg["disks"][1]["path"], "/tmp/x/seed.iso")
+        self.assertTrue(cfg["disks"][1]["readonly"])
+        cfg2 = ex.build_vm_config("/tmp/x/disk.raw")
+        self.assertEqual(len(cfg2["disks"]), 1)
+
+    def test_provisioned_wake_mounts_seed_disk(self):
+        ex = self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        out = ex.wake_vm({"id": "seed1"})
+        self.assertEqual(out["state"], "running")
+        payload = self.api_payloads["vm.create"]
+        self.assertEqual(len(payload["disks"]), 2)
+        seed_disk = payload["disks"][1]["path"]
+        self.assertTrue(seed_disk.endswith("seed.iso"))
+        self.assertTrue(payload["disks"][1]["readonly"])
+        self.assertTrue(os.path.exists(seed_disk))
+        self.assertIn("genisoimage", [c[0] for c in self.calls])
+
+    def test_seed_disk_build_failure_is_fail_fast(self):
+        def bad_runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd and cmd[0] == "genisoimage":
+                return FakeResult(1, "", "no genisoimage")
+            return self._default_runner(cmd, timeout)
+
+        ex = self.make_exec(
+            runner=bad_runner,
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        with self.assertRaises(RuntimeError):
+            ex.wake_vm({"id": "seedfail"})
+        # 制盘失败在起 VMM 之前，不应走到 vm.create。
+        self.assertNotIn("/vm.create",
+                         [p for _, p in self.api_calls])
+
+    def test_build_seed_disk_rejects_missing_product(self):
+        with self.assertRaises(RuntimeError):
+            build_seed_disk(self.dir, os.path.join(self.dir, "no.iso"),
+                            runner=lambda cmd: FakeResult(0, "", ""))
+
+
+# -- P0-2 restore 不回落 ------------------------------------------------
+class RestoreNoFallbackTest(ChvFixCase):
+    def test_wait_guest_ready_provisioned_missing_known_hosts(self):
+        ex = self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready({"id": "nopin"})
+        self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
+        self.assertFalse(any(c[0] == "ssh-keyscan" for c in self.calls))
+
+    def test_restore_provisioned_missing_known_hosts_fails_before_launch(self):
+        ex = self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        launches = []
+        ex.launcher = lambda cmd: launches.append(cmd) or FakeProc(True)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm({"id": "r1"}, "file:///snap/x")
+        self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
+        self.assertEqual(launches, [])
+
+
+# -- P0-3 reuse 拦截 ----------------------------------------------------
+class ReuseBlockTest(ChvFixCase):
+    def test_reuse_blocked_without_experiment_flag(self):
+        ex = self.make_exec(env_extra={"CHV_DISK_MODE": "reuse"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind,
+                         "disk_mode_reuse_blocked")
+
+    def test_reuse_passes_with_experiment_flag_and_is_audited(self):
+        ex = self.make_exec(env_extra={
+            "CHV_DISK_MODE": "reuse",
+            "CHV_REUSE_SINGLE_TENANT_EXPERIMENT": "1"})
+        self.assertTrue(ex.preflight())
+        out = ex.wake_vm({"id": "reuse1"})
+        self.assertEqual(out["state"], "running")
+        self.assertTrue(any(e == "disk_reuse_experiment_allowed"
+                            for e, _, _, _ in self.audit))
+
+
+# -- P0-4 SSH 目标解析 --------------------------------------------------
+class SshTargetTest(ChvFixCase):
+    def test_explicit_host_wins(self):
+        ex = self.make_exec(ssh_host="10.9.8.7")
+        self.assertEqual(ex.resolve_ssh_host(), "10.9.8.7")
+
+    def test_net_guest_ip_used_when_no_explicit_host(self):
+        ex = self.make_exec(env_extra={
+            "CHV_NET_JSON": '[{"tap": "", "ip": "192.168.100.5",'
+                            '"mask": "255.255.255.0"}]',
+            "CHV_SSH_FORWARD": "0"})
+        # env_extra 覆盖 forward=0 后仍应走 net ip，不报未声明。
+        self.assertEqual(ex.resolve_ssh_host(), "192.168.100.5")
+        cmd = ex.build_ssh_command({"id": "t"}, ["true"])
+        self.assertIn("root@192.168.100.5", " ".join(cmd))
+
+    def test_undeclared_target_is_fail_fast(self):
+        ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.resolve_ssh_host()
+        self.assertEqual(ctx.exception.kind, "ssh_target_undeclared")
+
+    def test_forward_attestation_allows_loopback(self):
+        ex = self.make_exec()
+        self.assertEqual(ex.resolve_ssh_host(), "127.0.0.1")
+
+
+# -- P1-1 网关 extra_env ------------------------------------------------
+class GatewayExtraEnvTest(ChvFixCase):
+    def test_chv_ssh_command_carries_extra_env(self):
+        ex = self.make_exec()
+        cmd = ex.build_ssh_command({"id": "t"}, ["true"],
+                                   extra_env={"SUB2API_API_KEY": "k1"})
+        joined = " ".join(cmd)
+        self.assertIn("SUB2API_API_KEY=k1", joined)
+        self.assertIn("HTTPS_PROXY=http://10.0.0.1:8888", joined)
+
+    def test_worker_gateway_guard_allows_chv(self):
+        from cloud.server import init_db
+        db_path = os.path.join(self.dir, "control.db")
+        init_db(db_path)
+        ex = self.make_exec()
+        worker = Worker(db_path, "/fake/host.py", executor=ex,
+                        env={"XCM_API_KEY": "test-placeholder"})
+        task = {"id": "gw1", "project_id": "p1", "prompt": "做目标",
+                "goal_mode": 1,
+                "_gateway_provider": "sub2api",
+                "_gateway_env": {"SUB2API_API_KEY": "gwkey",
+                                 "SUB2API_BASE_URL": "http://gw/"}}
+        result, error = worker._run_attempt(task)
+        self.assertIsNone(error)
+        self.assertEqual(result["status"], "completed")
+        ssh_calls = [c for c in self.calls if c and c[0] == "ssh"]
+        self.assertTrue(any("SUB2API_API_KEY=gwkey" in " ".join(c)
+                            for c in ssh_calls))
+
+
+# -- P1-2 取消/清理/看门狗 ----------------------------------------------
+class CancelCleanupTest(ChvFixCase):
+    def test_worker_cancel_calls_cancel_vm(self):
+        from cloud.server import init_db
+        db_path = os.path.join(self.dir, "control.db")
+        init_db(db_path)
+        ex = self.make_exec()
+        cancelled = []
+        ex.cancel_vm = lambda name: cancelled.append(name) or True
+        worker = Worker(db_path, "/fake/host.py", executor=ex,
+                        env={"XCM_API_KEY": "test-placeholder"})
+        worker._running_set("t-cancel")
+        import sqlite3, time
+        conn = sqlite3.connect(db_path)
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, ?, ?, ?, NULL, ?, ?, 'tester', ?, ?, NULL)",
+            ("t-cancel", "p1", "做事", "running", "[]", 0,
+             time.time(), time.time()))
+        conn.commit()
+        conn.close()
+        self.assertTrue(worker.cancel("t-cancel"))
+        self.assertEqual(cancelled, ["t-cancel"])
+
+    def test_worker_cleanup_calls_cleanup_stale_vms(self):
+        from cloud.server import init_db
+        db_path = os.path.join(self.dir, "control.db")
+        init_db(db_path)
+        ex = self.make_exec()
+        ex.cleanup_stale_vms = lambda: ["chv-stale"]
+        worker = Worker(db_path, "/fake/host.py", executor=ex,
+                        env={"XCM_API_KEY": "test-placeholder"})
+        self.assertIn("chv-stale", worker.cleanup_stale_containers())
+
+    def test_chv_cancel_watchdog_kills_vm_on_cancel(self):
+        import threading
+        from cloud.server import init_db
+        db_path = os.path.join(self.dir, "control.db")
+        init_db(db_path)
+        ex = self.make_exec()
+        killed = []
+        ex.cancel_vm = lambda name: killed.append(name) or True
+        worker = Worker(db_path, "/fake/host.py", executor=ex,
+                        env={"XCM_API_KEY": "test-placeholder"})
+        worker._cancel_requested.add("t-wd")
+        stop = threading.Event()
+        worker._chv_cancel_watchdog("t-wd", "chv-t-wd", stop)
+        self.assertEqual(killed, ["chv-t-wd"])
+
+
+# -- P1-3 WarmPool acquire resume ---------------------------------------
+class PoolAcquireResumeTest(unittest.TestCase):
+    def test_acquire_resumes_paused_vm(self):
+        class Stub:
+            def __init__(self):
+                self.resumed = []
+                self.state_dir = None
+                self.warm_pool_size = 1
+                self.snapshot_url = "file:///snap/base"
+
+            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):
+                return {"vm": self.vm_name(task)}
+
+            def pause_vm(self, task):
+                return True
+
+            def resume_vm(self, task):
+                self.resumed.append(task["id"])
+                return True
+
+            def shutdown_vm(self, task_or_name):
+                return True
+
+            def _emit(self, event, task, detail="", kind=None):
+                pass
+
+        stub = Stub()
+        pool = WarmPool(stub, size=1, snapshot_url="file:///snap/base",
+                        registry_path=None)
+        pool.prewarm()
+        got = pool.acquire({"id": "t1"})
+        self.assertEqual(got["mode"], "warm")
+        self.assertEqual(stub.resumed, ["warm-1"])
+
+    def test_acquire_discards_vm_whose_resume_fails(self):
+        class Stub:
+            def __init__(self):
+                self.state_dir = None
+                self.warm_pool_size = 1
+                self.snapshot_url = "file:///snap/base"
+
+            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):
+                return {"vm": self.vm_name(task)}
+
+            def pause_vm(self, task):
+                return True
+
+            def resume_vm(self, task):
+                raise RuntimeError("resume 失败")
+
+            def shutdown_vm(self, task_or_name):
+                return True
+
+            def _emit(self, event, task, detail="", kind=None):
+                pass
+
+        pool = WarmPool(Stub(), size=1, snapshot_url="file:///snap/base",
+                        registry_path=None)
+        pool.prewarm()
+        self.assertEqual(pool.acquire({"id": "t1"})["mode"], "cold")
+
+
+# -- P1-4 容量持久化计数 -------------------------------------------------
+class CapacityPersistentTest(ChvFixCase):
+    def test_socket_counts_after_restart(self):
+        ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"})
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-other.sock"),
+                  "w") as fh:
+            fh.write("")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm({"id": "new1"})
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+
+    def test_own_socket_does_not_block_rewake(self):
+        ex = self.make_exec(env_extra={"CHV_MAX_VMS": "1"})
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-t1.sock"),
+                  "w") as fh:
+            fh.write("")
+        # 同任务重复 wake：自己的残留 socket 不计入容量。
+        out = ex.wake_vm({"id": "t1"})
+        self.assertEqual(out["state"], "running")
+
+
+# -- P1-5 磁盘配额 -------------------------------------------------------
+class DiskQuotaTest(ChvFixCase):
+    def test_unreadable_image_is_not_silently_skipped(self):
+        with self.assertRaises(ChvFailure) as ctx:
+            check_disk_quota("/nonexistent/disk.raw", 1.0)
+        self.assertEqual(ctx.exception.kind, "disk_quota_unverified")
+
+    def test_total_usage_over_quota(self):
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "old.raw"),
+                  "wb") as fh:
+            fh.write(b"X" * 100)
+        # 配额 150 字节：已用 100 + 本次 100 > 150。
+        with self.assertRaises(ChvFailure) as ctx:
+            check_disk_quota(self.rootfs, 150 / (1024 ** 3),
+                             state_dir=self.state_dir)
+        self.assertEqual(ctx.exception.kind, "disk_quota_exceeded")
+
+    def test_existing_disk_size_mismatch_is_recopied(self):
+        task_disk = provision_task_disk(self.state_dir, "chv-a",
+                                        self.rootfs, "copy")
+        with open(task_disk, "wb") as fh:
+            fh.write(b"Z" * 10)
+        task_disk2 = provision_task_disk(self.state_dir, "chv-a",
+                                         self.rootfs, "copy")
+        self.assertEqual(task_disk, task_disk2)
+        with open(task_disk2, "rb") as fh:
+            self.assertEqual(fh.read(), b"R" * 100)
+
+
+# -- P1-6 API 截断 -------------------------------------------------------
+class ApiTruncationTest(unittest.TestCase):
+    def _serve(self, raw):
+        tmp = tempfile.TemporaryDirectory()
+        sock_path = os.path.join(tmp.name, "vmm.sock")
+        srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        srv.bind(sock_path)
+        srv.listen(1)
+
+        def _run():
+            conn, _ = srv.accept()
+            try:
+                conn.recv(65536)
+                conn.sendall(raw)
+            finally:
+                conn.close()
+                srv.close()
+
+        threading.Thread(target=_run, daemon=True).start()
+        return tmp, sock_path
+
+    def test_truncated_body_raises(self):
+        body = b'{"ok": true}'
+        raw = (b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n"
+               + body)
+        tmp, sock_path = self._serve(raw)
+        with self.assertRaises(RuntimeError):
+            _unix_socket_api(sock_path, "GET", "/vmm.ping")
+        tmp.cleanup()
+
+    def test_chunked_rejected(self):
+        raw = (b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n"
+               b"5\r\nhello\r\n0\r\n\r\n")
+        tmp, sock_path = self._serve(raw)
+        with self.assertRaises(RuntimeError):
+            _unix_socket_api(sock_path, "GET", "/vmm.ping")
+        tmp.cleanup()
+
+    def test_full_body_parses(self):
+        body = b'{"ok": true}'
+        raw = (b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n"
+               % len(body)) + body
+        tmp, sock_path = self._serve(raw)
+        status, parsed = _unix_socket_api(sock_path, "GET", "/vmm.ping")
+        self.assertEqual(status, 200)
+        self.assertTrue(parsed["ok"])
+        tmp.cleanup()
+
+
+# -- P1-7 快照落盘验证 ---------------------------------------------------
+class SnapshotPersistTest(ChvFixCase):
+    def test_snapshot_missing_file_does_not_shutdown(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.hibernate_vm({"id": "h1"}, "file:///nonexistent/snap")
+        self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
+        kinds = [c[-1] for c in self.calls if c[0] == self.ch_remote]
+        self.assertIn("pause", kinds)
+        self.assertNotIn("shutdown", kinds)
+
+
+# -- P1-8 execute 死端口归类 ---------------------------------------------
+class ExecuteDeadPortTest(ChvFixCase):
+    def test_dead_socket_ssh_failure_is_vm_not_running(self):
+        def raising_runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd and cmd[0] == "ssh":
+                raise RuntimeError("connection refused")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=raising_runner)
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-dead.sock"),
+                  "w") as fh:
+            fh.write("")
+        ex._vms["chv-dead"] = FakeProc(alive=False)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.execute({"id": "dead"})
+        self.assertEqual(ctx.exception.kind, "vm_not_running")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_chv_gaps.py

@@ -119,7 +119,10 @@ class ChvGapCase(unittest.TestCase):
         return FakeProc(alive=True)
 
     def make_exec(self, runner=None, env_extra=None, **overrides):
-        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888"}
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               # 实验室宣称：mock runner 假装回环端口已转发到客机
+               #（feat-chv-fix 起，执行器不再无条件假设回环能通）。
+               "CHV_SSH_FORWARD": "1"}
         if env_extra:
             env.update(env_extra)
         kwargs = dict(

cloud/tests/test_chv_optimize.py

@@ -108,7 +108,10 @@ class ChvOptCase(unittest.TestCase):
 
     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"}
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               # 实验室宣称：mock runner 假装回环端口已转发到客机
+               #（feat-chv-fix 起，执行器不再无条件假设回环能通）。
+               "CHV_SSH_FORWARD": "1"}
         if env_extra:
             env.update(env_extra)
         if runner is None:

cloud/tests/test_cloud_hypervisor.py

@@ -98,7 +98,10 @@ class ChvCase(unittest.TestCase):
 
     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"}
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               # 实验室宣称：mock runner 假装回环端口已转发到客机
+               #（feat-chv-fix 起，执行器不再无条件假设回环能通）。
+               "CHV_SSH_FORWARD": "1"}
         if env_extra:
             env.update(env_extra)
         if runner is None:
@@ -410,9 +413,15 @@ class LifecycleTest(ChvCase):
 
     def test_hibernate_is_pause_snapshot_shutdown(self):
         ex = self.make_exec()
-        out = ex.hibernate_vm(self._task(), "file:///snap/h1")
+        # feat-chv-fix（P1-7）：file:// 快照须真实存在且非空，
+        # 否则 hibernate 不 shutdown；用临时文件作快照目标。
+        snap = os.path.join(self.dir, "h1.snap")
+        with open(snap, "wb") as fh:
+            fh.write(b"S" * 64)
+        snap_url = "file://" + snap
+        out = ex.hibernate_vm(self._task(), snap_url)
         self.assertEqual(out["state"], "hibernated")
-        self.assertEqual(out["snapshot"], "file:///snap/h1")
+        self.assertEqual(out["snapshot"], snap_url)
         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")
@@ -420,7 +429,7 @@ class LifecycleTest(ChvCase):
         self.assertIn("shutdown", kinds)
         # 从休眠唤醒 = restore + resume：
         ex2 = self.make_exec()
-        out2 = ex2.wake_from_hibernate(self._task(), "file:///snap/h1")
+        out2 = ex2.wake_from_hibernate(self._task(), snap_url)
         self.assertEqual(out2["mode"], "restored")
 
     def test_stop_dead_proc(self):

cloud/worker.py

@@ -333,22 +333,28 @@ def _merge_env(base, override):
     return merged
 
 
-def _call_goal_resume_breakpoint(executor, task, model_provider=None):
+def _call_goal_resume_breakpoint(executor, task, model_provider=None,
+                               extra_env=None):
     """通用断点调用：执行器提供 goal_resume_breakpoint_task 时调用它。
 
     Local 仍走原路径（调用方直接传 model_provider，不经过这里，
     以免改生产语义）；这里只用于非 Local 执行器：签名接受
     model_provider 时传它（CHV/Docker 均接受），否则只传 task
-    （旧假执行器单参）。网关 key 绝不经这里传值：worker 侧
+    （旧假执行器单参）。feat-chv-fix（P1-1）：CHV 的
-    gateway_env 只按名注入，见 _run_attempt 的网关守卫。"""
+    gateway_env 经此进入 SSH env 前缀注入客机环境（不进
+    镜像/代码/prompt）；CHV 没有 Local 那样的进程 env 通道，
+    SSH env 前缀是唯一通道，值只出现在宿主侧 ssh argv。"""
     method = getattr(executor, "goal_resume_breakpoint_task")
     try:
         params = inspect.signature(method).parameters
     except (TypeError, ValueError):
         return method(task)
+    kwargs = {}
     if "model_provider" in params:
-        return method(task, model_provider=model_provider)
+        kwargs["model_provider"] = model_provider
-    return method(task)
+    if "extra_env" in params and extra_env is not None:
+        kwargs["extra_env"] = extra_env
+    return method(task, **kwargs)
 
 
 def _resolve_model(model=None, config_path=None):
@@ -2368,6 +2374,12 @@ class Worker:
                 self.executor.kill_container(entry["container"])
             except Exception:
                 pass
+        # feat-chv-fix（P1-2）：CHV 取消真关 VM，不只改库状态。
+        if hasattr(self.executor, "cancel_vm"):
+            try:
+                self.executor.cancel_vm(task_id)
+            except Exception:
+                pass
         self._running_clear(task_id)
         conn = _db_connect(self.db_path)
         try:
@@ -2470,22 +2482,87 @@ class Worker:
                 pass
             return
 
+    def _chv_cancel_watchdog(self, task_id, vm_name, stop_event):
+        """CHV 同步路径的取消看门狗（feat-chv-fix，P1-2，Docker
+        取消看门狗同款模式）：同步 SSH 阻塞时 worker 的 cancel()
+        只能改库、清 _running 项，阻塞中的 run 本身不会醒；
+        本线程每 1 秒查库/取消标志，命中时调 cancel_vm 真关
+        VM，然后退出。查库失败时当没取消，不误杀。"""
+        while not stop_event.wait(1.0):
+            try:
+                if task_id in self._cancel_requested:
+                    cancelled = True
+                else:
+                    db_task = self.get_task(task_id)
+                    cancelled = (db_task is not None and
+                                 db_task.get("status") == "interrupted")
+            except Exception:
+                cancelled = False
+            if not cancelled:
+                continue
+            try:
+                cancel = getattr(self.executor, "cancel_vm", None)
+                if cancel is not None and vm_name:
+                    cancel(vm_name)
+                    log_event(self.logger, "worker_chv_killed",
+                              task_id=task_id,
+                              message="cancel watchdog killed VM %s "
+                                      "(task interrupted/cancelled)"
+                                      % vm_name)
+            except Exception:
+                pass
+            return
+
+    def _chv_quota_watchdog(self, task_id, vm_name, stop_event,
+                            quota_check, interval=30.0):
+        """CHV 同步路径的配额看门狗（feat-chv-fix，P1-2，Docker
+        配额看门狗同款）：命中时记 _quota_hit_tasks 并调
+        cancel_vm 真关 VM；_run_attempt 在同步返回后把结果
+        改写成 quota_exceeded，走统一配额中断收尾。"""
+        while not stop_event.wait(interval):
+            try:
+                over = bool(quota_check()) if quota_check else False
+            except Exception:
+                over = False
+            if not over:
+                continue
+            self._mark_quota_hit(task_id)
+            try:
+                cancel = getattr(self.executor, "cancel_vm", None)
+                if cancel is not None and vm_name:
+                    cancel(vm_name)
+                    log_event(self.logger, "worker_chv_killed",
+                              task_id=task_id,
+                              message="quota watchdog killed VM %s "
+                                      "(quota exceeded)" % vm_name)
+            except Exception:
+                pass
+            return
+
     def cleanup_stale_containers(self):
         """启动清理兜底：清掉上一轮残留的 codex- 容器。
 
-        只在执行器有 cleanup_stale_containers 时做（Docker）；
+        Docker 执行器清容器、CHV 执行器清残留 VM（feat-chv-fix）；
         列出或删除失败只记日志，绝不抛异常炸掉 worker。
-        返回已清理的容器名列表（非 Docker 执行器返回 []）。
+        返回已清理的容器/VM 名列表（两者皆无的执行器返回 []）。
         """
+        cleaned = []
         cleanup = getattr(self.executor, "cleanup_stale_containers", None)
-        if cleanup is None:
+        if cleanup is not None:
-            return []
+            try:
-        try:
+                cleaned = list(cleanup() or [])
-            cleaned = cleanup()
+            except Exception as exc:
-        except Exception as exc:
+                log_event(self.logger, "worker_cleanup_failed",
-            log_event(self.logger, "worker_cleanup_failed",
+                          message="cleanup stale containers failed: %s" % exc)
-                      message="cleanup stale containers failed: %s" % exc)
+        # feat-chv-fix（P1-2）：CHV 残留 VM 走通用清理路径，
-            return []
+        # 不再只靠 worker 启动/关闭的 WarmPool reconcile 兜底。
+        chv_cleanup = getattr(self.executor, "cleanup_stale_vms", None)
+        if chv_cleanup is not None:
+            try:
+                cleaned = cleaned + list(chv_cleanup() or [])
+            except Exception as exc:
+                log_event(self.logger, "worker_cleanup_failed",
+                          message="cleanup stale vms failed: %s" % exc)
         if cleaned:
             log_event(self.logger, "worker_cleanup",
                       message="cleaned stale containers: %s"
@@ -4639,6 +4716,29 @@ class Worker:
                               _watchdog_stop, quota_check),
                         daemon=True)
                     _quota_watchdog_thread.start()
+        # feat-chv-fix（P1-2）：CHV 同步路径同款看门狗（VM 名
+        # 由任务 id 派生）。进程模式不重复起（_execute_via_
+        # process 已轮询 quota_check）。
+        if isinstance(self.executor, CloudHypervisorExecutor) \
+                and hasattr(self.executor, "cancel_vm"):
+            try:
+                _chv_vm_name = self.executor.vm_name(task)
+            except Exception:
+                _chv_vm_name = None
+            if _chv_vm_name:
+                _watchdog_stop = threading.Event()
+                _watchdog_thread = threading.Thread(
+                    target=self._chv_cancel_watchdog,
+                    args=(task_id, _chv_vm_name, _watchdog_stop),
+                    daemon=True)
+                _watchdog_thread.start()
+                if quota_check is not None and not self._use_popen():
+                    _quota_watchdog_thread = threading.Thread(
+                        target=self._chv_quota_watchdog,
+                        args=(task_id, _chv_vm_name,
+                              _watchdog_stop, quota_check),
+                        daemon=True)
+                    _quota_watchdog_thread.start()
         # 启动前分段日志（2026-10-29 任务 A）：认领后 61 秒中断，
         # 但阻塞的 docker run 中途没有任何标记、容器 --rm 后输出
         # 也没了，事后完全无法定位卡在哪一段。派发开始/结束进
@@ -4774,8 +4874,12 @@ class Worker:
                     # Docker 路径上面已有 is_docker 守卫，不走这里。
                     # 网关任务在非 Local 执行器上无法注入 key 与
                     # provider，必须 loud fail，不静默跑错网关。
+                    # feat-chv-fix（P1-1）：守卫收窄——CHV 已补
+                    # extra_env SSH 注入路径，不再被“非 Local 即
+                    # 拦”；无注入能力的执行器仍 loud fail。
                     if gateway_provider and not isinstance(
-                            self.executor, LocalExecutor):
+                            self.executor,
+                            (LocalExecutor, CloudHypervisorExecutor)):
                         raise RuntimeError(
                             "sub2api 网关不支持该执行器：%s"
                             % type(self.executor).__name__)
@@ -4796,7 +4900,8 @@ class Worker:
                             # 故这里不放宽网关规则。
                             result = _call_goal_resume_breakpoint(
                                 self.executor, task,
-                                model_provider=gateway_provider)
+                                model_provider=gateway_provider,
+                                extra_env=gateway_env)
                         elif task.get("_goal_resume_breakpoint"):
                             # 假执行器/CloudExecutor 没有断点方法时
                             # 绝不回落成 goal（会撞已存在）或
@@ -4825,6 +4930,11 @@ class Worker:
                                 task, seed_dir=seed_host_dir,
                                 extra_env=gateway_env,
                                 model_provider=gateway_provider)
+                        elif isinstance(
+                                self.executor, CloudHypervisorExecutor):
+                            result = self.executor.goal_task(
+                                task, extra_env=gateway_env,
+                                model_provider=gateway_provider)
                         else:
                             result = self.executor.goal_task(task)
                     else:
@@ -4833,6 +4943,11 @@ class Worker:
                                 task, seed_dir=seed_host_dir,
                                 extra_env=gateway_env,
                                 model_provider=gateway_provider)
+                        elif isinstance(
+                                self.executor, CloudHypervisorExecutor):
+                            result = self.executor.dispatch_task(
+                                task, extra_env=gateway_env,
+                                model_provider=gateway_provider)
                         else:
                             result = self.executor.dispatch_task(task)
                 combined = (result.get("stdout", "") or "") + \
@@ -4856,14 +4971,18 @@ class Worker:
                             quota_check)
                     else:
                         if gateway_provider and not isinstance(
-                                self.executor, LocalExecutor):
+                                self.executor,
+                                (LocalExecutor,
+                                 CloudHypervisorExecutor)):
                             raise RuntimeError(
                                 "sub2api 网关不支持该执行器：%s"
                                 % type(self.executor).__name__)
                         result = self.executor.run_task(
                             task, extra_env=gateway_env,
                             model_provider=gateway_provider) \
-                            if isinstance(self.executor, LocalExecutor) \
+                            if isinstance(self.executor,
+                                          (LocalExecutor,
+                                           CloudHypervisorExecutor)) \
                             else self.executor.run_task(task)
         except Exception as exc:  # runner 异常也如实写回 failed
             result = {"executor": "local", "task_id": task_id,

docs/chv-production-readiness.md

@@ -36,7 +36,7 @@
   shutdown，同一任务重复 wake 先关旧 VM；VM 名、api-socket、
   每任务磁盘路径均按任务派生，互不相同（测试断言）。
 - ✅🧪 每任务磁盘独立：copy 模式复制基础镜像为每任务磁盘；
-  reuse 模式仅单租户实验，生产禁用（代码与文档双写）。
+  reuse 模式仅单租户实验，preflight 拦截（须显式实验二次确认并审计放行，feat-chv-fix 起代码真正拦截；此前“代码与文档双写”的说法是错的，代码当时并不拦截）。
 - ✅🧪 租户无共享路径：执行器不挂载宿主根、宿主代码目录或含
   密钥目录；客机代码在镜像内 `/app`，不在宿主共享挂载上。
 - ✅🧪 密钥不进镜像/代码/prompt：SSH argv 只有 key 文件路径；
@@ -44,7 +44,7 @@
   命令中不含哨兵密钥值。
 - ✅🧪 SSH host key 按 VM 钉死（优化轨，#216 的 no + /dev/null
   已废除）：每 VM 在 state_dir/<vm>/known_hosts 有独立文件，
-  首次启动经 ssh-keyscan 在本机回环端口捕获 key 写入该文件，
+  首次启动经 ssh-keyscan 在解析出的 SSH 目标上捕获 key 写入该文件，
   此后 SSH 一律 StrictHostKeyChecking=yes + 该文件校验；
   非 root 客机用户（CHV_SSH_USER，显式空串直接拒绝）与端口
   （CHV_SSH_PORT，1-65535）为显式配置。mock 只验证捕获→
@@ -209,3 +209,33 @@
   chv-guest 镜像构建发布 → 真实 VM 启动/快照恢复/任务端到端
   验证 → 压测 → 安全审计 → CI 变绿后按当时 main 头复核 diff
   再合并部署。缺一不可。
+
+## 10. 审核修复轨 feat-chv-fix（堆叠于 #230 之上）
+
+- ✅🧪 P0 已修：seed 磁盘（CIDATA ISO，第二块只读盘）挂进
+  VmConfig；restore 不回落 keyscan（provisioned 下复用已钉
+  死 known_hosts、缺失 fail-fast，快照内 host key 不重预置）；
+  reuse 由 preflight 拦截（显式实验二次确认 + 审计放行）；
+  SSH 目标选客机 IP 方案（显式 host > net 静态 ip > 显式
+  转发宣称下的回环，三者皆无 fail-fast；vsock 与纯 NAT 方案
+  未选，理由见 config.py）。
+- ✅🧪 P1 已修或有结论：网关 extra_env 经 SSH env 前缀注入，
+  worker 守卫收窄到 Local/CHV；取消/清理/看门狗接 CHV 通用
+  路径；WarmPool acquire 内 resume（但 worker dispatch 前查
+  池仍未接线，预热收益仍不是已交付特性）；容量门改持久化
+  计数（socket 联合计数）；磁盘配额改总量+余量、已存在磁盘
+  校验重拷；API 客户端按 Content-Length 验完整性；快照
+  file:// 目标验存在且非空再 shutdown；execute 死端口归类
+  vm_not_running。
+- ⚠️ 仍未接线/未验证（不得声称已交付）：worker dispatch 前查
+  WarmPool；CHV + 网关真实任务；真机 wake/boot/SSH（含 seed
+  注入与 NAT/客机 IP 实测）；快照恢复端到端；压测；安全审计；
+  CI 账单恢复。另：P0-2 审计原说法“restore 须把 seed 接入
+  VmConfig”已纠正——restore 不走 vm.create，seed 无从挂盘，
+  正确不变量是复用钉死 known_hosts。
+- ⛔ P2 全部排除、未做（用户紧急范围收窄令，留待以后一轨）：
+  长函数拆分、第二启动路径清理 / Docker goal 解析注释、
+  test_chv_gaps.py 的 grep 文本断言改行为断言、build.sh 与
+  README 口径对齐、stop_vm 吞异常改审计、配置校验时机统一。
+  本轨的 grep 文本测试、README/build.sh、stop_vm 吞异常、
+  preflight 校验口径均保持 #230 原样。