fix(chv): V-7 复测缺陷复修 D1/D2 + F1 瞬时 reset 重试

D1：snapshot_vm 在配额预检后、ch-remote 前建好 file:// 目标
目录（新 helper _ensure_snapshot_dir；fresh URL 直接 snapshot
曾 rc=1 Destination is not a directory）。建目录只在 snapshot_vm
一处，hibernate 经此调用；配额拒绝仍先于任何文件改动。
D2：keep_disk=True 时 stop 保留 seed.iso（_remove_vm_seed_files
抽出；快照 config.json 的 disks 同时引用 disk.raw 与 seed.iso），
且 cleanup_stale_secrets 不清有登记快照的 VM 的 seed（与
gc_stale_disks 的盘保护同款逻辑）；正常 stop 仍删两者。
F1：首条任务 SSH 撞 kex 阶段瞬时 reset（rc=255 + kex/reset 标记）
时重试恰好 1 次（_call_ssh_with_transient_retry，_run_guest 与
_run_guest_goal 共用；host key 拒绝/认证失败/远端失败不重试；
审计 vm_ssh_transient_retry）。§5“执行器内部不重试”的唯一例外，
结论写入 docs/chv-production-readiness.md §3/§5/§7。
测试：test_chv_v3_fixes 新增 12 项（D1×3、D2×4、F1×5）；旧代码
上 12 项中 9 项失败（余 3 项为守卫既有行为）。全量 test_chv*
470 测试通过。

cloud/tests/test_chv_v3_fixes.py

@@ -13,6 +13,11 @@
   guest-env 归 SSH 用户所有；
 - V-1 附带：build.sh 预建 /app/work（1777 粘滞位 + 断言）；
 - V-1 附带：host.py default_codex_bin() 走 PATH 查找。
+- V-7 复修（feat-chv-v3-repair）：D1 snapshot 先建目录（旧
+  fixture 预建目录 + mock ch-remote 不校验，458 测试全过也
+  盖不住缺失目录的真实路径）、D2 hibernate 留 seed.iso +
+  残留清理保护有登记快照的 VM 的 seed、F1 任务 SSH 连接建立
+  阶段瞬时 reset 重试恰好 1 次。
 
 全部用假 runner/launcher/api_client + 真实 /proc 扫描（F2
 的被杀进程是真实 spawn 的 sleep 进程，argv[0] 伪装成
@@ -545,5 +550,217 @@ class GuestBuildAndHostBinV3Test(unittest.TestCase):
                              "/opt/custom/codex")
 
 
+# -- V-7/D1：snapshot 先建目标目录 ---------------------------------------
+class SnapshotMkdirV7Test(V3Case):
+    def _snapshot_creating_runner(self):
+        """模拟真实 ch-remote：往目标目录写快照文件。目录不
+        存在时 open 即 FileNotFoundError——若执行器没先
+        mkdir，测试直接失败，这正是旧 fixture 盖不住的点
+        （旧 fixture 预建目录，mock 又不写文件）。"""
+        def runner(cmd, timeout=None):
+            if cmd and cmd[0] == self.ch_remote and "snapshot" in cmd:
+                self.calls.append(list(cmd))
+                target = cmd[-1][len("file://"):]
+                for fname, content in (
+                        ("config.json", b"{}"),
+                        ("state.json", b"{}"),
+                        ("memory-ranges", b"m")):
+                    with open(os.path.join(target, fname), "wb") as fh:
+                        fh.write(content)
+                return FakeResult(0, "", "")
+            return self._default_runner(cmd, timeout)
+        return runner
+
+    def test_snapshot_creates_missing_snapshot_dir(self):
+        ex = self.make_exec(runner=self._snapshot_creating_runner())
+        url = "file://" + os.path.join(self.state_dir, "fresh-snap")
+        self.assertFalse(os.path.exists(url[len("file://"):]))
+        out = ex.snapshot_vm(self._task(), url)
+        self.assertEqual(out, url)
+        snap_path = url[len("file://"):]
+        self.assertTrue(os.path.isdir(snap_path))
+        self.assertIn("config.json", os.listdir(snap_path))
+        with open(os.path.join(self.state_dir, "chv-abc123",
+                               "snapshots.json")) as fh:
+            entries = json.load(fh)
+        self.assertEqual(entries[0]["path"], snap_path)
+
+    def test_hibernate_creates_missing_snapshot_dir(self):
+        ex = self.make_exec(runner=self._snapshot_creating_runner())
+        ex.wake_vm(self._task())
+        url = "file://" + os.path.join(self.state_dir, "fresh-snap")
+        out = ex.hibernate_vm(self._task(), url)
+        self.assertEqual(out["state"], "hibernated")
+        self.assertTrue(os.path.isdir(url[len("file://"):]))
+
+    def test_snapshot_target_existing_file_rejected(self):
+        ex = self.make_exec()
+        target = os.path.join(self.state_dir, "not-a-dir")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(target, "wb") as fh:
+            fh.write(b"x")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.snapshot_vm(self._task(), "file://" + target)
+        self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
+        self.assertEqual(self.calls, [],
+                         "目标不是目录时 ch-remote 根本不应被调")
+
+
+# -- V-7/D2：hibernate 留 seed.iso ----------------------------------------
+class HibernateSeedIsoV7Test(V3Case):
+    def _prov_exec(self):
+        # keyscan 模式不制 seed.iso；provisioned 模式 wake 即
+        # 建 seed.iso（真机休眠的形态），用它走全链路。
+        return self.make_exec(
+            env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+
+    def _seed_iso(self):
+        return os.path.join(self.state_dir, "chv-abc123", "seed.iso")
+
+    def _disk(self):
+        return os.path.join(self.state_dir, "chv-abc123", "disk.raw")
+
+    def _snap_with_disks(self):
+        """快照 config.json 同时引用 disk.raw 与 seed.iso
+        （真机快照的 disks 形态）。"""
+        return self._snap_dir(files={
+            "config.json": json.dumps(
+                {"disks": [{"path": self._disk()},
+                           {"path": self._seed_iso()}]}).encode(),
+            "state.json": b"{}"})
+
+    def test_hibernate_keeps_seed_iso(self):
+        ex = self._prov_exec()
+        ex.wake_vm(self._task())
+        self.assertTrue(os.path.exists(self._seed_iso()))
+        url, _ = self._snap_dir()
+        out = ex.hibernate_vm(self._task(), url)
+        self.assertEqual(out["state"], "hibernated")
+        self.assertTrue(os.path.exists(self._disk()),
+                        "hibernate 仍须留 disk.raw")
+        self.assertTrue(os.path.exists(self._seed_iso()),
+                        "hibernate 须留 seed.iso：快照 config.json "
+                        "的 disks 引用它，删了 restore 即失败")
+        # 只留 iso：seed 目录（制盘输入）与宿主私钥仍删，
+        # config.json 不引用它们，私钥明文不留宿主。
+        self.assertFalse(os.path.exists(
+            os.path.join(self.state_dir, "chv-abc123", "seed")))
+        self.assertFalse(os.path.exists(
+            os.path.join(self.state_dir, "chv-abc123",
+                         "ssh_host_ed25519_key")))
+
+    def test_stop_without_keep_disk_removes_seed_iso(self):
+        ex = self._prov_exec()
+        ex.wake_vm(self._task())
+        self.assertTrue(ex.stop_vm(self._task()))
+        self.assertFalse(os.path.exists(self._seed_iso()),
+                         "正常 stop 仍须删 seed.iso：它含宿主私钥"
+                         "明文，不得随 VM 残留")
+        self.assertFalse(os.path.exists(self._disk()))
+
+    def test_cleanup_stale_secrets_spares_seed_iso_while_snapshot_recorded(self):
+        ex = self._prov_exec()
+        ex.wake_vm(self._task())
+        url, _ = self._snap_dir()
+        ex.hibernate_vm(self._task(), url)
+        cleaned = ex.cleanup_stale_secrets()
+        self.assertNotIn("chv-abc123", cleaned)
+        self.assertTrue(os.path.exists(self._seed_iso()),
+                        "有登记快照的 VM 的 seed.iso 不是孤儿，"
+                        "残留清理不得清掉（与 gc_stale_disks 的"
+                        "盘保护同款逻辑）")
+        # 快照登记清掉后，seed.iso 成孤儿，才可清。
+        os.remove(os.path.join(self.state_dir, "chv-abc123",
+                               "snapshots.json"))
+        cleaned = ex.cleanup_stale_secrets()
+        self.assertIn("chv-abc123", cleaned)
+        self.assertFalse(os.path.exists(self._seed_iso()))
+
+    def test_restore_after_hibernate_with_seed_iso_passes_precheck(self):
+        ex = self._prov_exec()
+        ex.wake_vm(self._task())
+        url, _ = self._snap_with_disks()
+        ex.hibernate_vm(self._task(), url)
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+
+
+# -- V-7/F1：任务 SSH 连接建立阶段瞬时 reset 重试 1 次 --------------------
+class SshTransientRetryV7Test(V3Case):
+    def _dispatch_runner(self, first, second=None):
+        """首条 dispatch SSH 返回 first，之后返回 second
+        （默认成功）。只记 dispatch SSH 的调用次数。"""
+        state = {"n": 0}
+
+        def runner(cmd, timeout=None):
+            if cmd and cmd[0] == "ssh" and "dispatch" in cmd:
+                state["n"] += 1
+                if state["n"] == 1:
+                    return first
+                return second or FakeResult(
+                    0, json.dumps({"status": "completed",
+                                   "text": "ok"}), "")
+            return self._default_runner(cmd, timeout)
+        runner.state = state
+        return runner
+
+    def test_dispatch_retries_once_on_kex_reset_then_succeeds(self):
+        runner = self._dispatch_runner(FakeResult(
+            255, "",
+            "kex_exchange_identification: Connection reset by peer"))
+        ex = self.make_exec(runner=runner)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(runner.state["n"], 2,
+                         "kex 阶段瞬时 reset 只重试恰好 1 次")
+        self.assertTrue(any(e == "vm_ssh_transient_retry"
+                            for e, _t, _d, _k in self.audit))
+
+    def test_dispatch_does_not_retry_on_host_key_mismatch(self):
+        runner = self._dispatch_runner(FakeResult(
+            255, "", "Host key verification failed."))
+        ex = self.make_exec(runner=runner)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "failed")
+        self.assertEqual(runner.state["n"], 1,
+                         "host key 拒绝（同为 255）绝不重试")
+        self.assertFalse(any(e == "vm_ssh_transient_retry"
+                             for e, _t, _d, _k in self.audit))
+
+    def test_dispatch_does_not_retry_on_remote_failure(self):
+        runner = self._dispatch_runner(FakeResult(
+            1, json.dumps({"status": "failed", "text": "boom"}), ""))
+        ex = self.make_exec(runner=runner)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "failed")
+        self.assertEqual(runner.state["n"], 1, "远端命令失败不重试")
+
+    def test_dispatch_retries_at_most_once(self):
+        runner = self._dispatch_runner(
+            FakeResult(255, "", "Connection reset by peer"),
+            second=FakeResult(255, "", "Connection reset by peer"))
+        ex = self.make_exec(runner=runner)
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "failed")
+        self.assertEqual(runner.state["n"], 2, "第二次失败即停，不再追")
+
+    def test_transient_retry_helper_retries_once(self):
+        """helper 本身（_run_guest 与 _run_guest_goal 共用）：
+        首条 255 + reset 标记即重试 1 次，第二次成功即返回。"""
+        calls = []
+
+        def runner(cmd, timeout=None):
+            calls.append(list(cmd))
+            if len(calls) == 1:
+                return FakeResult(255, "", "Connection reset by peer")
+            return FakeResult(0, "ok", "")
+
+        ex = self.make_exec(runner=runner)
+        result = ex._call_ssh_with_transient_retry(
+            self._task(), ["ssh", "true"])
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(len(calls), 2)
+
+
 if __name__ == "__main__":
     unittest.main()