feat(chv): 阶段 1 持久 VM、overlay 分层与预热快照

- disk.py：overlay 模式（qcow2 backing file 指只读 base，
  建盘/复用/重建校验 + qemu-img fail-closed）、数据盘
  data.qcow2 独立供给（不进快照/GC/stop）、配额增量口径
  （overlay 本次增量 0）、快照登记 tenant 字段、每 VM
  tenant.json、持久登记 persistent/<tenant>.json、快照
  登记按 URL 查找。
- config.py：CHV_DISK_MODE 白名单加 overlay（默认仍 copy；
  错误文案保留旧子串，防回归测试钉死）。
- executor.py：VmConfig image_type 按后缀（.qcow2→Qcow2，
  其余 Raw）+ 数据盘第三块盘；overlay 供给写 tenant 标记；
  snapshot_vm 登记带 tenant，_validate_restore 归属校验
  （跨用户 snapshot_tenant_mismatch 拒绝 + 审计，legacy
  未标记/未登记放行并警告）；stop/GC 在 overlay 模式下
  保护持久登记/快照认领的盘；持久 VM（prewarm/acquire/
  suspend/dispatch_task_persistent/recycle_persistent_vm，
  CHV_WARM_POOL_SIZE>=1 启用，=0 退化冷启旧语义），
  resume 只认 vm.info=Paused + socket 活性 + 归属一致。
- tests：cloud/tests/test_chv_opt_stage1.py 26 项（overlay、
  tenant、持久池、跨用户拒绝、两连任务、显式回收）。
回归：discover test_chv_*.py 与 test_cloud_hypervisor.py
全过（结构契约：函数 ≤50 行、公开 API 全注解、文案钉死）。
不推、不合、不部署；只本地分支。

cloud/chv/executor.py

@@ -42,11 +42,22 @@ from cloud.chv.config import _parse_memory_bytes, build_chv_config
 from cloud.chv.disk import (
     check_disk_quota,
     check_snapshot_limits,
+    clear_persistent_vm,
+    data_disk_path,
+    delete_data_disk,
     delete_task_disk,
+    find_snapshot_entry,
     has_recorded_snapshots,
+    overlay_disk_path,
+    provision_data_disk,
     provision_task_disk,
+    read_persistent_vm,
+    read_vm_tenant,
     record_snapshot,
     task_disk_path,
+    task_disk_paths,
+    write_persistent_vm,
+    write_vm_tenant,
 )
 from cloud.chv.errors import ChvFailure
 from cloud.chv.hardening import (
@@ -211,6 +222,13 @@ def _ensure_snapshot_dir(snapshot_url, snapshot_path) -> None:
             "快照目录无法创建：%s：%s" % (snapshot_url, exc))
 
 
+def _disk_image_type(path: str) -> str:
+    """VmConfig 磁盘 image_type 枚举原名（feat-chv-opt-landing）：
+    .qcow2 后缀即 Qcow2，其余即 Raw（raw 无魔数、JSON 里写
+    小写 raw 会报 unknown variant，见 _vm_config_disks）。"""
+    return "Qcow2" if str(path).endswith(".qcow2") else "Raw"
+
+
 def _default_popen_launcher(cmd):
     """默认 VMM 启动器：独立会话子进程，输出丢弃（纯移动）。"""
     return subprocess.Popen(
@@ -304,11 +322,21 @@ class CloudHypervisorExecutor(Executor):
     - 单任务单 VM：wake → 跑 → finally 关，同一任务不共享 VM；
       每任务磁盘 copy 模式独立，租户无共享路径、无东西向互访；
       VM 内 root 只能影响本 VM。
-    - 快速启动：WarmPool（cloud/chv/experimental_pool.py）预热干净快照 VM，
-      acquire 消耗、release 销毁，池空回落冷启动，绝不降级；
-      池状态另落 state_dir/warm-pool.json registry，worker
-      重启后载入的条目一律当孤儿、只能 reconcile_orphans()
-      清理，绝不重新发给任务（见 orphans.py）。
+    - 快速启动：持久 VM + 预热池（feat-chv-opt-landing，
+      阶段 1，生产路径，见 pause/resume 与
+      dispatch_task_persistent/prewarm_persistent_vm）：
+      同一租户的任务共用一台持久 VM（overlay 系统盘 +
+      独立数据盘），任务间隙 pause、到达时 resume；
+      CHV_WARM_POOL_SIZE>=1 时控制面常备 1 台已开机、已
+      完成 Gitea 首次同步的 paused 持久 VM，任务到达先
+      领它、结束暂停回池，池空回落冷启不报错；为 0（默认）
+      时全部退化为冷启旧语义。WarmPool（experimental_pool.py）
+      仍是纯实验（ADR-5/ADR-12），生产 dispatch 绝不查它，
+      阶段 1 的池不经过它；池状态另落
+      state_dir/persistent/<tenant>.json 登记，worker
+      重启后载入的条目须经 vm.info=Paused + socket 活性
+      + tenant 归属三重核对才认领，否则只当孤儿清理，
+      绝不直接发给任务（见 orphans.py 与 acquire_for_task）。
     - 长任务中途死亡：_run_with_vm 在客内命令抛传输异常、
       或非 goal 结果已 failed 且 VM 登记进程已死时，改判
       interrupted（断点可续跑）而不是笼统 failed；VM 还活
@@ -603,6 +631,25 @@ class CloudHypervisorExecutor(Executor):
         return os.path.join(self.state_dir,
                             self.vm_name(task_or_name) + ".sock")
 
+    def _tenant_id(self, task) -> str | None:
+        """归属租户 id（feat-chv-opt-landing）：任务 dict 的
+        tenant_id > project_id > 单租户声明
+        CHV_SINGLE_TENANT_PROJECT；三者皆无即 None（无归属
+        可标，快照/overlay 按 legacy 处理，不伪造）。"""
+        if isinstance(task, dict):
+            tenant = task.get("tenant_id") or task.get("project_id")
+            if tenant:
+                return str(tenant)
+        return getattr(self, "single_tenant_project", None)
+
+    def persistent_vm_name(self, tenant_id: str | None) -> str:
+        """该租户的持久 VM 名：chv-persist-<安全字符租户>
+        （阶段 1：同一租户的连续任务共用这一台 VM，任务间
+        暂停不断电、不删盘；租户为空即 default 单租户）。"""
+        safe = re.sub(r"[^A-Za-z0-9_.-]", "-",
+                      str(tenant_id or "default"))
+        return "chv-persist-" + safe
+
     # -- 前置校验 ------------------------------------------------------
     @staticmethod
     def _binary_ok(path):
@@ -1138,22 +1185,30 @@ class CloudHypervisorExecutor(Executor):
 
     # -- API 模式：VmConfig / API 调用 / 就绪轮询 -----------------------
     def _vm_config_disks(self, task_disk: str | None,
-                         seed_disk: str | None) -> list:
+                         seed_disk: str | None,
+                         data_disk: str | None = None) -> list:
         """VmConfig 磁盘列表（to10 提取，含 L2 白名单校验）。
         任务盘 + 可选 seed 只读盘（CIDATA ISO，cloud-init
-        NoCloud 数据源）；显式给的盘若不是基础镜像本身，
-        须在 state_dir 内，否则 VmConfig 不能挂进宿主任意文件。
-        image_type 必须显式给 Raw：CHV 54 的磁盘类型靠文件
+        NoCloud 数据源）+ 可选数据盘（overlay 模式第二块盘，
+        工作区/浏览器 profile 载体，客内按 LABEL=CHVDATA 挂
+        载到 /home/chv/workspace）；显式给的盘若不是基础
+        镜像本身，须在 state_dir 内，否则 VmConfig 不能挂
+        进宿主任意文件。
+        image_type 必须显式给：CHV 54 的磁盘类型靠文件
         魔数探测，raw 盘无魔数、探测失败即 vm.boot 400
         "Image type required for disk"（2026-11-07 WSL 真机
         CHV 54.0.0 实测）；值须是枚举原名 Raw（CLI 的小写
         raw 是另一层映射，JSON 里写 raw 会报 unknown variant）；
         CLI 各轮一直传 image_type=raw（seed ISO 亦然，客内
-        仍以 /dev/vdb 读到 CIDATA）。"""
+        仍以 /dev/vdb 读到 CIDATA）。
+        feat-chv-opt-landing：.qcow2 后缀的盘（overlay 与
+        数据盘）image_type 为 Qcow2（qcow2 有魔数可探测，
+        但与 raw 同款必须显式给枚举原名，不依赖探测）。"""
         # 旧文案逐字保留：任务盘“任务磁盘须在…”，seed 盘
         # “seed 磁盘须在…”（label 本身含“磁盘”与空格）。
         for path, label in ((task_disk, "任务磁盘"),
-                            (seed_disk, "seed 磁盘")):
+                            (seed_disk, "seed 磁盘"),
+                            (data_disk, "数据磁盘")):
             if path and path != self.rootfs:
                 if not os.path.realpath(path).startswith(
                         os.path.realpath(self.state_dir) + os.sep):
@@ -1161,13 +1216,18 @@ class CloudHypervisorExecutor(Executor):
                         "disk_path_not_allowed",
                         "%s须在 state_dir 内：%s" % (label, path))
         disks: list = [{"path": task_disk or self.rootfs,
-                        "image_type": "Raw"}]
+                        "image_type": _disk_image_type(
+                            task_disk or self.rootfs)}]
         if seed_disk:
             disks.append({"path": seed_disk, "readonly": True,
                           "image_type": "Raw"})
+        if data_disk:
+            disks.append({"path": data_disk,
+                          "image_type": _disk_image_type(data_disk)})
         return disks
 
-    def build_vm_config(self, task_disk: str | None = None, seed_disk: str | None = None) -> dict:
+    def build_vm_config(self, task_disk: str | None = None, seed_disk: str | None = None,
+                        data_disk: str | None = None) -> dict:
         """API 模式的 VmConfig（vm.create 请求体，纯函数）：
         cpus{boot_vcpus,max_vcpus}、memory{size: 字节}、
         payload{kernel, cmdline}、disks[{path: 每任务磁盘}]、
@@ -1194,7 +1254,8 @@ class CloudHypervisorExecutor(Executor):
             "cpus": {"boot_vcpus": cpus, "max_vcpus": cpus},
             "memory": {"size": _parse_memory_bytes(self.memory)},
             "payload": payload,
-            "disks": self._vm_config_disks(task_disk, seed_disk),
+            "disks": self._vm_config_disks(task_disk, seed_disk,
+                                           data_disk),
             "rng": {"src": "/dev/urandom"},
         }
         if self.net_json:
@@ -1317,9 +1378,14 @@ class CloudHypervisorExecutor(Executor):
 
     def _check_disk_quota(self):
         """磁盘配额门（实现在 cloud/chv/disk.py，feat-chv-fix
-        口径：总量 + 本次 + 宿主余量，镜像不可读不静默跳过）。"""
+        口径：总量 + 本次 + 宿主余量，镜像不可读不静默跳过）。
+        feat-chv-opt-landing：overlay 模式本次增量为 0（只
+        建空 overlay，不拷 base），配额仍按 base 大小硬帽。"""
+        increment = 0 if getattr(self, "disk_mode", "copy") \
+            == "overlay" else None
         check_disk_quota(self.rootfs, self.disk_gb,
-                         state_dir=self.state_dir)
+                         state_dir=self.state_dir,
+                         increment_bytes=increment)
 
     def _provision_task_disk(self, task):
         """每任务磁盘供给（实现在 cloud/chv/disk.py）：
@@ -1329,10 +1395,29 @@ class CloudHypervisorExecutor(Executor):
         - reuse（仅单租户实验）：直接返回基础镜像路径，磁盘
           不独立；preflight 拦截（须显式实验二次确认并审计
           放行，见 preflight，feat-chv-fix）。
+        - overlay（feat-chv-opt-landing，阶段 1）：在
+          state_dir/<vm>/disk.qcow2 建空 overlay，backing
+          file 指向只读 base；同时供给独立数据盘
+          state_dir/<vm>/data.qcow2（工作区/浏览器 profile
+          载体，不进快照）；供给后写 tenant 归属标记
+          tenant.json（reuse 不写——盘就是共享 base，标给
+          某租户是错的）。
         返回每任务磁盘路径。"""
-        return provision_task_disk(
+        disk = provision_task_disk(
             self.state_dir, self.vm_name(task), self.rootfs,
             self.disk_mode)
+        if getattr(self, "disk_mode", "copy") == "overlay":
+            provision_data_disk(self.state_dir, self.vm_name(task))
+            write_vm_tenant(self.state_dir, self.vm_name(task),
+                            self._tenant_id(task), disk=disk)
+        return disk
+
+    def _provision_data_disk_path(self, task):
+        """overlay 模式的数据盘路径（VmConfig 第三块盘）：
+        非 overlay 模式即 None（不挂数据盘，行为与旧版一致）。"""
+        if getattr(self, "disk_mode", "copy") != "overlay":
+            return None
+        return data_disk_path(self.state_dir, self.vm_name(task))
 
     def wait_api_ready(self, task: dict) -> bool:
         """等 vmm.ping 通：按次数轮询（api_timeout/1 秒取整，
@@ -1768,7 +1853,9 @@ class CloudHypervisorExecutor(Executor):
         self._check_api_socket_permissions(socket_path)
         status, _ = self._api_call(
             socket_path, "PUT", "/vm.create",
-            self.build_vm_config(task_disk, seed_disk=seed_disk))
+            self.build_vm_config(
+                task_disk, seed_disk=seed_disk,
+                data_disk=self._provision_data_disk_path(task)))
         if status not in (200, 204):
             raise ChvFailure(
                 "vm_create_failed",
@@ -1910,6 +1997,392 @@ class CloudHypervisorExecutor(Executor):
                    "vm=%s" % self.vm_name(task_or_name))
         return True
 
+    # -- 持久 VM 与预热池（feat-chv-opt-landing，阶段 1）-----------
+    # 模型（单租户灰度，CHV_WARM_POOL_SIZE=1 时启用）：
+    # 同一租户的任务共用一台持久 VM（chv-persist-<tenant>，
+    # overlay 系统盘 + 独立数据盘）；任务间隙 pause 存盘，
+    # 下个任务 resume，不走完整开机。预热池 = 常备 1 台已
+    # 开机、已完成 Gitea 首次同步、处于 paused 的持久 VM：
+    # 任务到达时先领它顶上，任务结束暂停回池（即补建回池）；
+    # 池空（无 paused 登记）自动回落冷启，不报错。
+    # 与 experimental_pool.WarmPool 的区别（ADR-5 的缺口
+    # 在此补上）：预热 VM 的盘从生下来就是该租户的持久盘
+    #（overlay + 数据盘 + tenant 标记），任务直接跑在它
+    # 自己的盘上，不存在“干净快照盘当任务盘用”的隔离击穿，
+    # 故 acquire 不需要热换盘。WarmPool 本身仍是实验模块，
+    # 生产路径不 import 它（ADR-12）；本节是生产路径。
+    # 开关：CHV_WARM_POOL_SIZE（默认 0）——为 0 时本节全部
+    # 入口退化为冷启旧语义（dispatch_task_persistent 直接
+    # 调 dispatch_task；prewarm/refill 为 no-op），绝不硬
+    # 编码改默认值偷换全局行为；阶段 1 验证用
+    # CHV_WARM_POOL_SIZE=1。
+
+    def _vm_info_state(self, task_or_name) -> str | None:
+        """GET /vm.info 取 VM 状态（Running/Paused/...）：
+        取不到/无 state 字段即 None（调用方按未知处理，绝
+        不猜它是 paused 还是 running）。"""
+        try:
+            status, body = self._api_call(
+                self.api_socket(task_or_name), "GET", "/vm.info")
+        except Exception:
+            return None
+        if status != 200 or not isinstance(body, dict):
+            return None
+        state = body.get("state")
+        return state if isinstance(state, str) else None
+
+    def _persistent_task(self, tenant_id):
+        """持久 VM 的合成任务 dict：id 固定为 persist-<租户>，
+        故 vm 名恒为 chv-persist-<租户>；gitea_sync 等任务
+        级同步变量由调用方经 extra_env 给，不进这里。"""
+        return {"id": "persist-%s" % (tenant_id or "default"),
+                "project_id": tenant_id}
+
+    def _wait_sync_workspace(self, vm_task, sync_workspace) -> bool:
+        """等 Gitea 首次同步完成：轮询客内 test -d
+        <sync_workspace>/.git（clone 落盘的硬证据）；每轮
+        间隔 ready_interval、上限 ready_timeout；超时即
+        False（调用方 fail-closed，不把未同步的 VM 当预热
+        机发给任务）。"""
+        attempts = max(
+            1, int(self.ready_timeout / self.ready_interval))
+        for attempt in range(attempts):
+            try:
+                result = self._call_runner(self.build_ssh_command(
+                    vm_task, ["test", "-d",
+                              sync_workspace + "/.git"]))
+                if getattr(result, "returncode", 1) == 0:
+                    return True
+            except Exception:
+                pass
+            if attempt + 1 < attempts:
+                self.sleeper(self.ready_interval)
+        return False
+
+    def _prewarm_build_new(self, tenant_id, extra_env, sync_workspace):
+        """预热新建段（prewarm_persistent_vm 的后半段）：冷启
+        一台持久 VM；sync_workspace 给出时等首次同步完成，
+        等不到即关 VM 报 warm_pool_sync_incomplete（绝不把
+        未同步的 VM 记成预热机）。返回 (vm_task, gitea_synced)。"""
+        vm_task = self._persistent_task(tenant_id)
+        self.wake_vm(vm_task, extra_env=extra_env)
+        synced = False
+        if sync_workspace:
+            synced = self._wait_sync_workspace(
+                vm_task, sync_workspace)
+            if not synced:
+                try:
+                    self.stop_vm(vm_task)
+                except Exception:
+                    pass
+                raise ChvFailure(
+                    "warm_pool_sync_incomplete",
+                    "预热 VM %s 在 %s 秒内未完成 Gitea 首次同步"
+                    "（%s/.git 未出现），已关 VM，不进池"
+                    % (self.vm_name(vm_task), self.ready_timeout,
+                       sync_workspace))
+        else:
+            synced = bool(getattr(self, "gitea_sync_env", None))
+        return vm_task, synced
+
+    def prewarm_persistent_vm(self, tenant_id: str | None,
+                              extra_env: dict | None = None,
+                              sync_workspace: str | None = None) -> dict:
+        """预热池进池（幂等）：CHV_WARM_POOL_SIZE>=1 才生效，
+        为 0 即 {pooled: False, reason: warm_pool_disabled}。
+        已有 paused 登记且 socket 活着即返回它（already-warm，
+        不重建）；否则冷启一台持久 VM（overlay 盘 + Gitea
+        同步 env），sync_workspace 给出时等首次同步完成
+        （等不到即关 VM 报 warm_pool_sync_incomplete，绝不
+        把未同步的 VM 记成预热机），再 pause、写登记。
+        返回 {pooled, vm, mode, gitea_synced}。"""
+        if int(getattr(self, "warm_pool_size", 0)) <= 0:
+            return {"pooled": False, "vm": None,
+                    "reason": "warm_pool_disabled"}
+        vm = self.persistent_vm_name(tenant_id)
+        record = read_persistent_vm(self.state_dir, tenant_id)
+        if record is not None and record.get("state") == "paused" \
+                and record.get("vm") == vm \
+                and os.path.exists(self.api_socket({"id": vm})):
+            return {"pooled": True, "vm": vm,
+                    "mode": "already-warm",
+                    "gitea_synced": bool(
+                        record.get("gitea_synced"))}
+        vm_task, synced = self._prewarm_build_new(
+            tenant_id, extra_env, sync_workspace)
+        self.pause_vm(vm_task)
+        import time as _time
+        write_persistent_vm(self.state_dir, tenant_id, {
+            "vm": vm, "tenant_id": tenant_id,
+            "state": "paused", "origin": "prewarm",
+            "gitea_synced": synced, "updated_at": _time.time()})
+        self._emit("warm_pool_prewarm", vm_task,
+                   "vm=%s gitea_synced=%s" % (vm, synced))
+        return {"pooled": True, "vm": vm, "mode": "prewarmed",
+                "gitea_synced": synced}
+
+    def refill_warm_pool(self, tenant_id: str | None,
+                         extra_env: dict | None = None,
+                         sync_workspace: str | None = None) -> dict:
+        """后台补建回池（调用方自行起线程跑，本方法本身
+        同步）：池里已有 paused 登记即 no-op；池空即
+        prewarm 一台。warm_pool_size==0 时 no-op。"""
+        record = read_persistent_vm(self.state_dir, tenant_id)
+        if record is not None and record.get("state") == "paused" \
+                and os.path.exists(self.api_socket(
+                    {"id": record.get("vm") or ""})):
+            return {"pooled": True, "vm": record.get("vm"),
+                    "mode": "already-warm",
+                    "gitea_synced": bool(record.get("gitea_synced"))}
+        return self.prewarm_persistent_vm(
+            tenant_id, extra_env=extra_env,
+            sync_workspace=sync_workspace)
+
+    def _acquire_paused_candidate(self, tenant):
+        """认领的登记核对（acquire_for_task 的前半段）：登记
+        是 paused、socket 存在、tenant.json 归属一致才返回
+        (vm, record)；任一不符即清登记、返回 (None, None)
+        （调用方回落冷启）。归属不符与 socket 消失是登记
+        本身已坏，清掉；状态未知（vm.info 取不到）不在此
+        判——那是认领时的第二道门，见 acquire_for_task。"""
+        record = read_persistent_vm(self.state_dir, tenant)
+        if record is None or record.get("state") != "paused":
+            return None, None
+        vm = record.get("vm")
+        if not vm:
+            return None, None
+        owner = read_vm_tenant(self.state_dir, vm)
+        if owner is not None and owner != tenant:
+            # 盘的归属与请求方不一致：这台不是你的 VM，绝不
+            # resume，清登记回落冷启（legacy 未标记盘 owner
+            # 为 None 时放行，见 _check_snapshot_tenant 同款）。
+            clear_persistent_vm(self.state_dir, tenant)
+            return None, None
+        if not os.path.exists(self.api_socket({"id": vm})):
+            clear_persistent_vm(self.state_dir, tenant)
+            return None, None
+        return vm, record
+
+    def acquire_for_task(self, task: dict,
+                         extra_env: dict | None = None) -> dict:
+        """取本租户的可用 VM（任务到达时的认领）：
+        - 持久登记是 paused、socket 活着、vm.info 确认
+          Paused、tenant.json 归属一致 → resume + 等 SSH
+          就绪，返回 {mode: resumed, vm, resume_seconds,
+          warm_hit}（warm_hit 表示这台来自 prewarm，
+          不是上个任务暂停留下的）。
+        - 其它情形（无登记 / VM 已死 / 状态未知 /
+          resume 或 SSH 就绪失败）→ 丢弃该登记（若有），
+          返回 {mode: cold, vm: None}，调用方回落冷启，
+          不报错。warm_pool_size==0 时直接 cold。
+        resume_seconds 只计 resume+SSH 就绪，不含任务
+        自身启动。"""
+        if int(getattr(self, "warm_pool_size", 0)) <= 0:
+            return {"mode": "cold", "vm": None}
+        tenant = self._tenant_id(task)
+        vm, record = self._acquire_paused_candidate(tenant)
+        if vm is None:
+            return {"mode": "cold", "vm": None}
+        state = self._vm_info_state({"id": vm})
+        if state != "Paused":
+            # 状态未知或非 paused 时不猜：不是 paused 就不
+            # 认领（running 的 VM resume 会错、死了的 VM
+            # resume 也会错），回落冷启，登记留给清理。
+            return {"mode": "cold", "vm": None}
+        return self._acquire_resume_recorded(task, tenant, vm, record)
+
+    def _acquire_resume_recorded(self, task, tenant, vm, record):
+        """认领的 resume 段（acquire_for_task 的后半段）：
+        resume + 等 SSH 就绪，计时写回登记（state=running）
+        并返回 resumed；resume/就绪失败即清登记、回落 cold。
+        resume_seconds 只计 resume+SSH 就绪。"""
+        import time as _time
+        t0 = _time.time()
+        try:
+            self.resume_vm({"id": vm})
+            self.wait_guest_ready({"id": vm})
+        except Exception as exc:
+            self._emit("warm_pool_acquire_failed", task,
+                       "vm=%s resume/就绪失败，回落冷启：%s"
+                       % (vm, exc), "warm_pool_acquire_failed")
+            clear_persistent_vm(self.state_dir, tenant)
+            return {"mode": "cold", "vm": None}
+        resume_seconds = _time.time() - t0
+        write_persistent_vm(self.state_dir, tenant, {
+            "vm": vm, "tenant_id": tenant,
+            "state": "running",
+            "origin": record.get("origin") or "task",
+            "gitea_synced": bool(record.get("gitea_synced")),
+            "updated_at": _time.time()})
+        self._emit("warm_pool_acquire", task,
+                   "vm=%s mode=resumed resume_seconds=%.2f"
+                   " warm_hit=%s"
+                   % (vm, resume_seconds,
+                      record.get("origin") == "prewarm"))
+        return {"mode": "resumed", "vm": vm,
+                "resume_seconds": resume_seconds,
+                "warm_hit": record.get("origin") == "prewarm"}
+
+    def suspend_after_task(self, task_or_name: dict | str,
+                             tenant_id: str | None = None) -> bool:
+        """任务间隙暂停：pause + 写持久登记（state=paused，
+        origin=task）。pause 失败或 VM 已死即清登记并返回
+        False（调用方不得把死 VM 记成可用预热机）。"""
+        vm = self.vm_name(task_or_name)
+        tenant = tenant_id or self._tenant_id(
+            task_or_name if isinstance(task_or_name, dict)
+            else {"id": task_or_name})
+        try:
+            self.pause_vm(task_or_name)
+        except Exception as exc:
+            self._emit("vm_suspend_failed",
+                       task_or_name if isinstance(task_or_name, dict)
+                       else {"id": task_or_name},
+                       "vm=%s 暂停失败，清持久登记：%s" % (vm, exc),
+                       "vm_suspend_failed")
+            if tenant:
+                clear_persistent_vm(self.state_dir, tenant)
+            return False
+        if tenant:
+            import time as _time
+            record = read_persistent_vm(self.state_dir, tenant) or {}
+            write_persistent_vm(self.state_dir, tenant, {
+                "vm": vm, "tenant_id": tenant,
+                "state": "paused", "origin": "task",
+                "gitea_synced": bool(record.get("gitea_synced")),
+                "updated_at": _time.time()})
+        return True
+
+    def _run_guest_on_vm(self, task, vm, inner_args, goal,
+                         extra_env=None, **kwargs):
+        """在指定 VM 上跑客内命令并做中途死亡分类（
+        _run_with_persistent_vm 的执行段）：登记进程已死即
+        interrupted（断点可续跑），VM 活着时的异常/失败原
+        样处理，不误判。返回结果 dict（未附 vm_mode，调用
+        方贴）。"""
+        try:
+            if goal:
+                result = self._run_guest_goal(
+                    task, inner_args, extra_env=extra_env,
+                    vm_name=vm, **kwargs)
+            else:
+                result = self._run_guest(
+                    task, inner_args, extra_env=extra_env,
+                    vm_name=vm, **kwargs)
+        except Exception as exc:
+            if self._vm_proc_dead({"id": vm}):
+                return self._vm_death_result(
+                    task, goal, str(exc)[:500])
+            raise
+        if not goal and result.get("status") == "failed" \
+                and self._vm_proc_dead({"id": vm}):
+            result = dict(result)
+            result["status"] = "interrupted"
+            result["vm_dead"] = True
+            result["interrupted_reason"] = (
+                "VM 中途死亡，任务未完成、断点可续跑："
+                "客内命令已 failed 且登记的 VMM 进程已死")
+        return result
+
+    def _run_with_persistent_vm(self, task, inner_args, goal=False,
+                                extra_env=None, **kwargs):
+        """持久 VM 任务生命周期（阶段 1，与 _run_with_vm 对
+        照）：acquire（resumed）或冷启持久 VM → 跑客内命令
+        → finally 暂停回池（suspend_after_task），不关机、
+        不删盘。warm_pool_size==0 时本方法不被调（调用方
+        直接走 _run_with_vm）。冷启路径：wake 持久 VM
+        （合成任务 persist-<租户>），任务本身的 host 任务
+        名仍按真实任务 id 派生（_run_guest 的 vm_name 参数
+        分离 VM 身份与任务身份）。中途死亡分类见
+        _run_guest_on_vm（登记进程已死即 interrupted）。"""
+        tenant = self._tenant_id(task)
+        vm_task = self._persistent_task(tenant)
+        acquired = self.acquire_for_task(task, extra_env=extra_env)
+        vm = acquired.get("vm") or self.vm_name(vm_task)
+        mode = acquired["mode"]
+        resume_seconds = acquired.get("resume_seconds")
+        if mode == "cold":
+            self.wake_vm(vm_task, extra_env=extra_env)
+        try:
+            result = self._run_guest_on_vm(
+                task, vm, inner_args, goal,
+                extra_env=extra_env, **kwargs)
+            if isinstance(result, dict):
+                result = dict(result)
+                result["vm_mode"] = mode
+                if resume_seconds is not None:
+                    result["resume_seconds"] = resume_seconds
+            return result
+        finally:
+            try:
+                if self._vm_proc_dead({"id": vm}):
+                    # VM 已死：暂停无意义，清登记，免得下次
+                    # acquire 把死 VM 当可用预热机。
+                    if tenant:
+                        clear_persistent_vm(self.state_dir, tenant)
+                else:
+                    self.suspend_after_task({"id": vm},
+                                            tenant_id=tenant)
+            except Exception:
+                pass
+
+    def dispatch_task_persistent(self, task: dict,
+                                 extra_env: dict | None = None,
+                                 model_provider: str | None = None) -> dict:
+        """持久 VM 任务入口（阶段 1）：warm_pool_size>=1 时
+        走 _run_with_persistent_vm（resume/冷启持久 VM →
+        跑 → 暂停回池）；==0 时退化为 dispatch_task（冷启
+        → 跑 → 关机），与旧行为一致。结果 dict 附 vm_mode
+        （resumed/cold）与 resume_seconds（resumed 时）。"""
+        if not isinstance(task, dict) or task.get("id") is None:
+            raise ValueError("任务缺少 id")
+        if not task.get("prompt"):
+            raise ValueError("任务缺少 prompt")
+        if int(getattr(self, "warm_pool_size", 0)) <= 0:
+            return self.dispatch_task(
+                task, extra_env=extra_env,
+                model_provider=model_provider)
+        args = ["dispatch", "--name", self._host_name(task),
+                "--task", task["prompt"]]
+        repo = task.get("repo")
+        if repo:
+            args += ["--repo", repo]
+        return self._run_with_persistent_vm(
+            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 recycle_persistent_vm(self, task_or_name: dict | str,
+                              tenant_id: str | None = None) -> dict:
+        """显式回收持久 VM（唯一删用户盘的入口）：先 shutdown
+       （stop_vm，keep_disk=False），再删 overlay 与数据盘、
+        清 tenant 标记与持久登记。返回 {vm, state: recycled,
+        deleted_disks}。与 recycle_vm 的区别：后者只关 VM
+        不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。"""
+        vm = self.vm_name(task_or_name)
+        tenant = tenant_id or self._tenant_id(
+            task_or_name if isinstance(task_or_name, dict)
+            else {"id": task_or_name})
+        self.shutdown_vm(task_or_name)
+        deleted = []
+        if delete_task_disk(self.state_dir, vm):
+            deleted.append("task")
+        if delete_data_disk(self.state_dir, vm):
+            deleted.append("data")
+        _tenant_path = os.path.join(self.state_dir, vm,
+                                    "tenant.json")
+        if os.path.exists(_tenant_path):
+            os.remove(_tenant_path)
+        if tenant:
+            clear_persistent_vm(self.state_dir, tenant)
+        self._emit("vm_persistent_recycled",
+                   task_or_name if isinstance(task_or_name, dict)
+                   else {"id": task_or_name},
+                   "vm=%s deleted_disks=%s" % (vm, deleted))
+        return {"vm": vm, "state": "recycled",
+                "deleted_disks": deleted}
+
     def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str:
         """给（已暂停的）VM 做快照：ch-remote snapshot <裸 URL>。
         注意：snapshot 的参数是裸 URL，不是 source_url=...
@@ -1960,21 +2433,70 @@ class CloudHypervisorExecutor(Executor):
     def _record_snapshot_and_emit(self, task_or_name, vm_name,
                                   snapshot_url):
         """快照成功后登记 + 总量超帽时最旧优先剪枝（删动作
-        审计）+ vm_snapshot 审计，返回 snapshot_url。"""
+        审计）+ vm_snapshot 审计，返回 snapshot_url。
+        feat-chv-opt-landing：登记条目带 tenant 字段（归属
+        标记，见 disk.record_snapshot）；审计 detail 同样
+        带 tenant，便于事后核对谁的快照。"""
         task = task_or_name if isinstance(task_or_name, dict) \
             else {"id": task_or_name}
+        tenant = self._tenant_id(task)
         _entries, _removed = record_snapshot(
             self.state_dir, vm_name, snapshot_url,
-            getattr(self, "snapshot_max_total_gb", 20.0))
+            getattr(self, "snapshot_max_total_gb", 20.0),
+            tenant_id=tenant)
         for _old in _removed:
             self._emit("vm_snapshot_pruned", task,
                        "vm=%s pruned snapshot=%s（总量超帽，最旧优先）"
                        % (vm_name, _old.get("url")))
         self._emit("vm_snapshot", task,
-                   "vm=%s snapshot=%s"
-                   % (self.vm_name(task_or_name), snapshot_url))
+                   "vm=%s snapshot=%s tenant=%s"
+                   % (self.vm_name(task_or_name), snapshot_url, tenant))
         return snapshot_url
 
+    def _check_snapshot_tenant(self, task, snapshot_url) -> None:
+        """restore 的归属校验（feat-chv-opt-landing，阶段 1
+        必须补的代码项）：按 URL 在全部 VM 的快照登记里找
+        条目；条目带 tenant 且与请求方 tenant 不一致时，
+        直接拒绝（ChvFailure snapshot_tenant_mismatch）并
+        发 vm_restore_tenant_denied 审计——A 的快照绝不能
+        恢复到 B 名下。条目无 tenant 是阶段 1 之前的 legacy
+        登记：放行但发 vm_restore_tenant_unmarked 审计警告
+        （单租户灰度下的旧快照，不静默当作已验）；条目根本
+        未登记：同样放行 + 警告（未登记快照无从核对归属，
+        登记才是归属的唯一账本——生产须要求快照必登记，
+        见报告残留）。请求方无 tenant（None）而条目有
+        tenant 时同样拒绝（无法证明归属即 fail-closed）。"""
+        _owner_vm, entry = find_snapshot_entry(
+            self.state_dir, snapshot_url)
+        if entry is None:
+            self._emit("vm_restore_tenant_unmarked", task,
+                       "vm=%s snapshot=%s 未在登记中找到，归属"
+                       "无法核对（未登记快照，放行并警告）"
+                       % (self.vm_name(task), snapshot_url),
+                       "snapshot_tenant_unmarked")
+            return
+        marked = entry.get("tenant")
+        if marked is None:
+            self._emit("vm_restore_tenant_unmarked", task,
+                       "vm=%s snapshot=%s 登记无 tenant 字段"
+                       "（legacy 旧快照，放行并警告）"
+                       % (self.vm_name(task), snapshot_url),
+                       "snapshot_tenant_unmarked")
+            return
+        requester = self._tenant_id(task)
+        if requester != marked:
+            self._emit("vm_restore_tenant_denied", task,
+                       "vm=%s snapshot=%s 归属 tenant=%s，请求方"
+                       " tenant=%s，拒绝跨用户恢复"
+                       % (self.vm_name(task), snapshot_url,
+                          marked, requester),
+                       "snapshot_tenant_mismatch")
+            raise ChvFailure(
+                "snapshot_tenant_mismatch",
+                "快照归属 tenant=%s，与请求方 tenant=%s 不一致，"
+                "拒绝跨用户恢复：%s" % (marked, requester,
+                                       snapshot_url))
+
     def _validate_restore(self, task, snapshot_url):
         """restore 前置校验：返回（归一化后的 snapshot_url，
         归一化后的 net 条目）。
@@ -2002,6 +2524,7 @@ class CloudHypervisorExecutor(Executor):
             snapshot_url, self.state_dir,
             snapshot_dir=getattr(self, "snapshot_dir", None),
             allow_remote=getattr(self, "allow_remote_snapshot", False))
+        self._check_snapshot_tenant(task, snapshot_url)
         self._check_snapshot_disk(snapshot_url)
         self.preflight()
         self.resolve_ssh_host()
@@ -2010,11 +2533,16 @@ class CloudHypervisorExecutor(Executor):
                 "host_key_not_pinned",
                 "provisioned 模式下 restore 缺 known_hosts，拒绝 "
                 "keyscan 回落：VM %s" % self.vm_name(task))
-        net_entries = []
+        return snapshot_url, self._validate_restore_net(task)
+
+    def _validate_restore_net(self, task):
+        """restore 的 net 隔离段（_validate_restore 的尾段）：
+        与 wake 的 _wake_prechecks 同款，冲突在起 VMM 前
+        拒绝（net_tap_conflict）；net 未配即空列表。"""
         if not self.net_json:
-            return snapshot_url, net_entries
+            return []
         try:
-            net_entries = validate_net_entries(
+            return validate_net_entries(
                 parse_net_entries(self.net_json),
                 self._registered_net_entries(
                     exclude_vm=self.vm_name(task)))
@@ -2024,7 +2552,6 @@ class CloudHypervisorExecutor(Executor):
                        % (self.vm_name(task), exc.kind),
                        exc.kind)
             raise
-        return snapshot_url, net_entries
 
     def _check_snapshot_disk(self, snapshot_url):
         """restore 前的快照磁盘预检（feat-chv-v3，hibernate
@@ -2199,6 +2726,49 @@ class CloudHypervisorExecutor(Executor):
             if os.path.exists(_p):
                 os.remove(_p)
 
+    def _persistent_record_for_vm(self, name):
+        """该 VM 是否被某租户的持久登记认领：读 tenant.json
+        得租户，再读该租户的持久登记、核对 vm 名一致。
+        返回登记 dict 或 None（未认领：冷路径 VM 或旧 VM，
+        stop/GC 按旧语义处理）。"""
+        tenant = read_vm_tenant(self.state_dir, name)
+        if not tenant:
+            return None
+        record = read_persistent_vm(self.state_dir, tenant)
+        if record is not None and record.get("vm") == name:
+            return record
+        return None
+
+    def _overlay_disk_protected(self, name) -> bool:
+        """该 VM 的 overlay 盘是否受保护（stop/GC 不删）：
+        有持久登记认领，或有登记快照（hibernate 的 restore
+        输入）。copy 模式的 disk.raw 不受此保护（旧语义，
+        _cleanup_vm_files 只在 overlay 模式下查它——见下），
+        但本方法本身不分模式，调用方按模式决定是否查。
+        数据盘 data.qcow2 永远不受 GC/stop 删除（另有显式
+        回收入口），本方法不覆盖它。"""
+        if self._persistent_record_for_vm(name) is not None:
+            return True
+        return has_recorded_snapshots(self.state_dir, name)
+
+    def _cleanup_vm_task_disk(self, name, task, keep_disk) -> bool:
+        """stop 的任务盘删除块（_cleanup_vm_files 的中段）：
+        keep_disk 或 overlay 受保护盘（持久登记认领/有登记
+        快照）即跳过；删掉返回 False（无失败），删失败记
+        审计并返回 True（脏标记由调用方写）。"""
+        _protected = getattr(self, "disk_mode", "copy") == "overlay" \
+            and self._overlay_disk_protected(name)
+        try:
+            if not keep_disk and not _protected \
+                    and delete_task_disk(self.state_dir, name):
+                self._emit("vm_disk_deleted", task,
+                           "vm=%s disk deleted on stop" % name)
+        except Exception as _disk_exc:
+            self._emit("vm_disk_delete_failed", task,
+                       "vm=%s error=%s" % (name, _disk_exc))
+            return True
+        return False
+
     def _cleanup_vm_files(self, name, task, keep_disk=False):
         """stop 的三块文件清理：seed/私钥、任务磁盘、net 登记。
 
@@ -2209,7 +2779,12 @@ class CloudHypervisorExecutor(Executor):
         与 seed.iso，任一删了 restore 即失败——见
         hibernate_vm 的评估注释与 V-7/D2；seed 目录与宿主
         私钥仍删，config.json 不引用它们，私钥明文已嵌进
-        seed.iso、宿主侧不留）。"""
+        seed.iso、宿主侧不留）。
+        feat-chv-opt-landing：overlay 模式下，任务盘是持久
+        用户盘——该 VM 有持久登记认领或有登记快照时，stop
+        不删 overlay（删了等于销毁用户 VM / 毁掉快照输入）；
+        删盘只走显式回收（recycle_persistent_vm）。无认领
+        又无快照的 overlay VM（冷路径残留）仍按旧语义删。"""
         from cloud.chv.hostkeys import dirty_marker_path  # noqa: F401
         failed = False
         try:
@@ -2218,14 +2793,11 @@ class CloudHypervisorExecutor(Executor):
             failed = True
             self._emit("vm_key_cleanup_failed", task,
                        "vm=%s error=%s" % (name, _cleanup_exc))
-        try:
-            if not keep_disk and delete_task_disk(self.state_dir, name):
-                self._emit("vm_disk_deleted", task,
-                           "vm=%s disk deleted on stop" % name)
-        except Exception as _disk_exc:
+        # feat-chv-opt-landing：overlay 模式下受保护的盘（持久
+        # 登记认领或有登记快照）stop 不删；copy 模式一律旧语义。
+        # 见 _cleanup_vm_task_disk。
+        if self._cleanup_vm_task_disk(name, task, keep_disk):
             failed = True
-            self._emit("vm_disk_delete_failed", task,
-                       "vm=%s error=%s" % (name, _disk_exc))
         try:
             _net_path = os.path.join(self.state_dir, name, "net.json")
             if os.path.exists(_net_path):
@@ -2321,12 +2893,15 @@ class CloudHypervisorExecutor(Executor):
         return had
 
     def gc_stale_disks(self) -> dict:
-        """残留盘 GC（H3）：扫描 state_dir/*/disk.raw，VM
-        非 live（无 api-socket 文件且无登记活进程）即删盘
-        并审计；删失败审计 vm_disk_delete_failed 并计入
-        failed。返回 {deleted, failed}。
+        """残留盘 GC（H3）：扫描 state_dir/*/disk.raw 与
+        disk.qcow2，VM 非 live（无 api-socket 文件且无登记
+        活进程）即删盘并审计；删失败审计 vm_disk_delete_failed
+        并计入 failed。返回 {deleted, failed}。
         feat-chv-v3：有登记快照的 VM 不 GC 盘——那是休眠
-        VM 的盘，restore 还要读它（见 hibernate_vm）。"""
+        VM 的盘，restore 还要读它（见 hibernate_vm）。
+        feat-chv-opt-landing：有持久登记认领的 VM 同样不
+        GC（那是用户的持久盘）；数据盘 data.qcow2 永不进
+        GC（用户数据，只走显式回收 recycle_persistent_vm）。"""
         deleted: list = []
         failed: list = []
         try:
@@ -2336,11 +2911,12 @@ class CloudHypervisorExecutor(Executor):
         for entry in entries:
             if not entry.startswith("chv-"):
                 continue
-            disk = task_disk_path(self.state_dir, entry)
-            if not (os.path.islink(disk) or os.path.exists(disk)):
+            if not task_disk_paths(self.state_dir, entry):
                 continue
             if has_recorded_snapshots(self.state_dir, entry):
                 continue
+            if self._persistent_record_for_vm(entry) is not None:
+                continue
             live = os.path.exists(
                 os.path.join(self.state_dir, entry + ".sock"))
             with self._vms_lock:
@@ -2523,15 +3099,20 @@ class CloudHypervisorExecutor(Executor):
         return cleaned
 
     # -- 客内执行 ------------------------------------------------------
-    def _run_guest(self, task, inner_args, extra_env=None, **kwargs):
+    def _run_guest(self, task, inner_args, extra_env=None, vm_name=None, **kwargs):
         """SSH 调客内 host.py，返回 Local/Docker 同款结果 dict
         （executor 字段改写为 cloud-hypervisor，并附 vm 名与
         api-socket）。runner 异常即原样传播，由 worker 侧统一
         写回 failed——本类不在这里把异常吞成假成功。
-        extra_env（P1-1）：经 SSH env 前缀进客机环境。"""
+        extra_env（P1-1）：经 SSH env 前缀进客机环境。
+        feat-chv-opt-landing：vm_name 给出时 SSH 目标是该
+        VM（持久 VM 路径：任务 id ≠ VM 名，host 任务名仍按
+        任务 id 派生）；不给时目标即 vm_name(task)，与旧
+        行为一致。"""
         self._validate()
+        vm_task = {"id": vm_name} if vm_name else task
         ssh_cmd = self.build_ssh_command(
-            task, self._guest_host_command(*inner_args, **kwargs),
+            vm_task, self._guest_host_command(*inner_args, **kwargs),
             extra_env=extra_env)
         result = self._call_ssh_with_transient_retry(
             task, ssh_cmd, timeout=self.task_timeout)
@@ -2540,8 +3121,8 @@ class CloudHypervisorExecutor(Executor):
         parsed = _parse_host_result(task["id"],
                                     SimpleNamespaceShim(result))
         parsed["executor"] = "cloud-hypervisor"
-        parsed["vm"] = self.vm_name(task)
-        parsed["api_socket"] = self.api_socket(task)
+        parsed["vm"] = self.vm_name(vm_task)
+        parsed["api_socket"] = self.api_socket(vm_task)
         return parsed
 
     def _goal_result(self, task_id, result):
@@ -2562,10 +3143,11 @@ class CloudHypervisorExecutor(Executor):
                                  "cloud-hypervisor")
 
     def _run_guest_goal(self, task, inner_args, extra_env=None,
-                        **kwargs):
+                        vm_name=None, **kwargs):
         self._validate()
+        vm_task = {"id": vm_name} if vm_name else task
         ssh_cmd = self.build_ssh_command(
-            task, self._guest_host_command(*inner_args, **kwargs),
+            vm_task, self._guest_host_command(*inner_args, **kwargs),
             extra_env=extra_env)
         return self._goal_result(
             task["id"],