fix(chv): V-3/F1/F2 真机缺陷修复 + V-1/V-4 附带修复

V-3 快照/restore：
- snapshot 改传裸 URL（source_url= 是 restore 的键，服务端
  Could not extract path from URL）；快照产物是目录，校验
  改为目录非空，登记计量/剪枝同步支持目录
- restore 起 VMM 前补 _restore_prepare：杀旧 VMM（登记的 +
  未登记跨实例扫描）并清旧 socket，与 _wake_prepare 对等
- hibernate 评估结论：快照 config.json 引用原盘，旧 stop 删
  盘致 restore 必失败；hibernate 的 stop 改 keep_disk=True，
  GC 不删有登记快照的 VM 的盘，restore 前预检快照引用的盘
  （snapshot_disk_missing fail-fast）

F1：re-wake 删旧盘——H5 删私钥致 re-wake 必然生成新 key 对，
旧盘复用时 cloud-init 不重应用新 seed，sshd 仍呈旧 key 而
known_hosts 钉新 key；wake 本就是冷启动，旧盘不是要保留的状态

F2：stop_vm 跨实例按 socket 进程扫描杀未登记 VMM
（/proc cmdline 精确 socket + VMM 二进制名 + 跳过自身/pid 1，
绝不误杀 ch-remote 与他人 VMM）

V-4：remote_command source 失败改 || exit 126 fail-fast；
guest-env 条目 owner 改归 SSH 用户（0600 不变），root 仍 root:root

V-1 附带：build.sh 预建 /app/work（1777 粘滞位 + stat 断言）；
host.py default_codex_bin() 改 shutil.which，裸名不再让
health 报 ok:false

测试：新 test_chv_v3_fixes.py 23 项；旧快照测试按目录产物改写；
test_chv* 458 全过，全量 cloud 套件 1930 全过（skipped=8）

cloud/tests/test_chv_closeup.py

@@ -392,16 +392,24 @@ class RestoreSnapshotTest(CloseupCase):
         self.assertEqual(self.launches, [])
 
     def test_snapshot_not_persisted_missing_and_empty(self):
+        # feat-chv-v3：v54 快照产物是目录，校验改为“目录存在
+        # 且非空”——文件目标/空目录/缺失一律 snapshot_not_persisted。
         ex = self.make_exec()
-        missing = "file://" + os.path.join(self.state_dir, "nope.raw")
+        missing = "file://" + os.path.join(self.state_dir, "nope.snap")
         os.makedirs(self.state_dir, exist_ok=True)
         with self.assertRaises(ChvFailure) as ctx:
             ex.snapshot_vm(self._task(), missing)
         self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
-        empty_path = os.path.join(self.state_dir, "empty.raw")
-        open(empty_path, "wb").close()
+        empty_dir = os.path.join(self.state_dir, "empty.snap")
+        os.makedirs(empty_dir, exist_ok=True)
         with self.assertRaises(ChvFailure) as ctx:
-            ex.snapshot_vm(self._task(), "file://" + empty_path)
+            ex.snapshot_vm(self._task(), "file://" + empty_dir)
+        self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
+        file_target = os.path.join(self.state_dir, "file.snap")
+        with open(file_target, "wb") as fh:
+            fh.write(b"S" * 60)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.snapshot_vm(self._task(), "file://" + file_target)
         self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
 
     def test_snapshot_prunes_oldest_on_total_cap(self):
@@ -416,15 +424,18 @@ class RestoreSnapshotTest(CloseupCase):
             json.dump([{"url": "file://" + old_path, "path": old_path,
                         "size": 100, "created_at": time.time()}], fh)
         # 上限 150 字节：预检时旧 100 字节通过，新 60 字节记入后
-        # 总量 160 超帽，最旧一条被剪枝并审计。
+        # 总量 160 超帽，最旧一条被剪枝并审计。feat-chv-v3：
+        # 新快照是目录（config.json + state.json 各 30 字节）。
         cap_gb = 150 / (1024 ** 3)
         ex = self.make_exec(env_extra={
             "CHV_SNAPSHOT_MAX_TOTAL_GB": repr(cap_gb)})
-        new_path = os.path.join(self.state_dir, "new.raw")
-        with open(new_path, "wb") as fh:
-            fh.write(b"N" * 60)
-        out = ex.snapshot_vm(self._task(), "file://" + new_path)
-        self.assertEqual(out, "file://" + new_path)
+        new_dir = os.path.join(self.state_dir, "new.snap")
+        os.makedirs(new_dir, exist_ok=True)
+        for name in ("config.json", "state.json"):
+            with open(os.path.join(new_dir, name), "wb") as fh:
+                fh.write(b"N" * 30)
+        out = ex.snapshot_vm(self._task(), "file://" + new_dir)
+        self.assertEqual(out, "file://" + new_dir)
         self.assertFalse(os.path.exists(old_path))
         self.assertIn("vm_snapshot_pruned", self._audit_events())
 

cloud/tests/test_chv_secfix.py

@@ -501,7 +501,13 @@ class SnapshotLimitTest(SecfixCase):
         os.makedirs(vm_dir, exist_ok=True)
         with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh:
             _json.dump(entries, fh)
-        new_url = self._snap_file("new.snap")
+        # feat-chv-v3：新快照是目录（v54 产物），旧登记仍是
+        # 文件——剪枝须两种产物都能删。
+        new_dir = os.path.join(self.state_dir, "new.snap")
+        os.makedirs(new_dir, exist_ok=True)
+        with open(os.path.join(new_dir, "config.json"), "wb") as fh:
+            fh.write(b"N" * 64)
+        new_url = "file://" + new_dir
         ex.snapshot_vm({"id": "r"}, new_url)
         self.assertFalse(os.path.exists(old_url[len("file://"):]))
         self.assertTrue(os.path.exists(new_url[len("file://"):]))

cloud/tests/test_chv_v3_fixes.py

@@ -0,0 +1,549 @@
+"""CHV 第三棒（feat-chv-v3）回归测试：WSL 真机 V-1～V-4/F1/F2
+实测缺陷的代码层回归。
+
+- V-3① snapshot 传裸 URL（source_url= 是 restore 的键）；
+- V-3② 快照产物是目录（v54），校验改为目录非空；
+- V-3③ restore 起 VMM 前清旧 VMM + 旧 socket；
+- V-3 附带：hibernate 留盘（快照 config.json 引用原盘），
+  GC 不删有登记快照的 VM 的盘，restore 前预检快照引用的盘；
+- F1：re-wake 删旧盘（H5 删私钥致新 key 对 + 旧盘 cloud-init
+  不重应用 = host_key_mismatch）；
+- F2：跨实例 stop 按 socket 进程扫描清未登记 VMM；
+- V-4：remote_command source 失败 fail-fast（exit 126），
+  guest-env 归 SSH 用户所有；
+- V-1 附带：build.sh 预建 /app/work（1777 粘滞位 + 断言）；
+- V-1 附带：host.py default_codex_bin() 走 PATH 查找。
+
+全部用假 runner/launcher/api_client + 真实 /proc 扫描（F2
+的被杀进程是真实 spawn 的 sleep 进程，argv[0] 伪装成
+cloud-hypervisor），绝不启动真实 VM。
+"""
+
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.hostkeys import write_seed
+    from cloud.chv.ssh_transport import remote_command
+    from cloud.sandbox import ChvFailure
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.hostkeys import write_seed
+    from cloud.chv.ssh_transport import remote_command
+    from cloud.sandbox import ChvFailure
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+class FakeResult:
+    def __init__(self, returncode=0, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class FakeProc:
+    def __init__(self, alive=True):
+        self._alive = alive
+        self.terminated = False
+        self.killed = False
+
+    def poll(self):
+        return None if self._alive else 1
+
+    def terminate(self):
+        self.terminated = True
+        self._alive = False
+
+    def wait(self, timeout=None):
+        return 0
+
+    def kill(self):
+        self.killed = True
+        self._alive = False
+
+
+def _write(path, content=b"x"):
+    with open(path, "wb") as fh:
+        fh.write(content)
+    os.chmod(path, 0o755)
+    return path
+
+
+def _load_host_py():
+    spec = importlib.util.spec_from_file_location(
+        "chv_host_py", str(ROOT / "host.py"))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class V3Case(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = self.tmp.name
+        self.binary = _write(os.path.join(self.dir, "cloud-hypervisor"))
+        self.ch_remote = _write(os.path.join(self.dir, "ch-remote"))
+        self.kernel = _write(os.path.join(self.dir, "vmlinux"))
+        self.rootfs = _write(os.path.join(self.dir, "disk.raw"),
+                             b"R" * 100)
+        self.ssh_key = _write(os.path.join(self.dir, "id_chv"))
+        self.kvm = _write(os.path.join(self.dir, "kvm"))
+        self.cpuinfo = os.path.join(self.dir, "cpuinfo")
+        with open(self.cpuinfo, "w") as fh:
+            fh.write("flags : fpu vmx sse\n")
+        self.state_dir = os.path.join(self.dir, "state")
+        self.calls = []
+        self.api_calls = []
+        self.launches = []
+        self.audit = []
+        self.proc = None
+        self.spawned = []
+
+    def tearDown(self):
+        for proc in self.spawned:
+            try:
+                if proc.poll() is None:
+                    proc.terminate()
+                    proc.wait(timeout=5)
+            except Exception:
+                pass
+        self.tmp.cleanup()
+
+    def make_exec(self, runner=None, api_client=None, launcher=None,
+                  env_extra=None, **overrides):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_HOST_KEY_MODE": "keyscan",
+               "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=launcher or self._default_launcher,
+            api_client=api_client or 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
+
+    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, "", "")
+            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, payload))
+        return (200, {})
+
+    def _default_launcher(self, cmd):
+        self.launches.append(list(cmd))
+        self.proc = FakeProc(alive=True)
+        return self.proc
+
+    def _task(self, task_id="abc123"):
+        return {"id": task_id, "prompt": "做一个加法函数",
+                "project_id": "proj1"}
+
+    def _snap_dir(self, name="snap1", files=None):
+        """建一个 v54 形态的快照目录，返回 (url, path)。"""
+        path = os.path.join(self.state_dir, name)
+        os.makedirs(path, exist_ok=True)
+        for fname, content in (files or {
+                "config.json": b"{}",
+                "state.json": b"{}",
+                "memory-ranges": b"m"}).items():
+            with open(os.path.join(path, fname), "wb") as fh:
+                fh.write(content)
+        return "file://" + path, path
+
+    def _spawn_fake_vmm(self, sock_path, argv0="cloud-hypervisor"):
+        """真实 spawn 一个长眠进程，argv[0] 伪装成 VMM，
+        命令行带 --api-socket <sock>（F2/V-3③ 的扫描目标）。"""
+        proc = subprocess.Popen(
+            [argv0, "-c", "import time; time.sleep(30)",
+             "--api-socket", sock_path],
+            executable=sys.executable,
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        self.spawned.append(proc)
+        return proc
+
+
+# -- V-3①/② 快照：裸 URL + 目录校验 ---------------------------------------
+class SnapshotV3Test(V3Case):
+    def test_snapshot_passes_bare_url(self):
+        url, _ = self._snap_dir()
+        ex = self.make_exec()
+        out = ex.snapshot_vm(self._task(), url)
+        self.assertEqual(out, url)
+        snap_calls = [c for c in self.calls
+                      if c[0] == self.ch_remote and "snapshot" in c]
+        self.assertEqual(len(snap_calls), 1)
+        self.assertEqual(snap_calls[0][-1], url)
+        self.assertFalse(any(a.startswith("source_url")
+                             for a in snap_calls[0]))
+
+    def test_restore_still_passes_source_url(self):
+        url, _ = self._snap_dir()
+        ex = self.make_exec()
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+        restore_calls = [c for c in self.calls
+                         if c[0] == self.ch_remote and "restore" in c]
+        self.assertEqual(len(restore_calls), 1)
+        self.assertEqual(restore_calls[0][-1], "source_url=%s" % url)
+
+    def test_snapshot_dir_records_dir_total_size(self):
+        url, path = self._snap_dir(files={
+            "config.json": b"C" * 30, "state.json": b"S" * 30})
+        ex = self.make_exec()
+        ex.snapshot_vm(self._task(), url)
+        with open(os.path.join(self.state_dir, "chv-abc123",
+                               "snapshots.json")) as fh:
+            entries = json.load(fh)
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["size"], 60)
+        self.assertEqual(entries[0]["path"], path)
+
+
+# -- V-3③ restore 准备：清旧 VMM + 旧 socket ------------------------------
+class RestorePrepareV3Test(V3Case):
+    def test_restore_kills_registered_old_vmm_and_socket(self):
+        url, _ = self._snap_dir()
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        old_proc = self.proc
+        self.assertTrue(old_proc.poll() is None)
+        sock = os.path.join(self.state_dir, "chv-abc123.sock")
+        with open(sock, "w") as fh:
+            fh.write("")
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+        self.assertTrue(old_proc.terminated)
+        # 新 VMM 已起（第 2 次 launch），旧 socket 文件在
+        # launch 前已被清掉（fake launcher 不重建 socket 文件）。
+        self.assertEqual(len(self.launches), 2)
+        self.assertFalse(os.path.exists(sock))
+
+    def test_restore_kills_unregistered_vmm_by_socket(self):
+        url, _ = self._snap_dir()
+        ex = self.make_exec()
+        sock = os.path.join(self.state_dir, "chv-abc123.sock")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(sock, "w") as fh:
+            fh.write("")
+        victim = self._spawn_fake_vmm(sock)
+        self.assertIsNone(victim.poll())
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+        self.assertIsNotNone(victim.wait(timeout=5))
+
+
+# -- V-3 附带：hibernate 留盘 + GC 保护 + restore 盘预检 -------------------
+class HibernateDiskV3Test(V3Case):
+    def test_hibernate_keeps_task_disk(self):
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        disk = os.path.join(self.state_dir, "chv-abc123", "disk.raw")
+        self.assertTrue(os.path.exists(disk))
+        url, _ = self._snap_dir()
+        out = ex.hibernate_vm(self._task(), url)
+        self.assertEqual(out["state"], "hibernated")
+        self.assertTrue(os.path.exists(disk),
+                        "hibernate 的 stop 不得删盘：快照 config.json "
+                        "引用该盘，删了 restore 即失败")
+
+    def test_gc_spares_disk_while_snapshot_recorded(self):
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        url, _ = self._snap_dir()
+        ex.hibernate_vm(self._task(), url)
+        disk = os.path.join(self.state_dir, "chv-abc123", "disk.raw")
+        result = ex.gc_stale_disks()
+        self.assertEqual(result["deleted"], [])
+        self.assertTrue(os.path.exists(disk))
+        # 快照登记清掉后，盘成孤儿，GC 才可删。
+        os.remove(os.path.join(self.state_dir, "chv-abc123",
+                               "snapshots.json"))
+        result = ex.gc_stale_disks()
+        self.assertEqual(result["deleted"], ["chv-abc123"])
+        self.assertFalse(os.path.exists(disk))
+
+    def test_restore_missing_snapshot_disk_fail_fast(self):
+        ex = self.make_exec()
+        missing_disk = os.path.join(self.state_dir, "gone", "disk.raw")
+        url, _ = self._snap_dir(files={
+            "config.json": json.dumps(
+                {"disks": [{"path": missing_disk}]}).encode(),
+            "state.json": b"{}"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm(self._task(), url)
+        self.assertEqual(ctx.exception.kind, "snapshot_disk_missing")
+        self.assertEqual(self.launches, [])
+
+    def test_restore_snapshot_disk_present_passes_precheck(self):
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        disk = os.path.join(self.state_dir, "chv-abc123", "disk.raw")
+        url, _ = self._snap_dir(files={
+            "config.json": json.dumps(
+                {"disks": [{"path": disk}]}).encode(),
+            "state.json": b"{}"})
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+
+
+# -- F1：re-wake 删旧盘 -----------------------------------------------------
+class RewakeDiskV3Test(V3Case):
+    def _corrupt_disk(self, task_id="abc123"):
+        disk = os.path.join(self.state_dir, "chv-" + task_id,
+                            "disk.raw")
+        with open(disk, "wb") as fh:
+            fh.write(b"STALE" * 20)
+        return disk
+
+    def test_rewake_live_vm_replaces_stale_disk(self):
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        disk = self._corrupt_disk()
+        out = ex.wake_vm(self._task())
+        self.assertEqual(out["state"], "running")
+        with open(disk, "rb") as fh:
+            self.assertEqual(fh.read(), b"R" * 100,
+                             "re-wake 须重拷新盘：旧盘复用 + 新 key 对 "
+                             "= cloud-init 不重应用 = host_key_mismatch")
+
+    def test_rewake_after_crash_replaces_stale_disk(self):
+        ex = self.make_exec()
+        ex.wake_vm(self._task())
+        disk = self._corrupt_disk()
+        ex2 = self.make_exec()
+        out = ex2.wake_vm(self._task())
+        self.assertEqual(out["state"], "running")
+        with open(disk, "rb") as fh:
+            self.assertEqual(fh.read(), b"R" * 100)
+
+    def test_provisioned_rewake_regenerates_keypair_on_fresh_disk(self):
+        ex = self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        out = ex.wake_vm(self._task())
+        self.assertEqual(out["state"], "running")
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        # H5：首 wake 后宿主私钥已删，re-wake 必然重新生成 key 对。
+        self.assertFalse(os.path.exists(
+            os.path.join(vm_dir, "ssh_host_ed25519_key")))
+        self.assertTrue(os.path.exists(
+            os.path.join(vm_dir, "known_hosts")))
+        disk = self._corrupt_disk()
+        geniso_before = sum(1 for c in self.calls
+                            if c and c[0] == "genisoimage")
+        out2 = ex.wake_vm(self._task())
+        self.assertEqual(out2["state"], "running")
+        with open(disk, "rb") as fh:
+            self.assertEqual(fh.read(), b"R" * 100)
+        geniso_after = sum(1 for c in self.calls
+                           if c and c[0] == "genisoimage")
+        self.assertEqual(geniso_after, geniso_before + 1,
+                         "re-wake 须重建 seed 盘（新 key 对进新 seed）")
+
+
+# -- F2：跨实例 stop 按 socket 进程扫描 -------------------------------------
+class CrossInstanceStopV3Test(V3Case):
+    def test_stop_kills_unregistered_vmm_by_socket(self):
+        ex = self.make_exec()
+        sock = os.path.join(self.state_dir, "chv-abc123.sock")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(sock, "w") as fh:
+            fh.write("")
+        victim = self._spawn_fake_vmm(sock)
+        self.assertIsNone(victim.poll())
+        self.assertTrue(ex.stop_vm(self._task()))
+        self.assertIsNotNone(victim.wait(timeout=5))
+        self.assertFalse(os.path.exists(sock))
+
+    def test_stop_does_not_kill_others_vmm_or_chremote(self):
+        ex = self.make_exec()
+        sock = os.path.join(self.state_dir, "chv-abc123.sock")
+        other_sock = os.path.join(self.state_dir, "chv-other.sock")
+        os.makedirs(self.state_dir, exist_ok=True)
+        for path in (sock, other_sock):
+            with open(path, "w") as fh:
+                fh.write("")
+        other_vmm = self._spawn_fake_vmm(other_sock)
+        # 同带本 socket 的 ch-remote 客户端：真实形态是
+        # `ch-remote --api-socket=<sock> shutdown`，不是 VMM，
+        # 绝不杀。
+        chremote = subprocess.Popen(
+            ["ch-remote", "-c", "import time; time.sleep(30)",
+             "--api-socket=" + sock],
+            executable=sys.executable,
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        self.spawned.append(chremote)
+        self.assertIsNone(other_vmm.poll())
+        self.assertIsNone(chremote.poll())
+        ex.stop_vm(self._task())
+        self.assertIsNone(other_vmm.poll())
+        self.assertIsNone(chremote.poll())
+
+
+# -- V-4：guest-env source 失败 fail-fast + 归 SSH 用户所有 ------------------
+class GuestEnvV4Test(V3Case):
+    def test_remote_command_source_failure_exits(self):
+        tokens = remote_command(["python3", "/app/host.py", "status"],
+                                guest_env_file="/etc/chv/guest-env")
+        joined = " ".join(tokens)
+        self.assertIn(". /etc/chv/guest-env || exit 126;", joined)
+        self.assertNotIn("/etc/chv/guest-env';", joined)
+
+    def test_remote_command_without_guest_env_unchanged(self):
+        tokens = remote_command(["true"])
+        self.assertEqual(tokens, ["true"])
+
+    def test_write_seed_owner_follows_ssh_user(self):
+        import yaml
+        for ssh_user, want_owner in (
+                (None, "root:root"), ("root", "root:root"),
+                ("chv", "chv:chv")):
+            seed_dir = os.path.join(self.dir, "seed-%s" % want_owner)
+            write_seed(seed_dir, "vm1", "PRIV",
+                       "ssh-ed25519 AAAAx",
+                       secret_env={"SUB2API_API_KEY": "s3cr3t"},
+                       ssh_user=ssh_user)
+            with open(os.path.join(seed_dir, "user-data")) as fh:
+                parsed = yaml.safe_load(fh.read())
+            guest = [e for e in parsed["write_files"]
+                     if e["path"] == "/etc/chv/guest-env"]
+            self.assertEqual(len(guest), 1)
+            self.assertEqual(guest[0]["owner"], want_owner)
+            self.assertEqual(guest[0]["permissions"], "0600")
+
+    def test_provisioned_wake_writes_guest_env_for_chv_user(self):
+        import yaml
+        ex = self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"},
+            ssh_user="chv")
+        out = ex.wake_vm(self._task(),
+                         extra_env={"SUB2API_API_KEY": "s3cr3t"})
+        self.assertEqual(out["state"], "running")
+        user_data = os.path.join(
+            self.state_dir, "chv-abc123", "seed", "user-data")
+        with open(user_data) as fh:
+            parsed = yaml.safe_load(fh.read())
+        guest = [e for e in parsed["write_files"]
+                 if e["path"] == "/etc/chv/guest-env"]
+        self.assertEqual(len(guest), 1)
+        self.assertEqual(guest[0]["owner"], "chv:chv")
+
+
+# -- V-1 附带：build.sh /app/work + host.py default_codex_bin --------------
+class GuestBuildAndHostBinV3Test(unittest.TestCase):
+    def test_build_sh_precreates_app_work_writable(self):
+        text = (ROOT / "chv-guest" / "build.sh").read_text()
+        self.assertIn("mkdir -p /app/work", text)
+        self.assertIn("chmod 1777 /app/work", text)
+        self.assertIn('stat -c %a /app/work', text)
+
+    def test_build_sh_dry_run_prints_app_work_step(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        bindir = os.path.join(tmp.name, "bin")
+        os.makedirs(os.path.join(bindir, "codex-resources"))
+        for name in ("codex", "codex-code-mode-host"):
+            p = os.path.join(bindir, name)
+            with open(p, "w") as fh:
+                fh.write("#!/bin/sh\necho \"codex-cli 0.164.0-alpha.9\"\n")
+            os.chmod(p, 0o755)
+        repo = os.path.join(tmp.name, "repo")
+        os.makedirs(repo)
+        with open(os.path.join(repo, "host.py"), "w") as fh:
+            fh.write("# host\n")
+        subprocess.run(["git", "init", "-q", repo],
+                       check=True, capture_output=True)
+        subprocess.run(["git", "-C", repo, "config", "user.email", "t@t"],
+                       check=True, capture_output=True)
+        subprocess.run(["git", "-C", repo, "config", "user.name", "t"],
+                       check=True, capture_output=True)
+        subprocess.run(["git", "-C", repo, "add", "-A"],
+                       check=True, capture_output=True)
+        subprocess.run(["git", "-C", repo, "commit", "-qm", "init"],
+                       check=True, capture_output=True)
+        commit = subprocess.run(
+            ["git", "-C", repo, "rev-parse", "--short", "HEAD"],
+            check=True, capture_output=True, text=True).stdout.strip()
+        env = dict(os.environ)
+        env.update({"CODEX_BIN_DIR": bindir,
+                    "UBUNTU_VERSION": "24.04",
+                    "CODEX_VERSION": "0.164.0-alpha.9",
+                    "REPO_COMMIT": commit,
+                    "UBUNTU_IMAGE_SHA256": "0" * 64,
+                    "REPO_DIR": repo})
+        r = subprocess.run(
+            ["bash", str(ROOT / "chv-guest" / "build.sh"),
+             "--dry-run", os.path.join(tmp.name, "out")],
+            check=False, capture_output=True, text=True, env=env)
+        self.assertEqual(r.returncode, 0, r.stderr)
+        self.assertIn("/app/work", r.stdout)
+        self.assertIn("1777", r.stdout)
+
+    def test_default_codex_bin_uses_path_lookup(self):
+        mod = _load_host_py()
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        fake = os.path.join(tmp.name, "codex")
+        with open(fake, "w") as fh:
+            fh.write("#!/bin/sh\n")
+        os.chmod(fake, 0o755)
+        env = dict(os.environ)
+        env.pop("CODEX_BIN", None)
+        env["PATH"] = tmp.name + os.pathsep + env.get("PATH", "")
+        with mock.patch.dict(os.environ, env, clear=True):
+            self.assertEqual(mod.default_codex_bin(), fake)
+
+    def test_default_codex_bin_falls_back_to_bare_name(self):
+        mod = _load_host_py()
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        env = {"PATH": tmp.name}
+        with mock.patch.dict(os.environ, env, clear=True):
+            self.assertEqual(mod.default_codex_bin(), "codex")
+
+    def test_default_codex_bin_env_still_wins(self):
+        mod = _load_host_py()
+        with mock.patch.dict(os.environ,
+                             {"CODEX_BIN": "/opt/custom/codex"}):
+            self.assertEqual(mod.default_codex_bin(),
+                             "/opt/custom/codex")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_cloud_hypervisor.py

@@ -429,21 +429,29 @@ class LifecycleTest(ChvCase):
     def test_hibernate_is_pause_snapshot_shutdown(self):
         ex = self.make_exec()
         # feat-chv-fix（P1-7）：file:// 快照须真实存在且非空，
-        # 否则 hibernate 不 shutdown；用临时文件作快照目标。
+        # 否则 hibernate 不 shutdown；feat-chv-v3：v54 快照
+        # 产物是目录，用含 config.json 的目录作快照目标。
         # feat-chv-secfix（L2）：快照须在白名单目录内，放在
         # state_dir 下。
         snap = os.path.join(self.state_dir, "h1.snap")
-        os.makedirs(self.state_dir, exist_ok=True)
-        with open(snap, "wb") as fh:
-            fh.write(b"S" * 64)
+        os.makedirs(snap, exist_ok=True)
+        with open(os.path.join(snap, "config.json"), "wb") as fh:
+            fh.write(b"{}")
         snap_url = "file://" + snap
         out = ex.hibernate_vm(self._task(), snap_url)
         self.assertEqual(out["state"], "hibernated")
         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")
-                            for c in self.calls for a in c))
+        # feat-chv-v3：snapshot 只传裸 URL（source_url= 是
+        # restore 的键，传给 snapshot 会被服务端当成 URL 本身
+        # 而失败）；restore 仍传 source_url=...。
+        snap_call = next(c for c in self.calls
+                         if c[0] == self.ch_remote and c[-1] != "pause"
+                         and c[-1] != "shutdown")
+        self.assertEqual(snap_call[-1], snap_url)
+        self.assertFalse(any(a.startswith("source_url")
+                             for a in snap_call))
         self.assertIn("shutdown", kinds)
         # 从休眠唤醒 = restore + resume：
         ex2 = self.make_exec()