diff --git a/cloud/console-react/src/api/chv.closeup.test.ts b/cloud/console-react/src/api/chv.closeup.test.ts
new file mode 100644
index 0000000..47adda4
--- /dev/null
+++ b/cloud/console-react/src/api/chv.closeup.test.ts
@@ -0,0 +1,76 @@
+import { describe, expect, it } from "vitest";
+import { chvAttachmentBlocked, chvAvailability, chvAvailabilityText, chvQueueHint, chvVmBadge, chvVmDisplay, chvVmStatusText, countChvWorkers, isWorkerAlive } from "./chv";
+
+// closeup 轨补边角：上一轨 chv.test.ts 已覆主路径后仍缺的
+// 显式 false、undefined、优先级与“旧 bug 文案不复发”断言。
+// 只测 chv.ts 纯逻辑，不渲染组件、不发请求。
+describe("chvVmDisplay 边角（closeup）", () => {
+  it("字段显式 false 与 absent 同属 not-reported", () => {
+    const d = chvVmDisplay({ executor: "cloud-hypervisor", chv_vm_dead: false });
+    expect(d.state).toBe("not-reported");
+    expect(d.text).toContain("不代表 VM 活着");
+  });
+  it("任务 undefined 与 null 同属 unknown", () => {
+    expect(chvVmDisplay(undefined).state).toBe("unknown");
+    expect(chvVmDisplay(undefined).text).toContain("无法判断");
+  });
+  it("非 CHV 且无 executor 字段时也是 not-applicable", () => {
+    expect(chvVmDisplay({}).state).toBe("not-applicable");
+    expect(chvVmDisplay({ executor: undefined }).text).toBe("");
+    expect(chvVmStatusText({ executor: "cloud-hypervisor", chv_vm_dead: false })).toContain("未报告 VM 死亡");
+  });
+  it("列表徽标：显式 false 不出徽标", () => {
+    expect(chvVmBadge({ executor: "cloud-hypervisor", chv_vm_dead: false })).toBe("");
+  });
+});
+
+describe("chvAvailability 优先级（closeup）", () => {
+  it("加载失败优先于聚合文案：不把失败写成 aggregate-only", () => {
+    expect(chvAvailability(null, "完整 worker 清单仅全局管理员可见", "加载失败：x")).toEqual(
+      { state: "unknown", reason: "load-failed" },
+    );
+  });
+  it("有清单时忽略错误与聚合文案", () => {
+    expect(chvAvailability([{ executor: "cloud-hypervisor", alive: false }], "完整 worker 清单仅全局管理员可见", "加载失败：x")).toEqual(
+      { state: "known", online: 0, total: 1 },
+    );
+  });
+  it("空清单与空文案是 not-loaded，不是已知 0", () => {
+    expect(chvAvailability(null, "", "")).toEqual({ state: "unknown", reason: "not-loaded" });
+  });
+  it("空 worker 数组是已知 0（与 null 区分）", () => {
+    expect(chvAvailability([], "", "")).toEqual({ state: "known", online: 0, total: 0 });
+    expect(countChvWorkers([])).toEqual({ online: 0, total: 0 });
+  });
+});
+
+describe("chvAvailabilityText：Workers 旧 bug 不复发（closeup）", () => {
+  it("load-failed / not-loaded 绝不写“仅全局管理员可见”", () => {
+    expect(chvAvailabilityText({ state: "unknown", reason: "load-failed" })).not.toContain("仅全局管理员可见");
+    expect(chvAvailabilityText({ state: "unknown", reason: "not-loaded" })).not.toContain("仅全局管理员可见");
+  });
+  it("known 总数为 0 时仍是已知 0，不写未知", () => {
+    const text = chvAvailabilityText({ state: "known", online: 0, total: 0 });
+    expect(text).toContain("在线 0 / 共 0");
+    expect(text).not.toContain("未知");
+  });
+});
+
+describe("附件拦截与排队提示边角（closeup）", () => {
+  it("executor 为 undefined/null 时附件不拦", () => {
+    expect(chvAttachmentBlocked(undefined, true)).toBe(false);
+    expect(chvAttachmentBlocked(null, true)).toBe(false);
+  });
+  it("排队提示：未知三原因一律为空串", () => {
+    for (const reason of ["aggregate-only", "load-failed", "not-loaded"] as const) {
+      expect(chvQueueHint("cloud-hypervisor", "queued", { state: "unknown", reason })).toBe("");
+    }
+  });
+});
+
+describe("isWorkerAlive 边界（closeup）", () => {
+  it("last_seen 恰好 120 秒算在线（<=120）", () => {
+    const now = Date.now() / 1000;
+    expect(isWorkerAlive({ executor: "cloud-hypervisor", last_seen: now - 120 })).toBe(true);
+  });
+});
diff --git a/cloud/tests/test_chv_closeup.py b/cloud/tests/test_chv_closeup.py
new file mode 100644
index 0000000..edd2a85
--- /dev/null
+++ b/cloud/tests/test_chv_closeup.py
@@ -0,0 +1,1326 @@
+"""CHV 收尾轨（feat-chv-closeup）新增测试。
+
+只补上一轨 coverage 下 cloud/chv/executor.py 仍未覆盖的
+深层生命周期分支：唤醒失败（vm.create/vm.boot/API 超时/
+guest 就绪超时/host key 不匹配/provisioned 缺钉死）、
+恢复与快照失败（restore 失败关 VM、快照未落盘/配额/
+总量剪枝）、取消与清理（cancel 真关、recycle、残留
+socket/登记/磁盘/密钥扫描、脏标记）、中途死亡分类
+（异常与 failed 状态改判 interrupted）、只读 execute
+归类、入口校验、preflight 事后改写校验、worker_adapter
+看门狗/调和的剩余分支。
+
+全部用标准库与 mock runner/launcher/api_client，绝不启动
+真实 VM（本机无 /dev/kvm，真机验证仍是 §13.2 的 V-1～V-9，
+不在本文件声称已验证）。
+"""
+
+import inspect
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.hostkeys import dirty_marker_path
+    from cloud.chv.worker_adapter import (
+        call_goal_resume_breakpoint,
+        cancel_watchdog_loop,
+        quota_watchdog_loop,
+        reconcile_warm_pool,
+        watchdog_vm_name,
+    )
+    from cloud.sandbox import ChvFailure
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.hostkeys import dirty_marker_path
+    from cloud.chv.worker_adapter import (
+        call_goal_resume_breakpoint,
+        cancel_watchdog_loop,
+        quota_watchdog_loop,
+        reconcile_warm_pool,
+        watchdog_vm_name,
+    )
+    from cloud.sandbox import ChvFailure
+
+
+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
+
+
+class CloseupCase(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
+
+    def tearDown(self):
+        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[0] == "ssh":
+            if cmd[-1] == "true":
+                return FakeResult(0, "", "")
+            if "dispatch" in cmd:
+                return FakeResult(0, json.dumps(
+                    {"status": "completed", "text": "ok-dispatch"}), "")
+            if "goal" in cmd or "goal-resume" in cmd:
+                return FakeResult(0, "goal met after rounds\n", "")
+            if "status" in cmd:
+                return FakeResult(0, json.dumps(
+                    {"status": "completed", "text": "ok-status"}), "")
+            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 _audit_events(self):
+        return [e for e, _t, _d, _k in self.audit]
+
+    def _wake_ok(self, ex, task=None):
+        return ex.wake_vm(task or self._task())
+
+    def _known_hosts(self, task_id="abc123"):
+        vm_dir = os.path.join(self.state_dir, "chv-" + task_id)
+        os.makedirs(vm_dir, exist_ok=True)
+        path = os.path.join(vm_dir, "known_hosts")
+        with open(path, "w") as fh:
+            fh.write("127.0.0.1 ssh-ed25519 AAAAC3\n")
+        return path
+
+
+# -- 唤醒失败 ------------------------------------------------------------
+class WakeFailureTest(CloseupCase):
+    def test_wake_prechecks_secret_keyscan_fail_closed(self):
+        ex = self.make_exec()
+        # 密钥子集的判定名是 SUB2API_API_KEY（见 ssh_transport
+        # 的 _CHV_SECRET_ENV_NAMES），其它名走普通 env 前缀。
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task(),
+                       extra_env={"SUB2API_API_KEY": "secret-value"})
+        self.assertEqual(ctx.exception.kind,
+                         "secret_env_requires_provisioned")
+        self.assertIn("vm_failed", self._audit_events())
+        self.assertEqual(self.launches, [])
+
+    def test_wake_vm_create_failed_closes_vm(self):
+        def api(socket_path, method, path, payload=None):
+            self.api_calls.append((method, path))
+            if method == "PUT" and path == "/vm.create":
+                return (500, {})
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_create_failed")
+        self.assertIn("vm_failed", self._audit_events())
+        self.assertEqual(ex._vms, {})
+        self.assertTrue(any(c[-1] == "shutdown" for c in self.calls))
+
+    def test_wake_vm_boot_failed_closes_vm(self):
+        def api(socket_path, method, path, payload=None):
+            self.api_calls.append((method, path))
+            if method == "PUT" and path == "/vm.boot":
+                return (500, {})
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_boot_failed")
+        self.assertIn("vm_failed", self._audit_events())
+        self.assertEqual(ex._vms, {})
+
+    def test_wake_api_timeout(self):
+        def api(socket_path, method, path, payload=None):
+            return (404, {})
+        ex = self.make_exec(api_client=api, api_timeout=2)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "api_timeout")
+        self.assertIn("vm_failed", self._audit_events())
+
+    def test_wake_failure_stop_exception_swallowed(self):
+        def api(socket_path, method, path, payload=None):
+            if method == "PUT" and path == "/vm.create":
+                return (500, {})
+            return (200, {})
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[-1] == "shutdown":
+                raise RuntimeError("shutdown boom")
+            return FakeResult(0, "", "")
+        ex = self.make_exec(api_client=api, runner=runner)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        # 原始失败仍是 vm_create_failed，不被 stop 的异常掩盖。
+        self.assertEqual(ctx.exception.kind, "vm_create_failed")
+
+    def test_wait_guest_ready_host_key_mismatch(self):
+        self._known_hosts()
+        ex = self.make_exec()
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and cmd[-1] == "true":
+                return FakeResult(1, "",
+                                  "Host key verification failed")
+            return FakeResult(0, "", "")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "host_key_mismatch")
+        self.assertIn("host_key_mismatch", self._audit_events())
+
+    def test_wait_guest_ready_timeout(self):
+        self._known_hosts()
+        ex = self.make_exec(ready_timeout=2, ready_interval=1)
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh":
+                return FakeResult(1, "", "")
+            return FakeResult(0, "", "")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "guest_not_ready")
+
+    def test_wait_guest_ready_provisioned_missing_known_hosts(self):
+        ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
+
+
+# -- 恢复与快照失败 ------------------------------------------------------
+class RestoreSnapshotTest(CloseupCase):
+    def _snap_url(self, name="snap1.raw"):
+        path = os.path.join(self.state_dir, name)
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(b"S" * 60)
+        return "file://" + path, path
+
+    def test_restore_failure_closes_vm(self):
+        url, _ = self._snap_url()
+        ex = self.make_exec()
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == self.ch_remote and "restore" in cmd:
+                return FakeResult(1, "", "")
+            return self._default_runner(cmd, timeout)
+        ex.runner = runner
+        with self.assertRaises(RuntimeError):
+            ex.restore_vm(self._task(), url)
+        self.assertEqual(ex._vms, {})
+        self.assertTrue(any(c[-1] == "shutdown" for c in self.calls))
+
+    def test_restore_success_and_wake_from_hibernate(self):
+        url, _ = self._snap_url()
+        ex = self.make_exec()
+        out = ex.restore_vm(self._task(), url)
+        self.assertEqual(out["mode"], "restored")
+        self.assertEqual(out["state"], "running")
+        out2 = ex.wake_from_hibernate(self._task(), url)
+        self.assertEqual(out2["mode"], "restored")
+
+    def test_restore_provisioned_missing_known_hosts(self):
+        url, _ = self._snap_url()
+        ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm(self._task(), url)
+        self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
+        self.assertEqual(self.launches, [])
+
+    def test_snapshot_not_persisted_missing_and_empty(self):
+        ex = self.make_exec()
+        missing = "file://" + os.path.join(self.state_dir, "nope.raw")
+        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()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.snapshot_vm(self._task(), "file://" + empty_path)
+        self.assertEqual(ctx.exception.kind, "snapshot_not_persisted")
+
+    def test_snapshot_prunes_oldest_on_total_cap(self):
+        vm = "chv-abc123"
+        vm_dir = os.path.join(self.state_dir, vm)
+        os.makedirs(vm_dir, exist_ok=True)
+        old_path = os.path.join(self.state_dir, "old.raw")
+        with open(old_path, "wb") as fh:
+            fh.write(b"O" * 100)
+        import time
+        with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh:
+            json.dump([{"url": "file://" + old_path, "path": old_path,
+                        "size": 100, "created_at": time.time()}], fh)
+        # 上限 150 字节：预检时旧 100 字节通过，新 60 字节记入后
+        # 总量 160 超帽，最旧一条被剪枝并审计。
+        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)
+        self.assertFalse(os.path.exists(old_path))
+        self.assertIn("vm_snapshot_pruned", self._audit_events())
+
+    def test_snapshot_quota_count(self):
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        os.makedirs(vm_dir, exist_ok=True)
+        import time
+        with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh:
+            json.dump([{"url": "file://x", "path": None, "size": 0,
+                        "created_at": time.time()}], fh)
+        ex = self.make_exec(env_extra={"CHV_SNAPSHOT_MAX_COUNT": "1"})
+        url, _ = self._snap_url("snap2.raw")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.snapshot_vm(self._task(), url)
+        self.assertEqual(ctx.exception.kind, "snapshot_quota_exceeded")
+
+    def test_hibernate_snapshot_failure_does_not_shutdown(self):
+        ex = self.make_exec()
+        with self.assertRaises(ValueError):
+            ex.hibernate_vm(self._task(), "")
+        self.assertFalse(any(c[-1] == "shutdown" for c in self.calls))
+
+
+# -- 取消与清理 ----------------------------------------------------------
+class CancelCleanupTest(CloseupCase):
+    def test_cancel_vm_really_shuts_down(self):
+        ex = self.make_exec()
+        self._wake_ok(ex)
+        proc = self.proc
+        self.assertTrue(ex.cancel_vm(self._task()))
+        self.assertTrue(proc.terminated)
+        self.assertEqual(ex._vms, {})
+        self.assertIn("vm_shutdown", self._audit_events())
+        self.assertIn("vm_destroy", self._audit_events())
+
+    def test_shutdown_vm_socket_residual(self):
+        ex = self.make_exec()
+        os.makedirs(self.state_dir, exist_ok=True)
+        sock = os.path.join(self.state_dir, "chv-abc123.sock")
+        with open(sock, "w") as fh:
+            fh.write("")
+        self.assertTrue(ex.shutdown_vm(self._task()))
+        self.assertFalse(os.path.exists(sock))
+
+    def test_recycle_vm_success_and_missing_id(self):
+        ex = self.make_exec()
+        self._wake_ok(ex)
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "recycled")
+        self.assertEqual(out["preserved"],
+                         ["task_checkpoints", "task_attempts"])
+        self.assertIn("vm_recycled", self._audit_events())
+        with self.assertRaises(ValueError):
+            ex.recycle_vm({})
+
+    def test_cleanup_stale_vms_dead_socket_and_live_skip(self):
+        ex = self.make_exec()
+        os.makedirs(self.state_dir, exist_ok=True)
+        for name in ("chv-dead.sock", "chv-live.sock", "other.sock"):
+            with open(os.path.join(self.state_dir, name), "w") as fh:
+                fh.write("")
+        ex._vms["chv-live"] = FakeProc(alive=True)
+        cleaned = ex.cleanup_stale_vms()
+        self.assertEqual(cleaned, ["chv-dead"])
+        self.assertFalse(
+            os.path.exists(os.path.join(self.state_dir, "chv-dead.sock")))
+        self.assertTrue(
+            os.path.exists(os.path.join(self.state_dir, "other.sock")))
+
+    def test_cleanup_stale_vms_purges_dead_registry(self):
+        ex = self.make_exec()
+        os.makedirs(self.state_dir, exist_ok=True)
+        ex._vms["chv-ghost"] = FakeProc(alive=False)
+        self.assertEqual(ex.cleanup_stale_vms(), [])
+        self.assertEqual(ex._vms, {})
+
+    def test_cleanup_stale_vms_missing_state_dir(self):
+        ex = self.make_exec()
+        self.assertFalse(os.path.exists(self.state_dir))
+        self.assertEqual(ex.cleanup_stale_vms(), [])
+
+    def test_gc_stale_disks_deletes_stale_keeps_live(self):
+        ex = self.make_exec()
+        stale_dir = os.path.join(self.state_dir, "chv-stale")
+        live_dir = os.path.join(self.state_dir, "chv-live")
+        os.makedirs(stale_dir, exist_ok=True)
+        os.makedirs(live_dir, exist_ok=True)
+        with open(os.path.join(stale_dir, "disk.raw"), "wb") as fh:
+            fh.write(b"D")
+        with open(os.path.join(live_dir, "disk.raw"), "wb") as fh:
+            fh.write(b"D")
+        with open(os.path.join(self.state_dir, "chv-live.sock"),
+                  "w") as fh:
+            fh.write("")
+        out = ex.gc_stale_disks()
+        self.assertEqual(out["deleted"], ["chv-stale"])
+        self.assertFalse(
+            os.path.exists(os.path.join(stale_dir, "disk.raw")))
+        self.assertTrue(os.path.exists(os.path.join(live_dir, "disk.raw")))
+
+    def test_cleanup_stale_secrets_live_keeps_seed_dead_removes(self):
+        ex = self.make_exec()
+        live_dir = os.path.join(self.state_dir, "chv-live")
+        dead_dir = os.path.join(self.state_dir, "chv-dead")
+        os.makedirs(live_dir, exist_ok=True)
+        os.makedirs(dead_dir, exist_ok=True)
+        for vm_dir in (live_dir, dead_dir):
+            with open(os.path.join(vm_dir, "ssh_host_ed25519_key"),
+                      "w") as fh:
+                fh.write("priv")
+            with open(os.path.join(vm_dir, "seed.iso"), "wb") as fh:
+                fh.write(b"S")
+        with open(os.path.join(self.state_dir, "chv-live.sock"),
+                  "w") as fh:
+            fh.write("")
+        cleaned = ex.cleanup_stale_secrets()
+        self.assertEqual(sorted(cleaned), ["chv-dead", "chv-live"])
+        self.assertFalse(
+            os.path.exists(os.path.join(live_dir, "ssh_host_ed25519_key")))
+        self.assertTrue(os.path.exists(os.path.join(live_dir, "seed.iso")))
+        self.assertFalse(
+            os.path.exists(os.path.join(dead_dir, "ssh_host_ed25519_key")))
+        self.assertFalse(os.path.exists(os.path.join(dead_dir, "seed.iso")))
+
+    def test_stop_dirty_marker_on_cleanup_failure(self):
+        ex = self.make_exec()
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        seed_dir = os.path.join(vm_dir, "seed")
+        os.makedirs(seed_dir, exist_ok=True)
+        with open(os.path.join(seed_dir, "user-data"), "w") as fh:
+            fh.write("seed")
+        with open(os.path.join(self.state_dir, "chv-abc123.sock"),
+                  "w") as fh:
+            fh.write("")
+        # 本环境以 root 运行，chmod 000/0555 会被 root 绕过，
+        # 权限失败分支改用 rmtree 抛错模拟（只包 stop_vm 一次
+        # 调用，上下文退出即恢复，不影响 tearDown 清理）。
+        import shutil
+        real_rmtree = shutil.rmtree
+
+        def _failing_rmtree(path, *args, **kwargs):
+            if path == seed_dir:
+                raise OSError("mocked rmtree failure")
+            return real_rmtree(path, *args, **kwargs)
+        with mock.patch("shutil.rmtree", side_effect=_failing_rmtree):
+            self.assertTrue(ex.stop_vm(self._task()))
+        self.assertIn("vm_key_cleanup_failed", self._audit_events())
+        marker = dirty_marker_path(self.state_dir, "chv-abc123")
+        self.assertTrue(os.path.exists(marker))
+
+    def test_stop_clears_dirty_marker_on_success(self):
+        ex = self.make_exec()
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        os.makedirs(vm_dir, exist_ok=True)
+        marker = dirty_marker_path(self.state_dir, "chv-abc123")
+        os.makedirs(os.path.dirname(marker), exist_ok=True)
+        with open(marker, "w") as fh:
+            fh.write("cleanup failed\n")
+        with open(os.path.join(self.state_dir, "chv-abc123.sock"),
+                  "w") as fh:
+            fh.write("")
+        self.assertTrue(ex.stop_vm(self._task()))
+        self.assertFalse(os.path.exists(marker))
+
+
+# -- 中途死亡分类与只读 execute ------------------------------------------
+class VmDeathTest(CloseupCase):
+    def _wake_with_dead_on_dispatch(self, mode):
+        ex = self.make_exec()
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                proc = ex._vms.get("chv-abc123")
+                if proc is not None:
+                    proc._alive = False
+                if mode == "raise":
+                    raise RuntimeError("ssh broken")
+                return FakeResult(0, json.dumps(
+                    {"status": "failed", "text": "boom"}), "")
+            return self._default_runner(cmd, timeout)
+        ex.runner = runner
+        return ex
+
+    def test_run_with_vm_exception_vm_dead_interrupted(self):
+        ex = self._wake_with_dead_on_dispatch("raise")
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "interrupted")
+        self.assertTrue(out["vm_dead"])
+
+    def test_run_with_vm_failed_status_vm_dead_reclassified(self):
+        ex = self._wake_with_dead_on_dispatch("failed")
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "interrupted")
+        self.assertTrue(out["vm_dead"])
+        self.assertIn("VM 中途死亡", out["interrupted_reason"])
+
+    def test_run_with_vm_exception_vm_alive_reraises(self):
+        ex = self.make_exec()
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and "dispatch" in cmd:
+                raise RuntimeError("ssh broken")
+            return self._default_runner(cmd, timeout)
+        ex.runner = runner
+        with self.assertRaises(RuntimeError):
+            ex.dispatch_task(self._task())
+
+    def test_execute_dead_vm_reclassified(self):
+        ex = self.make_exec()
+        ex._vms["chv-abc123"] = FakeProc(alive=False)
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-abc123.sock"),
+                  "w") as fh:
+            fh.write("")
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            raise RuntimeError("ssh broken")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.execute(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_not_running")
+
+    def test_execute_not_running_no_runner_call(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.execute(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_not_running")
+        self.assertEqual(self.calls, [])
+
+
+# -- 入口校验 ------------------------------------------------------------
+class EntryGuardTest(CloseupCase):
+    def test_dispatch_run_goal_guards(self):
+        ex = self.make_exec()
+        with self.assertRaises(ValueError):
+            ex.dispatch_task({})
+        with self.assertRaises(ValueError):
+            ex.dispatch_task({"id": "x"})
+        with self.assertRaises(ValueError):
+            ex.run_task({"id": "x"})
+        with self.assertRaises(ValueError):
+            ex.goal_task({"id": "x", "goal": "g",
+                          "goal_max_rounds": 101})
+        with self.assertRaises(ValueError):
+            ex.goal_task({"id": "x", "goal": "g", "verify_command": "v",
+                          "goal_max_rounds": 0})
+        with self.assertRaises(ValueError):
+            ex.goal_resume_task({})
+        with self.assertRaises(ValueError):
+            ex.goal_resume_breakpoint_task({})
+        with self.assertRaises(ValueError):
+            ex.wake_vm({})
+
+    def test_dispatch_wake_failure_does_not_run_guest(self):
+        ex = self.make_exec(binary="/nonexistent/cloud-hypervisor")
+        with self.assertRaises(ChvFailure):
+            ex.dispatch_task(self._task())
+        self.assertFalse(any("dispatch" in c for c in self.calls))
+
+
+# -- preflight 事后改写校验 ------------------------------------------------
+class PreflightMutationTest(CloseupCase):
+    def test_reuse_blocked_and_tenant_undeclared(self):
+        ex = self.make_exec()
+        ex.disk_mode = "reuse"
+        ex.reuse_experiment = False
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "disk_mode_reuse_blocked")
+        ex.reuse_experiment = True
+        ex.single_tenant_project = None
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "reuse_tenant_undeclared")
+
+    def test_config_and_numeric_mutations(self):
+        cases = [
+            ("cpus", "not-an-int"),
+            ("memory", "bogus-mem"),
+            ("ssh_port", 99999),
+            ("max_vms", -1),
+            ("disk_gb", 0),
+            ("api_timeout", 0),
+            ("warm_pool_size", -1),
+            ("snapshot_max_count", -1),
+            ("host_key_mode", "bogus-mode"),
+        ]
+        for attr, value in cases:
+            with self.subTest(attr=attr):
+                ex = self.make_exec()
+                setattr(ex, attr, value)
+                with self.assertRaises(ChvFailure) as ctx:
+                    ex.preflight()
+                self.assertEqual(ctx.exception.kind, "config_invalid")
+
+    def test_cpu_flag_missing_and_kvm_unreadable(self):
+        with open(self.cpuinfo, "w") as fh:
+            fh.write("flags : fpu sse\n")
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "cpu_flag_missing")
+        with open(self.cpuinfo, "w") as fh:
+            fh.write("flags : fpu vmx sse\n")
+        ex = self.make_exec()
+        # 本环境以 root 运行，chmod 000 会被 root 绕过，
+        # kvm 不可读/写分支改用 os.access 模拟（只包本次
+        # preflight，上下文退出即恢复）。
+        real_access = os.access
+
+        def _fake_access(path, mode, *args, **kwargs):
+            if path == self.kvm and mode in (os.R_OK, os.W_OK):
+                return False
+            return real_access(path, mode, *args, **kwargs)
+        with mock.patch("os.access", side_effect=_fake_access):
+            with self.assertRaises(ChvFailure) as ctx:
+                ex.preflight()
+            self.assertEqual(ctx.exception.kind, "kvm_missing")
+
+    def test_net_dual_warns_but_passes(self):
+        ex = self.make_exec(env_extra={
+            "CHV_NET": "tap0",
+            "CHV_NET_JSON": '[{"tap": "tap0"}]'})
+        self.assertTrue(ex.preflight())
+        self.assertIn("net_config_dual", self._audit_events())
+
+    def test_net_json_invalid(self):
+        ex = self.make_exec(env_extra={"CHV_NET_JSON": "{not json"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "net_config_invalid")
+
+
+# -- worker_adapter 剩余分支 ------------------------------------------------
+class WorkerAdapterCloseupTest(unittest.TestCase):
+    def test_call_breakpoint_extra_env_forwarded(self):
+        seen = {}
+
+        class _Ex:
+            def goal_resume_breakpoint_task(self, task, model_provider=None,
+                                            extra_env=None):
+                seen["model_provider"] = model_provider
+                seen["extra_env"] = extra_env
+                return "ok"
+        out = call_goal_resume_breakpoint(
+            _Ex(), {"id": "x"}, model_provider="m",
+            extra_env={"A": "1"})
+        self.assertEqual(out, "ok")
+        self.assertEqual(seen["extra_env"], {"A": "1"})
+        out = call_goal_resume_breakpoint(_Ex(), {"id": "x"})
+        self.assertEqual(seen["extra_env"], None)
+
+    def test_call_breakpoint_signature_error_fallback(self):
+        class _BadSig:
+            __signature__ = "not-a-signature"
+
+            def __call__(self, task):
+                return "fallback-ok"
+
+        class _Ex:
+            def __init__(self):
+                self.goal_resume_breakpoint_task = _BadSig()
+        self.assertEqual(
+            call_goal_resume_breakpoint(_Ex(), {"id": "x"}), "fallback-ok")
+
+    def test_reconcile_non_chv_and_failure_audited(self):
+        self.assertEqual(reconcile_warm_pool(object(), lambda *a: None,
+                                             "when"), [])
+        audit = []
+        # CHV 失败分支：patch orphans 调和函数抛错，执行器
+        # 用 spec Mock 并把 __class__ 设为真执行器类，使
+        # isinstance 判定进入 CHV 分支。
+        from cloud.chv.executor import CloudHypervisorExecutor as _RealEx
+        self.assertTrue(inspect.isclass(_RealEx))
+        with mock.patch(
+                "cloud.chv.orphans.reconcile_orphans_for_executor",
+                side_effect=RuntimeError("boom")):
+            fake_ex = mock.Mock(spec=_RealEx)
+            # spec Mock 的 isinstance 判定走 __class__，需显式
+            # 把类设为真执行器类才能进 CHV 分支。
+            fake_ex.__class__ = _RealEx
+            out = reconcile_warm_pool(
+                fake_ex, lambda *a: audit.append(a), "when")
+        self.assertEqual(out, [])
+        self.assertTrue(any("reconcile_failed" in a[1] for a in audit))
+
+    def test_cancel_watchdog_cancel_vm_none(self):
+        stop = threading.Event()
+        requested = {"t1"}
+        cancel_watchdog_loop("t1", "chv-t1", stop, requested,
+                             lambda tid: None, None, object())
+        # 命中取消但 cancel_vm 为 None 时不炸、直接退出。
+
+    def test_quota_watchdog_not_over_then_stop(self):
+        stop = threading.Event()
+
+        def check():
+            stop.set()
+            return False
+        calls = []
+        quota_watchdog_loop("t1", "chv-t1", stop, check,
+                            lambda tid: calls.append(tid), None,
+                            object(), interval=0.01)
+        self.assertEqual(calls, [])
+
+    def test_watchdog_vm_name_non_callable(self):
+        self.assertIsNone(watchdog_vm_name(object(), {"id": "x"}))
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+# -- 收尾第二批：防御性 except 与文件/入口变体 --------------------------
+class RaisingProc:
+    def poll(self):
+        raise RuntimeError("poll boom")
+
+    def terminate(self):
+        raise RuntimeError("terminate boom")
+
+    def wait(self, timeout=None):
+        raise RuntimeError("wait boom")
+
+    def kill(self):
+        raise RuntimeError("kill boom")
+
+
+class DefensiveBranchTest(CloseupCase):
+    def test_default_popen_launcher_runs_true(self):
+        from cloud.chv.executor import _default_popen_launcher
+        proc = _default_popen_launcher(["true"])
+        self.assertEqual(proc.wait(), 0)
+
+    def test_binary_ok_bare_name_via_which(self):
+        ex = self.make_exec(binary="true", ch_remote="true")
+        self.assertTrue(ex.preflight())
+
+    def test_kill_proc_none_and_raising(self):
+        ex = self.make_exec()
+        ex._kill_proc(None)
+        ex._kill_proc(RaisingProc())  # terminate/wait/kill 全抛也不炸
+
+    def test_active_and_persistent_count_poll_exception(self):
+        ex = self.make_exec()
+        self.assertEqual(ex._active_count(), 0)
+        ex._vms["chv-bad"] = RaisingProc()
+        self.assertEqual(ex._active_count(), 1)
+        self.assertEqual(ex._persistent_vm_count(), 0)
+
+    def test_preflight_cpuinfo_missing_and_unset_files(self):
+        ex = self.make_exec(cpuinfo_path="/nonexistent/cpuinfo")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "cpu_flag_missing")
+        for kwargs in ({"kernel": None}, {"rootfs": None},
+                       {"ssh_key": None}):
+            with self.subTest(kwargs=kwargs):
+                ex = self.make_exec(**kwargs)
+                with self.assertRaises(ChvFailure):
+                    ex.preflight()
+
+    def test_preflight_unreadable_files_via_access_mock(self):
+        ex = self.make_exec()
+        real_access = os.access
+
+        def _fake_access(path, mode, *args, **kwargs):
+            if path in (self.kernel, self.rootfs, self.ssh_key) \
+                    and mode == os.R_OK:
+                return False
+            return real_access(path, mode, *args, **kwargs)
+        with mock.patch("os.access", side_effect=_fake_access):
+            with self.assertRaises(ChvFailure) as ctx:
+                ex.preflight()
+        self.assertIn(ctx.exception.kind,
+                      ("kernel_missing", "image_missing",
+                       "ssh_key_missing"))
+
+    def test_preflight_state_dir_is_file(self):
+        blocker = os.path.join(self.dir, "blocker")
+        with open(blocker, "w") as fh:
+            fh.write("x")
+        ex = self.make_exec(state_dir=blocker)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.preflight()
+        self.assertEqual(ctx.exception.kind, "state_dir_insecure")
+
+    def test_build_vm_command_with_net_and_config_errors(self):
+        ex = self.make_exec(net="tap0")
+        cmd = ex._build_vm_command(self._task())
+        self.assertIn("--net", cmd)
+        self.assertIn("tap0", cmd)
+        ex.cpus = "bogus"
+        with self.assertRaises(ValueError):
+            ex.build_vm_config()
+        ex.cpus = 2
+        ex.net_json = "{bad"
+        with self.assertRaises(ValueError):
+            ex.build_vm_config()
+
+    def test_net_guest_ip_invalid_and_ssh_target_undeclared(self):
+        ex = self.make_exec()
+        ex.net_json = "{bad"
+        self.assertIsNone(ex._net_guest_ip())
+        ex2 = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"})
+        with self.assertRaises(ChvFailure) as ctx:
+            ex2.resolve_ssh_host()
+        self.assertEqual(ctx.exception.kind, "ssh_target_undeclared")
+
+    def test_capture_host_key_resolve_and_runner_and_write_failures(self):
+        ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"})
+        self.assertFalse(ex.capture_host_key(self._task()))
+        ex = self.make_exec()
+
+        def raising_runner(cmd, timeout=None):
+            raise RuntimeError("keyscan boom")
+        ex.runner = raising_runner
+        self.assertFalse(ex.capture_host_key(self._task()))
+        ex = self.make_exec()
+        blocker = os.path.join(self.state_dir, "chv-abc123")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(blocker, "w") as fh:
+            fh.write("x")
+
+        def keyscan_runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh-keyscan":
+                return FakeResult(0, "[127.0.0.1]:2222 ssh-ed25519 AAAA-good", "")
+            return FakeResult(0, "", "")
+        ex.runner = keyscan_runner
+        self.assertFalse(ex.capture_host_key(self._task()))
+
+    def test_build_ssh_command_fallback_and_secret_not_delivered(self):
+        ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"})
+        cmd = ex.build_ssh_command(self._task(), ["true"])
+        self.assertIn("127.0.0.1", " ".join(cmd))
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.build_ssh_command(self._task(), ["true"],
+                                 extra_env={"SUB2API_API_KEY": "v"})
+        self.assertEqual(ctx.exception.kind, "secret_env_not_delivered")
+
+    def test_write_provisioned_known_hosts_fallback(self):
+        ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0",
+                                       "CHV_HOST_KEY_MODE": "provisioned"})
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        os.makedirs(vm_dir, exist_ok=True)
+        paths = {"known_hosts": os.path.join(vm_dir, "known_hosts")}
+        ex._write_provisioned_known_hosts(paths, "ssh-ed25519 AAAA")
+        with open(paths["known_hosts"]) as fh:
+            self.assertIn("127.0.0.1", fh.read())
+
+    def test_provision_host_key_dirty_refused(self):
+        ex = self.make_exec()
+        marker = dirty_marker_path(self.state_dir, "chv-abc123")
+        os.makedirs(os.path.dirname(marker), exist_ok=True)
+        with open(marker, "w") as fh:
+            fh.write("cleanup failed\n")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.provision_host_key(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_dirty_refused")
+
+    def test_api_call_raises_then_api_timeout(self):
+        def api(socket_path, method, path, payload=None):
+            raise ConnectionError("socket gone")
+        ex = self.make_exec(api_client=api, api_timeout=2)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_api_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "api_timeout")
+
+    def test_wait_guest_runner_exception_retried_then_timeout(self):
+        self._known_hosts()
+        ex = self.make_exec(ready_timeout=2, ready_interval=1)
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh":
+                raise RuntimeError("ssh boom")
+            return FakeResult(0, "", "")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "guest_not_ready")
+
+    def test_stop_exceptions_swallowed_around_stop_vm(self):
+        # wait_guest_ready 的 provisioned 守卫、mismatch 分支、
+        # wake_vm 与 restore_vm 的 stop_vm 外层 except：stop_vm
+        # 本身不抛，改用 mock 让它抛，验证原始失败不被掩盖。
+        ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
+        ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom"))
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
+
+        def api(socket_path, method, path, payload=None):
+            if method == "PUT" and path == "/vm.create":
+                return (500, {})
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom"))
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_create_failed")
+
+        url, _ = self._snap_url_for(ex)
+        ex = self.make_exec()
+        ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom"))
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == self.ch_remote and "restore" in cmd:
+                return FakeResult(1, "", "")
+            return self._default_runner(cmd, timeout)
+        ex.runner = runner
+        with self.assertRaises(RuntimeError):
+            ex.restore_vm(self._task(), url)
+
+    def _snap_url_for(self, ex):
+        path = os.path.join(self.state_dir, "snapx.raw")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(b"S" * 10)
+        return "file://" + path, path
+
+    def test_registered_net_entries_variants(self):
+        ex = self.make_exec()
+        os.makedirs(os.path.join(self.state_dir, "chv-skip"),
+                    exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-skip", "net.json"),
+                  "w") as fh:
+            json.dump([{"tap": "tap9", "ip": "10.0.0.9"}], fh)
+        os.makedirs(os.path.join(self.state_dir, "chv-bad"),
+                    exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-bad", "net.json"),
+                  "w") as fh:
+            fh.write('{"not": "a list"}')
+        os.makedirs(os.path.join(self.state_dir, "chv-empty"),
+                    exist_ok=True)
+        out = ex._registered_net_entries(exclude_vm="chv-skip")
+        self.assertEqual(out, [])
+        ex2 = self.make_exec(state_dir="/nonexistent-state-dir-x")
+        self.assertEqual(ex2._registered_net_entries(), [])
+
+    def test_socket_permissions_stat_failure(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("")
+        real_stat = os.stat
+        with mock.patch("os.stat", side_effect=OSError("stat boom")):
+            ex._check_api_socket_permissions(sock)  # 不抛，只跳过
+        self.assertTrue(real_stat(sock) is not None)
+
+    def test_rewake_kills_old_proc_and_start_vm_alias(self):
+        ex = self.make_exec()
+        self._wake_ok(ex)
+        old = self.proc
+        out = ex.start_vm(self._task())
+        self.assertEqual(out["mode"], "cold")
+        self.assertTrue(old.terminated)
+
+    def test_validate_restore_guards(self):
+        ex = self.make_exec()
+        with self.assertRaises(ValueError):
+            ex._validate_restore({}, "file://x")
+        with self.assertRaises(ValueError):
+            ex._validate_restore(self._task(), "")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._validate_restore(self._task(), "snap.raw")
+        self.assertEqual(ctx.exception.kind, "snapshot_path_not_allowed")
+
+    def test_stop_registered_proc_poll_exception(self):
+        ex = self.make_exec()
+        ex._vms["chv-bad"] = RaisingProc()
+        ex._stop_registered_proc("chv-bad", RaisingProc())
+        self.assertEqual(ex._vms, {})
+
+    def test_cleanup_net_remove_failure_marks_dirty(self):
+        ex = self.make_exec()
+        vm_dir = os.path.join(self.state_dir, "chv-abc123")
+        os.makedirs(vm_dir, exist_ok=True)
+        os.makedirs(os.path.join(vm_dir, "net.json"), exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-abc123.sock"),
+                  "w") as fh:
+            fh.write("")
+        self.assertTrue(ex.stop_vm(self._task()))
+        marker = dirty_marker_path(self.state_dir, "chv-abc123")
+        self.assertTrue(os.path.exists(marker))
+
+    def test_write_dirty_marker_failure_swallowed(self):
+        blocker = os.path.join(self.dir, "blocker2")
+        with open(blocker, "w") as fh:
+            fh.write("x")
+        ex = self.make_exec(state_dir=blocker)
+        ex._write_dirty_marker("chv-abc123", True)  # 不抛
+
+    def test_stop_socket_remove_failure_swallowed(self):
+        ex = self.make_exec()
+        os.makedirs(os.path.join(self.state_dir, "chv-abc123.sock"),
+                    exist_ok=True)
+        self.assertTrue(ex.stop_vm(self._task()))
+
+    def test_gc_poll_exception_and_delete_failure(self):
+        ex = self.make_exec()
+        vm_dir = os.path.join(self.state_dir, "chv-bad")
+        os.makedirs(vm_dir, exist_ok=True)
+        with open(os.path.join(vm_dir, "disk.raw"), "wb") as fh:
+            fh.write(b"D")
+        ex._vms["chv-bad"] = RaisingProc()
+        out = ex.gc_stale_disks()
+        self.assertEqual(out["deleted"], ["chv-bad"])
+        vm2 = os.path.join(self.state_dir, "chv-dir")
+        os.makedirs(os.path.join(vm2, "disk.raw"), exist_ok=True)
+        out = ex.gc_stale_disks()
+        self.assertEqual(out["failed"], ["chv-dir"])
+        self.assertIn("vm_disk_delete_failed", self._audit_events())
+
+    def test_is_live_and_proc_dead_poll_exception(self):
+        ex = self.make_exec()
+        ex._vms["chv-bad"] = RaisingProc()
+        self.assertFalse(ex._is_live_vm("chv-bad"))
+        self.assertFalse(ex._vm_proc_dead("chv-bad"))
+        self.assertFalse(ex._vm_is_running("chv-bad"))
+
+    def test_clean_one_vm_secrets_remove_failures(self):
+        ex = self.make_exec()
+        vm_dir = os.path.join(self.state_dir, "chv-x")
+        os.makedirs(os.path.join(vm_dir, "ssh_host_ed25519_key"),
+                    exist_ok=True)
+        removed_priv, removed_seed = ex._clean_one_vm_secrets(
+            "chv-x", vm_dir)
+        self.assertFalse(removed_priv)
+        vm2 = os.path.join(self.state_dir, "chv-y")
+        os.makedirs(os.path.join(vm2, "seed"), exist_ok=True)
+        import shutil
+        real_rmtree = shutil.rmtree
+        with mock.patch("shutil.rmtree",
+                        side_effect=OSError("rmtree boom")):
+            removed_priv, removed_seed = ex._clean_one_vm_secrets(
+                "chv-y", vm2)
+        self.assertFalse(removed_seed)
+        self.assertTrue(real_rmtree is not None)
+
+    def test_cleanup_stale_vms_subcall_failures_and_poll_branches(self):
+        ex = self.make_exec()
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "notes.txt"),
+                  "w") as fh:
+            fh.write("not a socket")
+        with open(os.path.join(self.state_dir, "chv-dead.sock"),
+                  "w") as fh:
+            fh.write("")
+        ex._vms["chv-dead"] = RaisingProc()
+        ex.cleanup_stale_secrets = mock.Mock(
+            side_effect=RuntimeError("secrets boom"))
+        ex.gc_stale_disks = mock.Mock(
+            side_effect=RuntimeError("gc boom"))
+        cleaned = ex.cleanup_stale_vms()
+        self.assertEqual(cleaned, ["chv-dead"])
+        ex2 = self.make_exec()
+        ex2._vms["chv-ghost"] = RaisingProc()
+        ex2._purge_dead_registry()
+        self.assertEqual(ex2._vms, {})
+
+    def test_run_with_vm_finally_shutdown_exception_swallowed(self):
+        ex = self.make_exec()
+        real_shutdown = ex.shutdown_vm
+        ex.shutdown_vm = mock.Mock(side_effect=RuntimeError("sd boom"))
+        # _run_with_vm 的 finally 里 shutdown 抛错应被吞掉，
+        # dispatch 仍返回 completed（finally 不改返回值）。
+        out = ex.dispatch_task(self._task())
+        self.assertEqual(out["status"], "completed")
+        ex.shutdown_vm = real_shutdown
+
+    def test_execute_guards_and_alive_reraise(self):
+        ex = self.make_exec()
+        with self.assertRaises(ValueError):
+            ex.execute({})
+        ex._vms["chv-abc123"] = FakeProc(alive=True)
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, "chv-abc123.sock"),
+                  "w") as fh:
+            fh.write("")
+
+        def runner(cmd, timeout=None):
+            raise RuntimeError("ssh boom")
+        ex.runner = runner
+        with self.assertRaises(RuntimeError):
+            ex.execute(self._task())
+
+    def test_entry_success_branches(self):
+        ex = self.make_exec()
+        out = ex.dispatch_task({**self._task(),
+                                "repo": "https://example.com/r.git",
+                                "plan": True})
+        self.assertEqual(out["status"], "completed")
+        ssh_cmds = [c for c in self.calls if c and c[0] == "ssh"]
+        self.assertTrue(any("https://example.com/r.git" in " ".join(c)
+                            for c in ssh_cmds))
+        self.calls.clear()
+        out = ex.run_task(self._task())
+        self.assertEqual(out["status"], "completed")
+        with self.assertRaises(ValueError):
+            ex.run_task({})
+        with self.assertRaises(ValueError):
+            ex.goal_task({"id": "x"})
+        with self.assertRaises(ValueError):
+            ex.goal_task({})
+        out = ex.goal_task({**self._task(), "goal": "g",
+                            "verify_command": "make check",
+                            "repo": "https://example.com/r.git"})
+        self.assertEqual(out["status"], "completed")
+        goal_cmds = [c for c in self.calls
+                     if c and c[0] == "ssh" and "goal" in c]
+        self.assertTrue(any("make check" in " ".join(c)
+                            for c in goal_cmds))
+        self.calls.clear()
+        out = ex.goal_resume_task(self._task(), goal_text="new goal",
+                                  max_rounds=2)
+        self.assertEqual(out["status"], "completed")
+        resume_cmds = [c for c in self.calls
+                       if c and c[0] == "ssh" and "goal-resume" in c]
+        self.assertTrue(any("--goal-text" in c for c in resume_cmds))
+        self.calls.clear()
+        out = ex.goal_resume_task(self._task(), max_rounds=5)
+        self.assertEqual(out["status"], "completed")
+        resume_cmds = [c for c in self.calls
+                       if c and c[0] == "ssh" and "goal-resume" in c]
+        self.assertTrue(any("--max-rounds" in c for c in resume_cmds))
+
+
+class WorkerAdapterCloseup2Test(unittest.TestCase):
+    def test_is_chv_executor(self):
+        from cloud.chv.worker_adapter import is_chv_executor
+        self.assertFalse(is_chv_executor(object()))
+
+    def test_reconcile_audit_fn_failures_swallowed(self):
+        from cloud.chv.executor import CloudHypervisorExecutor as _RealEx
+        from cloud.chv.worker_adapter import reconcile_warm_pool
+        with mock.patch(
+                "cloud.chv.orphans.reconcile_orphans_for_executor",
+                side_effect=RuntimeError("boom")):
+            fake_ex = mock.Mock(spec=_RealEx)
+            fake_ex.__class__ = _RealEx
+
+            def raising_audit(*args):
+                raise RuntimeError("audit boom")
+            out = reconcile_warm_pool(fake_ex, raising_audit, "when")
+        self.assertEqual(out, [])
+        with mock.patch(
+                "cloud.chv.orphans.reconcile_orphans_for_executor",
+                return_value=["chv-x"]):
+            fake_ex = mock.Mock(spec=_RealEx)
+            fake_ex.__class__ = _RealEx
+            out = reconcile_warm_pool(fake_ex, raising_audit, "when")
+        self.assertEqual(out, ["chv-x"])
+
+    def test_cancel_watchdog_get_task_variants(self):
+        from cloud.chv.worker_adapter import cancel_watchdog_loop
+        stop = threading.Event()
+
+        def get_task_raising(tid):
+            stop.set()
+            raise RuntimeError("db boom")
+        cancel_watchdog_loop("t1", "chv-t1", stop, set(),
+                             get_task_raising, None, object())
+        stop2 = threading.Event()
+
+        def get_task_running(tid):
+            stop2.set()
+            return {"status": "running"}
+        cancel_watchdog_loop("t1", "chv-t1", stop2, set(),
+                             get_task_running, None, object())
+
+    def test_cancel_watchdog_cancel_vm_raises_swallowed(self):
+        from cloud.chv.worker_adapter import cancel_watchdog_loop
+        stop = threading.Event()
+
+        def bad_cancel(vm):
+            raise RuntimeError("cancel boom")
+        cancel_watchdog_loop("t1", "chv-t1", stop, {"t1"},
+                             lambda tid: None, bad_cancel, object())
+
+    def test_quota_watchdog_check_raises_swallowed(self):
+        from cloud.chv.worker_adapter import quota_watchdog_loop
+        stop = threading.Event()
+
+        def bad_check():
+            stop.set()
+            raise RuntimeError("quota boom")
+        quota_watchdog_loop("t1", "chv-t1", stop, bad_check,
+                            lambda tid: None, None, object(),
+                            interval=0.01)
+
+
+class ExecutorLastLinesTest(CloseupCase):
+    def test_net_guest_ip_no_ip_entry_returns_none(self):
+        ex = self.make_exec()
+        ex.net_json = '[{"tap": "tap0"}]'
+        self.assertIsNone(ex._net_guest_ip())
+
+    def test_wait_guest_mismatch_stop_exception_swallowed(self):
+        self._known_hosts()
+        ex = self.make_exec()
+        ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom"))
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh" and cmd[-1] == "true":
+                return FakeResult(1, "",
+                                  "Host key verification failed")
+            return FakeResult(0, "", "")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "host_key_mismatch")
+
+    def test_wait_guest_timeout_stop_exception_swallowed(self):
+        self._known_hosts()
+        ex = self.make_exec(ready_timeout=2, ready_interval=1)
+        ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom"))
+
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh":
+                return FakeResult(1, "", "")
+            return FakeResult(0, "", "")
+        ex.runner = runner
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wait_guest_ready(self._task())
+        self.assertEqual(ctx.exception.kind, "guest_not_ready")
+
+    def test_record_net_entries_write_failure_only_warns(self):
+        ex = self.make_exec()
+        blocker = os.path.join(self.state_dir, "chv-x")
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(blocker, "w") as fh:
+            fh.write("x")
+        with self.assertLogs("cloud.cloud_hypervisor",
+                             level="WARNING") as logs:
+            ex._record_net_entries("chv-x", [{"tap": "tap0"}])
+        self.assertTrue(any("net_registry_write_failed" in line
+                            for line in logs.output))
+
+    def test_socket_permissions_stat_failure_after_exists(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("")
+        real_stat = os.stat
+        calls = {"n": 0}
+
+        def _stat_then_raise(path, *args, **kwargs):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                return real_stat(path, *args, **kwargs)
+            raise OSError("stat boom")
+        # 第 1 次 stat 是 exists() 的探查（须成功才进到显式
+        # stat），第 2 次显式 stat 抛错 → except 分支 return。
+        with mock.patch("os.stat", side_effect=_stat_then_raise):
+            ex._check_api_socket_permissions(sock)
+
+    def test_vm_proc_dead_unregistered_returns_false(self):
+        ex = self.make_exec()
+        self.assertFalse(ex._vm_proc_dead(self._task()))
diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md
index 6985ea2..97b4db1 100644
--- a/docs/chv-production-readiness.md
+++ b/docs/chv-production-readiness.md
@@ -572,3 +572,216 @@
 6. 负载测试（V-8）与安全审计结论更新（No-Go → 有条件启用）。
 7. 前端 live 点击验证（用户侧 live browser 检查清单见
    本 PR 描述；回环地址限制下浏览器任务无法代劳）。
+
+## 14. 收尾轨 feat-chv-closeup（堆叠于 #264 之上）
+
+本节是准备工作，不是启用结论。本轨只到 PR OPEN，不合并、
+不部署；生产就绪与动态安全的外部关卡（§13.4）原样未动。
+
+### 14.1 本轨量化结果（实测，非上一轨回执转述）
+
+- 后端：232 项 → 317 项（新增 `cloud/tests/test_chv_closeup.py`
+  85 项），全过。命令（仓库根目录，cwd 须为根目录——
+  在 `cloud/` 下跑会撞 `cloud/queue.py` 遮蔽 stdlib queue，
+  mypy 同理）：
+  `python3 -m unittest cloud.tests.test_chv_arch95
+  cloud.tests.test_chv_arch95plus cloud.tests.test_chv_arch_maint_fe
+  cloud.tests.test_chv_fix cloud.tests.test_chv_gaps
+  cloud.tests.test_chv_optimize cloud.tests.test_chv_secfix
+  cloud.tests.test_chv_to10 cloud.tests.test_cloud_hypervisor
+  cloud.tests.test_chv_closeup`
+- 覆盖率（同命令 + `coverage run --source=cloud/chv`）：
+  chv 总 87%（1925 行，缺 258）→ 96%（缺 80）；
+  `executor.py` 85%（缺 157）→ 100%（缺 0）；
+  `worker_adapter.py` 82%（缺 16）→ 100%（缺 0）。
+  本轨基线与上一轨回执的行数口径略有出入（上一轨称
+  1826 行/缺 242，本轨同命令实测 1925 行/缺 258），
+  本轨前后对比只用本轨同一命令，不混用上一轨数字。
+- 剩余未覆 80 行（按模块，全部在 executor/worker_adapter
+  之外，本轨未声称覆盖）：
+  `api_client.py` 11 行（socket 层 chunked/超时/异常分支，
+  需 socket 级故障注入）；`compat215.py` 2 行（旧兼容回落）；
+  `config.py` 5 行（非法 env 构造分支）；`disk.py` 22 行
+  （快照登记/剪枝边角与配额异常分支）；`experimental_pool.py`
+  16 行（实验池，非生产路径）；`hardening.py` 11 行
+  （真实 socket 权限与 net 校验边角）；`hostkeys.py` 13 行
+  （ssh-keygen/genisoimage 子进程失败分支）。以上均非死
+  代码的断言——未逐行证伪，留待后续轨按模块补，不在本
+  轨凑数。
+- mypy：silent 口径 `cloud/chv/` 0 错（不变）；
+  默认口径 `cloud/marker_tools.py` 仍 2 错（见 §14.3，
+  本轨未改该文件）。
+- ruff：`--select F,E9,W,I` 口径 `cloud/chv/` +
+  本轨新测试 0 错（不变；默认全规则基线不声称通过）。
+- 前端：Vitest 23 项 → 36 项（新增
+  `src/api/chv.closeup.test.ts` 13 项），全过。
+
+### 14.2 审计落盘文件运维 runbook（ADR-14 取舍不变，只补可操作性）
+
+- 定位：`<state_dir>/chv-audit-spill.jsonl` 与轮转文件
+  `<state_dir>/chv-audit-spill.jsonl.old`。`state_dir`
+  默认 `~/.local/share/codex-app-server/chv-vms`，
+  可由 `CHV_STATE_DIR` 改址；以部署的实际值为准，
+  不要猜。目录本身 0700（见 §13.1 的 H6 门）。
+- 触发：只有 `_emit` 的审计回调抛错时才写一行；回调
+  成功、无回调（只记日志）时不写。触发时宿主日志必有
+  `chv_audit_failed`，且 `audit_failure_count` +1——
+  落盘行是“哪次事件丢了”的唯一取证，计数只是数字。
+- 格式：JSONL，每行 `{ts, event, kind, detail, task_id}`。
+  `detail` 沿用 `_emit` 契约，只含 VM 名/路径/URL/kind，
+  绝不含密钥值；读文件时仍按敏感文件对待（0700 目录内）。
+- 轮转：当前文件 >1MB 时，先 `os.replace` 成 `.old`
+  再写新行；只留最新两代，更早的丢失事件会丢——
+  这是 ADR-14 已定的取舍，不是 bug，不要指望它当
+  完整审计账本。
+- 读取（宿主侧，只读）：
+  `cat <state_dir>/chv-audit-spill.jsonl`；
+  按事件过滤：`jq -c 'select(.event=="vm_failed")'
+  <state_dir>/chv-audit-spill.jsonl`；
+  两代一起看时先 `.old` 后当前文件（时间顺序）。
+- 清理：`.old` 可随时删（删的是已被轮转覆盖的旧代）；
+  当前 `.jsonl` 只有在已人工核对、且确认丢失事件已
+  无需追查时才删——删了就再也找不回那批事件。
+  落盘文件不是 DB 替代：任务 timeline 查询看不到它，
+  不要把它当成审计表去对账；审计完整性本身仍无保证
+  （见 ADR-14），保证需要 DB 事务级审计写入，那是
+  worker 全局语义改动，不在本轨。
+- 落盘写失败：只记 `chv_audit_spill_failed` 日志，
+  绝不中断任务；此时连落盘都没有，唯一信号就是这条
+  日志与计数。
+
+### 14.3 mypy 默认口径剩余 2 错评估（结论：本轨不修）
+
+- 位置：`cloud/marker_tools.py:142`，`try: from zoneinfo
+  import ZoneInfo / except Exception: ZoneInfo = None`。
+  mypy 把 import 推断为 `type[ZoneInfo]`，再把 `None`
+  赋值判为两错（misc + assignment）。
+- 仅标注修法（沙箱验证过，未进仓库）：在 import 区后加
+  `from typing import Any` 与 `ZoneInfo: Any`（标注语句
+  本身不产生绑定），默认口径 2 错 → 0 错；运行时
+  `ZoneInfo` 仍是 stdlib `zoneinfo.ZoneInfo` 同一对象，
+  行为不变。验证在 `/tmp` 副本上做，不碰仓库文件。
+- 不修的理由（两条，任一条已足够）：① 该文件属标记
+  工具链（git 历史为标记工具提交），不在本链 §13.1
+  文件清单内，本轨改它就是动其它链的文件，徒增与
+  仍 OPEN 的审计/标记 PR 的 rebase 冲突面；② CHV 链
+  自己的门是 silent 口径（`--follow-imports=silent
+  cloud/chv/`），该口径已 0 错，默认口径的 2 错是
+  作用域外既有问题。修法留给 marker_tools 的归属轨，
+  本轨只把配方写在这里，不为清零越界改文件。
+
+### 14.4 合并前检查脚本（只读）
+
+- 脚本：`tools/chv_merge_precheck.py`，用法见文件头。
+  只读：`gh pr view` / `gh api` 查询 + 本地 `git` 读
+  remote，不 fetch、不 checkout、不 merge、不改 CI。
+  退出码：0 就绪、1 未就绪、2 查询失败（结论不可信）。
+- 本轨实测（未含本轨新 PR，建好后须追加新 PR 号重跑）：
+  `python3 tools/chv_merge_precheck.py` → 退出码 1，
+  结论 NOT-READY：11 个 PR 全部 OPEN / MERGEABLE，
+  链衔接全部 OK，当前 main 头仍 `aebc6238`（与 §13.1
+  一致），但 11 个 PR 的 CI 全部 `FAIL [verify:FAILURE]`、
+  mergeState 全部 UNSTABLE。脚本不诊断失败原因，
+  不把秒级失败改判——账单恢复变绿前结论保持 NOT-READY。
+
+### 14.5 灰度三阶段可执行清单（准备工作，未执行、未验证）
+
+通用停止条件（任一触发即停在该阶段，不进下一阶段）：
+`vm_failed` 审计激增、脏标记（`dirty_marker`）出现、
+`chv-audit-spill.jsonl` 持续增长、残留盘/残留密钥扫描
+清出本阶段不应有的项、任一任务静默降级到 Docker/Local。
+通用回滚：停止向 CHV 路由新任务（改回 docker/local
+执行器选择），已起 VM 走取消/关闭路径（`cancel_vm` /
+`shutdown_vm`）；代码回滚另议（revert 合并），不与
+“停路由”混为一谈。
+
+- 阶段 0（合并后、默认关闭）：
+  1. 按 §14.4 脚本逐项核对，退出码 0 前不合并；合并
+     顺序即 §14.4 的链序，不得跳号。
+  2. 合并后确认 CHV 仍默认关闭（无生产任务路由到
+     `cloud-hypervisor`，任务创建白名单外的执行器值
+     仍被后端拒绝）。
+  3. 在 KVM 宿主（须见 `/dev/kvm`，本 VM 与生产宿主
+     均无 KVM，不能替代）上跑 §14.6 的 V-1～V-9，
+     全部通过才可谈阶段 1；任一未跑即记未验证。
+  回滚点：合并本身可 revert；未进生产路由前无任务影响。
+- 阶段 1（单租户可信项目）：
+  1. 只对明确指定的单租户项目派发 CHV 任务；复用
+     实验仍需 `CHV_REUSE_SINGLE_TENANT_EXPERIMENT` 与
+     `CHV_SINGLE_TENANT_PROJECT` 同时声明，缺一即
+     preflight 拒绝（代码行为，不放宽）。
+  2. 检查点：每任务 VM 销毁后 `state_dir` 无该任务
+     残留盘与 seed；`chv-audit-spill.jsonl` 无新增；
+     取消路径真关 VM（V-5 口径）。
+  3. 停止条件见通用条件；回滚：该项目任务改回原
+     执行器，CHV worker 停接新任务。
+- 阶段 2（受限多租户）：
+  1. 放开到受邀项目，连续观察一周：VM 残留盘、
+     审计落盘、容量门（`CHV_MAX_VMS`，含重启后残留
+     socket 扫描）、快照配额四项指标。
+  2. 检查点：每周复核四项指标零异常 + §14.2 落盘
+     文件复核；任一指标异常即退回阶段 1。
+  3. 停止条件见通用条件；回滚：收紧到阶段 1 项目集。
+- 阶段 3（全量多租户）：仅在阶段 2 指标干净 +
+  动态安全审计结论由 No-Go 改为有条件启用后，才可
+  讨论；绝不从阶段 0 直跳全量。本轨不给阶段 3 的
+  时间表，不写成已就绪。
+
+### 14.6 真机验证用例 V-1～V-9 验收检查表（全部未验证）
+
+每行格式：用例 — 验收标准（§13.2 细化）— 证据栏（待填）。
+状态栏一律“未验证”，跑过才改，跑过≠通过。
+
+- [ ] V-1 冷启动 — 任务 completed，宿主 state_dir 无该
+  任务残留盘与 seed。证据：______ 状态：未验证
+- [ ] V-2 seed 首启时序 — `sshd -T` 显示
+  PasswordAuthentication no 等加固项，known_hosts 钉死
+  校验通过，非 TOFU 回落。证据：______ 状态：未验证
+- [ ] V-3 快照恢复 — 恢复后 known_hosts 不变，快照文件
+  落盘非空且经 `snapshot_not_persisted` 校验。证据：______
+  状态：未验证
+- [ ] V-4 网关真实任务 — 任务完成 + 宿主进程表抽查无密钥
+  （密钥只走 seed 客内文件）。证据：______ 状态：未验证
+- [ ] V-5 取消与配额真杀 — VM 进程消失，任务 interrupted
+  且原因写回正确。证据：______ 状态：未验证
+- [ ] V-6 审计落盘 — 落盘行 event/task_id/detail 与丢失
+  事件一致，`audit_failure_count` +1，任务不中断。
+  证据：______ 状态：未验证
+- [ ] V-7 动态安全负向 — 直连被 tap/防火墙挡、跨 VM 探活
+  与 IP 冒用被 SSH 钉死拒、崩溃注入后残留被清理或脏
+  标记拒起，每项都有“被挡住”的证据。证据：______
+  状态：未验证
+- [ ] V-8 负载 — 4 VM 并发下容量门计数含重启后残留
+  socket 扫描，死 socket 不占名额、活 VM 不被超发，
+  无静默降级。证据：______ 状态：未验证
+- [ ] V-9 供应链清单 — 构建宿主产出真实 lockfile，
+  `--verify` 通过且 unresolved 为空，再用
+  `check_lockfile.py` 对 BUILDINFO.versions 复核。
+  证据：______ 状态：未验证
+
+### 14.7 本轨逐项评分（目标 10，无一项硬写 10）
+
+- 代码质量 9.8 → 9.8，未达 10：executor/worker_adapter
+  覆盖到 100%，但 chv 仍有 80 行未覆（非 executor
+  模块），且默认口径 mypy 余 2 错在作用域外。差：代码内。
+- 架构 9.7 → 9.7，未达 10：本轨未改架构，ADR-13/14
+  取舍不变。差：外部（生产验证后才可复评）。
+- 可维护性 9.7 → 9.8，未达 10：新增合并检查脚本与
+  §14 运维/灰度文档，落盘可操作性补齐；但跨链文件
+  的 mypy 归属未落。差：代码内（小）。
+- 文档 9.8 → 9.9，未达 10：§14 补齐 runbook、灰度
+  清单、V 检查表、评分表；仍缺真机证据填表。差：外部。
+- 测试 9.7 → 9.8，未达 10：后端 317 项、前端 36 项全过，
+  executor 100%；但 chv 总覆盖 96% 未满，且真机用例
+  0 项已跑。差：代码内（80 行）+ 外部（真机）。
+- 静态安全 9.6 → 9.6，未达 10：取舍不变，只补落盘
+  运维；lockfile 仍待构建宿主产出，绝不编造。差：外部。
+- 前端 9.5 → 9.6，未达 10：只读四态与可用性文案边角
+  全覆盖（显式 false、undefined、优先级、旧 bug 文案
+  不复发、120 秒边界）；控制按钮与新页面仍不做，
+  live 点击未验。差：代码内（有意不做）+ 外部（live）。
+- 生产就绪 5.0 → 5.5，未达 10：仅因合并/灰度准备
+  工作（可跑脚本 + 可执行清单）小幅上升；CI 仍
+  UNSTABLE、未合并、未部署、真机 0 项。差：外部为主。
+- 动态安全 2.0 → 2.0，未达 10：本轨无任何动态验证，
+  V-7 未跑。差：外部（KVM 宿主 + 审计结论）。
diff --git a/tools/chv_merge_precheck.py b/tools/chv_merge_precheck.py
new file mode 100644
index 0000000..b4d0a5c
--- /dev/null
+++ b/tools/chv_merge_precheck.py
@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""CHV 合并前检查（只读，绝不合并、绝不改 CI、绝不写库）。
+
+用途：合并 §13.1/§14 链上 PR 之前，逐个核对 base/head、
+mergeable、CI 状态、与 main 头的关系与堆叠顺序，输出
+诚实结论。CI 仍是 UNSTABLE 就如实报 NOT-READY，不把
+账单模式的秒级失败读成代码失败，也不把 UNSTABLE 读成绿。
+
+用法：
+    python3 tools/chv_merge_precheck.py [PR号 ...]
+缺省检查整条 CHV 链（不含本轨新 PR；新 PR 建好后把
+它的号追加到参数末尾再跑一次）：
+    python3 tools/chv_merge_precheck.py 216 221 230 232 234 \\
+        239 246 252 254 257 264 <本轨新 PR 号>
+
+依赖：gh（已登录）与 git（仅读本地 remote，不 fetch、
+不 checkout、不 merge）。gh 不可用或 API 失败时退出码 2。
+退出码：0 = 全部就绪（可按顺序合并）；1 = 未就绪（见原因）；
+2 = 工具/查询失败（结论不可信，不要据此合并）。
+"""
+
+import json
+import subprocess
+import sys
+
+DEFAULT_STACK = [216, 221, 230, 232, 234, 239, 246, 252, 254, 257, 264]
+
+
+def _run(cmd):
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise RuntimeError("%s 失败：%s" % (
+            " ".join(cmd), (proc.stderr or proc.stdout).strip()[:500]))
+    return proc.stdout
+
+
+def _gh_json(args):
+    return json.loads(_run(["gh"] + args))
+
+
+def _check_summary(rollup):
+    """把 statusCheckRollup 归纳成一句话，不猜失败原因。"""
+    if not rollup:
+        return "NO-CHECKS"
+    states = []
+    for item in rollup:
+        name = item.get("name") or item.get("workflowName") or "?"
+        conclusion = item.get("conclusion")
+        status = item.get("status")
+        states.append("%s:%s" % (name, conclusion or status))
+    conclusions = {item.get("conclusion") for item in rollup}
+    if "FAILURE" in conclusions or "CANCELLED" in conclusions:
+        verdict = "FAIL"
+    elif conclusions <= {"SUCCESS", "SKIPPED", "NEUTRAL", None} \
+            and all(item.get("status") == "COMPLETED" for item in rollup):
+        verdict = "GREEN"
+    else:
+        verdict = "PENDING"
+    return "%s [%s]" % (verdict, "; ".join(states))
+
+
+def main(argv):
+    try:
+        prs = [int(a) for a in argv] or list(DEFAULT_STACK)
+    except ValueError:
+        print("参数须为 PR 号（整数）", file=sys.stderr)
+        return 2
+    try:
+        repo = _gh_json(["repo", "view", "--json", "nameWithOwner"])[
+            "nameWithOwner"]
+        main_head = _run(
+            ["gh", "api", "repos/%s/commits/main" % repo, "--jq",
+             ".sha"]).strip()
+    except Exception as exc:
+        print("查询仓库/main 头失败：%s" % exc, file=sys.stderr)
+        return 2
+    print("仓库：%s" % repo)
+    print("当前 main 头：%s" % main_head)
+    print("链（按合并顺序）：%s" % " → ".join(str(p) for p in prs))
+    print("")
+    problems = []
+    tool_failed = False
+    prev_head_branch = "main"
+    for number in prs:
+        try:
+            pr = _gh_json(["pr", "view", str(number), "--json",
+                           "number,title,state,baseRefName,headRefName,"
+                           "headRefOid,mergeable,mergeStateStatus,"
+                           "statusCheckRollup,additions,deletions,"
+                           "changedFiles"])
+        except Exception as exc:
+            print("#%d：查询失败：%s" % (number, exc))
+            tool_failed = True
+            continue
+        chain_ok = pr["baseRefName"] == prev_head_branch
+        ci = _check_summary(pr.get("statusCheckRollup") or [])
+        ready = (pr["state"] == "OPEN" and pr["mergeable"] == "MERGEABLE"
+                 and pr["mergeStateStatus"] == "CLEAN" and chain_ok
+                 and ci.startswith("GREEN"))
+        print("#%(number)d %(title)s" % pr)
+        print("  base=%(baseRefName)s head=%(headRefName)s@%(headRefOid)s"
+              % pr)
+        print("  state=%(state)s mergeable=%(mergeable)s "
+              "mergeState=%(mergeStateStatus)s" % pr)
+        print("  diff: +%s/-%s, %s files" % (
+            pr.get("additions"), pr.get("deletions"),
+            pr.get("changedFiles")))
+        print("  CI: %s" % ci)
+        print("  链衔接（base 应为上一 PR 的 head 分支 %s）：%s" % (
+            prev_head_branch, "OK" if chain_ok else "MISMATCH"))
+        print("  本项：%s" % ("READY" if ready else "NOT-READY"))
+        if not ready:
+            reasons = []
+            if pr["state"] != "OPEN":
+                reasons.append("state=%s" % pr["state"])
+            if pr["mergeable"] != "MERGEABLE":
+                reasons.append("mergeable=%s" % pr["mergeable"])
+            if pr["mergeStateStatus"] != "CLEAN":
+                reasons.append("mergeState=%s"
+                               % pr["mergeStateStatus"])
+            if not chain_ok:
+                reasons.append("base 衔接断裂")
+            if not ci.startswith("GREEN"):
+                reasons.append("CI 未绿（%s）" % ci)
+            problems.append("#%d：%s" % (number, "; ".join(reasons)))
+        prev_head_branch = pr["headRefName"]
+    print("")
+    if tool_failed:
+        print("结论：查询失败，结论不可信，不要据此合并。")
+        return 2
+    if problems:
+        print("结论：NOT-READY，禁止合并。原因：")
+        for problem in problems:
+            print("  - %s" % problem)
+        print("注意：CI 的秒级 verify 失败此前被定性为账单模式，"
+              "脚本不据此改判——账单恢复变绿前，结论保持 NOT-READY。")
+        return 1
+    print("结论：READY，可按上面顺序逐个合并（合并后仍默认关闭，"
+          "再跑 §14 灰度清单）。")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))