codex-app-server/cloud/tests/test_chv_closeup.py

"""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


def _public_resolver(host, port):
    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
    实验室绝不联网，真实解析只在真机发生。"""
    return ["93.184.216.34"]


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",
               # feat-chv-muse-p9：推理出口预审的假 origin（公网
               # https，配假解析器；实验室不起真 VM、不联网）。
               "SUB2API_BASE_URL": "https://xgapi.lol",
               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
        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.sentinel_resolver = _public_resolver
        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 _register_other_vm_net(self, vm="chv-other"):
        vm_dir = os.path.join(self.state_dir, vm)
        os.makedirs(vm_dir, exist_ok=True)
        with open(os.path.join(vm_dir, "net.json"), "w") as fh:
            json.dump([{"tap": "tap0",
                        "mac": "aa:bb:cc:dd:ee:01",
                        "ip": "192.168.10.10",
                        "mask": ""}], fh)

    def test_restore_net_conflict_refused_before_launch(self):
        # feat-chv-cover100：restore 也须过 H7 冲突校验——
        # 此前 restore 路径不校验不登记，恢复出的 VM 会与
        # 已登记 VM 静默复用 tap/mac/ip，且自身不登记、后
        # 来的 wake 也看不到它，两边都漏判。
        url, _ = self._snap_url()
        self._register_other_vm_net()
        ex = self.make_exec(env_extra={
            "CHV_NET_JSON": '[{"tap": "tap0", '
                            '"mac": "aa:bb:cc:dd:ee:02"}]'})
        with self.assertRaises(ChvFailure) as ctx:
            ex.restore_vm(self._task(), url)
        self.assertEqual(ctx.exception.kind, "net_tap_conflict")
        self.assertIn("vm_failed", self._audit_events())
        self.assertEqual(self.launches, [])
        self.assertFalse(os.path.exists(
            os.path.join(self.state_dir, "chv-abc123", "net.json")))

    def test_restore_registers_net_entries(self):
        # feat-chv-cover100：恢复成功后登记 net.json，后来
        # 的 wake 才能看到恢复出的 VM 占用的 tap/mac/ip。
        url, _ = self._snap_url()
        ex = self.make_exec(env_extra={
            "CHV_NET_JSON": '[{"tap": "tap0", '
                            '"mac": "aa:bb:cc:dd:ee:01", '
                            '"ip": "192.168.10.10"}]'})
        out = ex.restore_vm(self._task(), url)
        self.assertEqual(out["mode"], "restored")
        with open(os.path.join(self.state_dir, "chv-abc123",
                               "net.json")) as fh:
            entries = json.load(fh)
        self.assertEqual(entries[0]["tap"], "tap0")
        self.assertEqual(entries[0]["mac"], "aa:bb:cc:dd:ee:01")
        self.assertEqual(entries[0]["ip"], "192.168.10.10")

    def test_restore_net_register_failure_refuses(self):
        # feat-chv-cover100：登记失败与 wake 同款审计+拒绝
        #（P2-19 语义），restore_vm 的 except 会 stop_vm
        # 清掉半恢复 VM（shutdown 已发出）。
        url, _ = self._snap_url()
        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")
        ex = self.make_exec(env_extra={
            "CHV_NET_JSON": '[{"tap": "tap0"}]'})
        with self.assertRaises(ChvFailure) as ctx:
            ex.restore_vm(self._task(), url)
        self.assertEqual(ctx.exception.kind,
                         "net_registry_write_failed")
        self.assertIn("net_registry_write_failed",
                      self._audit_events())
        self.assertTrue(any(c[-1] == "shutdown" for c in self.calls))

    def test_restore_without_net_json_writes_no_registry(self):
        url, _ = self._snap_url()
        ex = self.make_exec()
        out = ex.restore_vm(self._task(), url)
        self.assertEqual(out["mode"], "restored")
        self.assertFalse(os.path.exists(
            os.path.join(self.state_dir, "chv-abc123", "net.json")))

    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_audited_not_raised(self):
        # feat-chv-closeup（P2-17）：脏标记写失败仍不抛（stop
        # 契约不变），但不再全静默——返回 False + 独立审计
        # dirty_marker_write_failed。
        blocker = os.path.join(self.dir, "blocker2")
        with open(blocker, "w") as fh:
            fh.write("x")
        ex = self.make_exec(state_dir=blocker)
        ok = ex._write_dirty_marker("chv-abc123", True)  # 不抛
        self.assertFalse(ok)
        self.assertIn("dirty_marker_write_failed",
                      self._audit_events())

    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_audited_and_reported(self):
        # feat-chv-closeup（P2-19）：登记失败不再“只 warning”——
        # 返回 False + 独立审计 net_registry_write_failed，调用方
        # _boot_vm 据此拒起（见 test_chv_residual）。
        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:
            ok = ex._record_net_entries("chv-x", [{"tap": "tap0"}])
        self.assertFalse(ok)
        self.assertTrue(any("net_registry_write_failed" in line
                            for line in logs.output))
        self.assertIn("net_registry_write_failed",
                      self._audit_events())

    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()))