feat(chv): 阶段 1 数据盘与 tenant 标记同 overlay 解耦（copy 也供）

真机根因（2026-11-07，CHV v54 三变体实证）：vm.boot 对任何
带 backing file 的 qcow2 都报 Maximum disk nesting depth
exceeded——overlay-only 与 overlay-rawbase 均 500，base 直接
启动（reuse）22.5 秒成功，数据盘与压缩均已排除。overlay 在
本 VMM 上不可用，持久 VM 须能在 copy 模式下跑。
- _provision_task_disk：copy 与 overlay 都供数据盘并写
  tenant.json（只有 reuse 不供不写）；_provision_data_disk_path
  同理（reuse 才返回 None）。
- _cleanup_vm_task_disk：保护（持久登记/登记快照）不再只限
  overlay 模式；无登记无快照的 copy 盘仍按旧语义删除。
- provision_data_disk：数据盘改 raw 稀疏盘（os.truncate，
  零外部依赖）——copy 模式本来就不需要 qemu-img，数据盘
  绝不能悄悄加上这个依赖；CHV 按后缀认 Raw。
- 测试：新增 copy 供数据盘+tenant 与 copy 持久盘 stop 保护；
  VmConfig 三盘断言（任务+seed+数据）更新 arch95/fix 两处
  旧“恰好两盘”契约与本轨 qcow2 类型断言。
回归：test_chv_*.py、test_cloud_hypervisor.py 42、test_chv_v3_fixes
38 全过（全量 cloud/tests 在跑）。
不推、不合、不部署；只本地分支。

cloud/chv/disk.py

@@ -17,7 +17,7 @@ check_disk_quota 改为三项：state_dir 已用总量 + 本次
 
 feat-chv-opt-landing（阶段 1）：黄金镜像只读 base +
 qcow2 overlay 分层（disk_mode="overlay"）、数据盘独立
-持久（data.qcow2，绝不进快照/GC）、快照与 overlay 的
+持久（data.raw，绝不进快照/GC）、快照与 overlay 的
 tenant 归属标记（登记条目 tenant 字段 + 每 VM tenant.json
 + 持久 VM 登记 persistent/<tenant>.json）。overlay 与
 数据盘只在 overlay 模式下供给；copy/reuse 语义不变。
@@ -149,7 +149,7 @@ def task_disk_path(state_dir: str, vm_name: str) -> str:
 def task_disk_paths(state_dir: str, vm_name: str) -> list:
     """该 VM 现存的任务磁盘文件（copy 的 disk.raw 与/或
     overlay 的 disk.qcow2）：GC 与清理按此清单查，不猜
-    文件名；数据盘 data.qcow2 不在内——用户数据盘永不
+    文件名；数据盘 data.raw 不在内——用户数据盘永不
     进任务盘 GC，见 delete_data_disk。"""
     out = []
     for name in ("disk.raw", OVERLAY_DISK_NAME):
@@ -166,7 +166,7 @@ def delete_task_disk(state_dir: str, vm_name: str) -> bool:
     绝不静默——残留盘由 gc_stale_disks 下次再清）。
     符号链接只删链接本身，不跟随。
     feat-chv-opt-landing：disk.raw 与 disk.qcow2 都删
-    （任一存在即删）；数据盘 data.qcow2 绝不在此删——
+    （任一存在即删）；数据盘 data.raw 绝不在此删——
     它是用户数据，删它须走 delete_data_disk（显式回收）。"""
     removed = False
     for path in task_disk_paths(state_dir, vm_name):
@@ -178,14 +178,14 @@ def delete_task_disk(state_dir: str, vm_name: str) -> bool:
 # -- overlay 分层与数据盘（feat-chv-opt-landing，阶段 1）------
 # 黄金镜像只读 base（qcow2，基线 773 MiB）绝不被写：每台
 # 用户 VM 只有一个空 overlay（disk.qcow2，backing file 指
-# 向 base）；数据盘（data.qcow2，工作区 + 浏览器 profile
+# 向 base）；数据盘（data.raw，工作区 + 浏览器 profile
 # 的载体）是第二块独立盘——快照登记与 GC 只认任务盘，
 # 数据盘不进快照、不进 GC、不随 stop 删（删它须显式回收）。
 # 客内挂载见镜像 99-chv-sync.cfg runcmd 的 chv-data-mount
 #（按 LABEL=CHVDATA 找盘，不依赖 /dev/vd? 固定名）。
 
 OVERLAY_DISK_NAME = "disk.qcow2"
-DATA_DISK_NAME = "data.qcow2"
+DATA_DISK_NAME = "data.raw"
 DATA_DISK_GB = 10.0
 DATA_DISK_LABEL = "CHVDATA"
 TENANT_FILE_NAME = "tenant.json"
@@ -198,7 +198,7 @@ def overlay_disk_path(state_dir: str, vm_name: str) -> str:
 
 
 def data_disk_path(state_dir: str, vm_name: str) -> str:
-    """该 VM 的数据盘路径：state_dir/<vm>/data.qcow2。"""
+    """该 VM 的数据盘路径：state_dir/<vm>/data.raw。"""
     return os.path.join(state_dir, vm_name, DATA_DISK_NAME)
 
 
@@ -308,30 +308,31 @@ def create_overlay_disk(base_path: str, overlay_path: str,
 
 
 def provision_data_disk(state_dir: str, vm_name: str,
-                        size_gb: float = DATA_DISK_GB,
-                        runner: Callable[..., object] | None = None) -> str:
-    """建数据盘（空 qcow2 稀疏盘，虚大小默认 10 GB，实际
-    占用随写入增长）：已存在即复用（不 resize、不重建，
-    旧盘里可能已有用户数据）；不存在即 qemu-img create。
+                        size_gb: float = DATA_DISK_GB) -> str:
+    """建数据盘（空 raw 稀疏盘，虚大小默认 10 GB，实际占用
+    随写入增长）：已存在即复用（不 resize、不重建，旧盘里
+    可能已有用户数据）；不存在即 os.truncate 建稀疏文件。
+    raw 而不用 qcow2（2026-11-07）：数据盘只是空白可写卷，
+    qcow2 在此无收益；raw 零外部依赖——copy 模式本来就不
+    需要 qemu-img，数据盘绝不能悄悄给 copy 加上这个依赖
+    （Muse VM 无 qemu-img 时 copy 派发会全灭；CHV v54 又
+    拒绝带 backing 的 qcow2，数据盘走 qcow2 毫无必要）。
     失败即 ChvFailure disk_overlay_invalid（数据盘建不出
-    来时 overlay 模式拒绝启动，不让任务在无数据盘的
-    VM 里把工作区写进系统盘——静默回落等于丢“独立持久”
-    的保证）。"""
-    run = runner if runner is not None else _default_qemu_runner
-    qemu = "qemu-img" if runner is not None else _qemu_img_cmd()
+    来时拒绝启动，不让任务在无数据盘的 VM 里把工作区写进
+    系统盘——静默回落等于丢“独立持久”的保证）。"""
     path = data_disk_path(state_dir, vm_name)
     os.makedirs(os.path.dirname(path), exist_ok=True)
     if os.path.islink(path):
         os.remove(path)
     if os.path.exists(path):
         return path
-    result = run([qemu, "create", "-f", "qcow2",
-                  path, "%gG" % size_gb])
-    if getattr(result, "returncode", 1) != 0:
+    try:
+        with open(path, "wb") as fh:
+            fh.truncate(int(size_gb * 1024 ** 3))
+    except OSError as exc:
         raise ChvFailure(
             "disk_overlay_invalid",
-            "数据盘建盘失败：%s：%s"
-            % (path, getattr(result, "stderr", "") or ""))
+            "数据盘建盘失败：%s：%s" % (path, exc))
     if not os.path.exists(path):
         raise ChvFailure(
             "disk_overlay_invalid",

cloud/chv/executor.py

@@ -1397,25 +1397,32 @@ class CloudHypervisorExecutor(Executor):
           放行，见 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，标给
-          某租户是错的）。
+          file 指向只读 base。
+        数据盘与 tenant 标记与 overlay 解耦（2026-11-07 真机
+        实测：CHV v54 的 vm.boot 对任何带 backing file 的
+        qcow2 都报 Maximum disk nesting depth exceeded，
+        overlay 在本 VMM 上不可用——见阶段 1 报告 §overlay；
+        但数据盘（工作区/浏览器 profile 载体，不进快照）与
+        tenant.json（持久 VM 归属）是独立持久/归属机制，
+        不依赖 backing file）：copy 与 overlay 都供给独立
+        数据盘 state_dir/<vm>/data.raw 并写 tenant 归属
+        标记 tenant.json；只有 reuse 不写（盘就是共享 base，
+        标给某租户是错的）。
         返回每任务磁盘路径。"""
         disk = provision_task_disk(
             self.state_dir, self.vm_name(task), self.rootfs,
             self.disk_mode)
-        if getattr(self, "disk_mode", "copy") == "overlay":
+        if getattr(self, "disk_mode", "copy") in ("overlay", "copy"):
             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":
+        """数据盘路径（VmConfig 第三块盘）：copy 与 overlay 都
+        挂（见 _provision_task_disk 的解耦说明）；只有 reuse
+        返回 None（共享 base，不挂私有数据盘）。"""
+        if getattr(self, "disk_mode", "copy") == "reuse":
             return None
         return data_disk_path(self.state_dir, self.vm_name(task))
 
@@ -2740,12 +2747,11 @@ class CloudHypervisorExecutor(Executor):
         return None
 
     def _overlay_disk_protected(self, name) -> bool:
-        """该 VM 的 overlay 盘是否受保护（stop/GC 不删）：
+        """该 VM 的任务盘是否受保护（stop/GC 不删）：
         有持久登记认领，或有登记快照（hibernate 的 restore
-        输入）。copy 模式的 disk.raw 不受此保护（旧语义，
-        _cleanup_vm_files 只在 overlay 模式下查它——见下），
-        但本方法本身不分模式，调用方按模式决定是否查。
-        数据盘 data.qcow2 永远不受 GC/stop 删除（另有显式
+        输入）。不分模式（2026-11-07：copy 模式也要能跑持
+        久 VM；无登记/无快照的盘仍按旧语义删除）。
+        数据盘 data.raw 永远不受 GC/stop 删除（另有显式
         回收入口），本方法不覆盖它。"""
         if self._persistent_record_for_vm(name) is not None:
             return True
@@ -2753,11 +2759,12 @@ class CloudHypervisorExecutor(Executor):
 
     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)
+        keep_disk 或受保护盘（持久登记认领/有登记快照）即
+        跳过；删掉返回 False（无失败），删失败记审计并返回
+        True（脏标记由调用方写）。保护不分模式（2026-11-07：
+        copy 模式也要能跑持久 VM；无登记/无快照的 copy 盘
+        仍按旧语义删除，只有被认领/被快照引用的盘才留）。"""
+        _protected = self._overlay_disk_protected(name)
         try:
             if not keep_disk and not _protected \
                     and delete_task_disk(self.state_dir, name):
@@ -2900,7 +2907,7 @@ class CloudHypervisorExecutor(Executor):
         feat-chv-v3：有登记快照的 VM 不 GC 盘——那是休眠
         VM 的盘，restore 还要读它（见 hibernate_vm）。
         feat-chv-opt-landing：有持久登记认领的 VM 同样不
-        GC（那是用户的持久盘）；数据盘 data.qcow2 永不进
+        GC（那是用户的持久盘）；数据盘 data.raw 永不进
         GC（用户数据，只走显式回收 recycle_persistent_vm）。"""
         deleted: list = []
         failed: list = []

cloud/tests/test_chv_arch95.py

@@ -102,15 +102,20 @@ class Arch95Case(unittest.TestCase):
         return FakeResult(0, "", "")
 
     def _strict_api(self, socket_path, method, path, payload=None):
-        """严格替身：vm.create 必须带任务盘 + seed 盘、net 无重复。"""
+        """严格替身：vm.create 必须带任务盘 + seed 盘 + 数据盘、
+        net 无重复（阶段 1：数据盘是第三块盘，见
+        _provision_task_disk 的解耦说明）。"""
         if path == "/vm.create":
             self.api_payloads.append(payload)
             disks = [d.get("path") for d in payload.get("disks", [])]
-            self.assertEqual(len(disks), 2, "VmConfig 必须恰好两盘")
+            self.assertEqual(len(disks), 3,
+                             "VmConfig 必须恰好三盘（任务+seed+数据）")
             self.assertTrue(disks[0].endswith("disk.raw"),
                             "第一盘必须是任务盘")
             self.assertTrue(disks[1].endswith("seed.iso"),
                             "第二盘必须是 seed 盘，缺 seed 即假保证")
+            self.assertTrue(disks[2].endswith("data.raw"),
+                            "第三盘必须是数据盘")
             self.assertTrue(payload.get("payload", {}).get("kernel"))
             nets = payload.get("net", [])
             taps = [n.get("tap") for n in nets]

cloud/tests/test_chv_fix.py

@@ -176,11 +176,15 @@ class SeedDiskTest(ChvFixCase):
         out = ex.wake_vm({"id": "seed1"})
         self.assertEqual(out["state"], "running")
         payload = self.api_payloads["vm.create"]
-        self.assertEqual(len(payload["disks"]), 2)
+        self.assertEqual(len(payload["disks"]), 3)
         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))
+        # 第三盘是数据盘（阶段 1：copy 模式也挂数据盘）。
+        self.assertTrue(payload["disks"][2]["path"].endswith(
+            "data.raw"))
+        self.assertEqual(payload["disks"][2]["image_type"], "Raw")
         self.assertIn("genisoimage", [c[0] for c in self.calls])
 
     def test_seed_disk_build_failure_is_fail_fast(self):

cloud/tests/test_chv_opt_stage1.py

@@ -268,19 +268,19 @@ class OverlayDiskTest(unittest.TestCase):
         self.assertEqual(disk,
                          os.path.join(self.state, "chv-a", "disk.qcow2"))
         self.assertTrue(os.path.exists(disk))
-        data = provision_data_disk(self.state, "chv-a", runner=runner)
+        data = provision_data_disk(self.state, "chv-a")
         self.assertEqual(data,
-                         os.path.join(self.state, "chv-a", "data.qcow2"))
+                         os.path.join(self.state, "chv-a", "data.raw"))
         self.assertTrue(os.path.exists(data))
         # 数据盘已存在即复用，不重建。
-        data2 = provision_data_disk(self.state, "chv-a", runner=runner)
+        data2 = provision_data_disk(self.state, "chv-a")
         self.assertEqual(data2, data)
 
     def test_delete_task_disk_removes_both_but_not_data(self):
         runner = _fake_qemu_runner_factory({})
         provision_task_disk(self.state, "chv-a", self.base,
                             "overlay", runner=runner)
-        provision_data_disk(self.state, "chv-a", runner=runner)
+        provision_data_disk(self.state, "chv-a")
         raw = os.path.join(self.state, "chv-a", "disk.raw")
         with open(raw, "wb") as fh:
             fh.write(b"R")
@@ -290,7 +290,7 @@ class OverlayDiskTest(unittest.TestCase):
         self.assertFalse(os.path.exists(raw))
         # 数据盘还在：删任务盘绝不删用户数据。
         self.assertTrue(os.path.exists(
-            os.path.join(self.state, "chv-a", "data.qcow2")))
+            os.path.join(self.state, "chv-a", "data.raw")))
         self.assertTrue(delete_data_disk(self.state, "chv-a"))
         self.assertFalse(delete_task_disk(self.state, "chv-a"))
 
@@ -324,9 +324,10 @@ class VmConfigOverlayTest(ChvOpt1Case):
             os.path.join(self.state_dir, "chv-a", "disk.qcow2"),
             seed_disk=None,
             data_disk=os.path.join(self.state_dir, "chv-a",
-                                   "data.qcow2"))
+                                   "data.raw"))
         self.assertEqual(cfg["disks"][0]["image_type"], "Qcow2")
-        self.assertEqual(cfg["disks"][1]["image_type"], "Qcow2")
+        # 数据盘是 raw 稀疏盘（2026-11-07：不用 qcow2，零外部依赖）。
+        self.assertEqual(cfg["disks"][1]["image_type"], "Raw")
         self.assertEqual(len(cfg["disks"]), 2)
 
     def test_raw_disks_stay_raw(self):
@@ -466,22 +467,10 @@ class PersistentPoolTest(ChvOpt1Case):
                 fh.write(b"QFI\xfb fake")
             return path
 
-        def fake_data(state_dir, vm_name, size_gb=10.0, runner=None):
-            path = os.path.join(state_dir, vm_name, "data.qcow2")
-            os.makedirs(os.path.dirname(path), exist_ok=True)
-            if not os.path.exists(path):
-                with open(path, "wb") as fh:
-                    fh.write(b"QFI\xfb fake-data")
-            return path
-
         p1 = mock.patch("cloud.chv.executor.provision_task_disk",
                         side_effect=fake_provision)
-        p2 = mock.patch("cloud.chv.executor.provision_data_disk",
-                        side_effect=fake_data)
         p1.start()
-        p2.start()
         self.addCleanup(p1.stop)
-        self.addCleanup(p2.stop)
         return ex
 
     def _socket(self, vm):
@@ -598,7 +587,7 @@ class PersistentPoolTest(ChvOpt1Case):
         self.assertTrue(os.path.exists(os.path.join(
             self.state_dir, "chv-persist-proj1", "disk.qcow2")))
         self.assertTrue(os.path.exists(os.path.join(
-            self.state_dir, "chv-persist-proj1", "data.qcow2")))
+            self.state_dir, "chv-persist-proj1", "data.raw")))
 
     def test_dispatch_persistent_cold_when_pool_empty(self):
         ex = self.make_exec(
@@ -638,7 +627,49 @@ class PersistentPoolTest(ChvOpt1Case):
         self.assertIn("data", out["deleted_disks"])
         self.assertFalse(os.path.exists(overlay))
         self.assertFalse(os.path.exists(os.path.join(
-            self.state_dir, "chv-persist-proj1", "data.qcow2")))
+            self.state_dir, "chv-persist-proj1", "data.raw")))
+        self.assertIsNone(read_persistent_vm(self.state_dir, "proj1"))
+
+    def test_copy_mode_provisions_data_disk_and_tenant(self):
+        # 数据盘与 tenant 标记与 overlay 解耦（2026-11-07：CHV v54
+        # 拒绝一切带 backing file 的 qcow2，持久 VM 在 copy 模式
+        # 下也要有数据盘与归属标记）。
+        ex = self.make_exec(env_extra={"CHV_DISK_MODE": "copy"})
+        self._patch_overlay_provision(ex)
+        disk = ex._provision_task_disk(self._task())
+        self.assertEqual(disk, os.path.join(
+            self.state_dir, "chv-abc123", "disk.raw"))
+        self.assertTrue(os.path.exists(os.path.join(
+            self.state_dir, "chv-abc123", "data.raw")))
+        self.assertEqual(ex._provision_data_disk_path(self._task()),
+                         os.path.join(self.state_dir, "chv-abc123",
+                                      "data.raw"))
+        from cloud.chv.disk import read_vm_tenant
+        self.assertEqual(read_vm_tenant(self.state_dir, "chv-abc123"),
+                         "proj1")
+
+    def test_stop_keeps_protected_copy_disk(self):
+        # 保护不分模式：copy 持久盘 stop/GC 同样不删，显式回收才删。
+        ex = self.make_exec(
+            env_extra={"CHV_WARM_POOL_SIZE": "1",
+                       "CHV_DISK_MODE": "copy"})
+        self._patch_overlay_provision(ex)
+        ex.prewarm_persistent_vm("proj1")
+        self._socket("chv-persist-proj1")
+        raw = os.path.join(self.state_dir, "chv-persist-proj1",
+                           "disk.raw")
+        self.assertTrue(os.path.exists(raw))
+        self.assertTrue(ex.stop_vm({"id": "chv-persist-proj1"}))
+        self.assertTrue(os.path.exists(raw))
+        self.assertEqual(ex.gc_stale_disks(), {"deleted": [],
+                                               "failed": []})
+        self.assertTrue(os.path.exists(raw))
+        out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"},
+                                       tenant_id="proj1")
+        self.assertEqual(out["state"], "recycled")
+        self.assertIn("task", out["deleted_disks"])
+        self.assertIn("data", out["deleted_disks"])
+        self.assertFalse(os.path.exists(raw))
         self.assertIsNone(read_persistent_vm(self.state_dir, "proj1"))
 
     def test_refill_warm_pool_noop_when_warm(self):