fix(chv-hybrid): 严格快照默认翻转的旧测试适配 + 长函数拆分

全量复核 4676 项出现 4 失败 6 错误，全是集成时翻为默认开启的
严格快照校验（设计 §2.2）的连锁反应——旧测试用未登记/未标记
快照走 legacy 放行路径。按新语义逐项适配，不改生产校验本身：

- test_chv_opt_stage1.CrossTenantRestoreTest：跨租户拒绝的审计
  事件改断言 vm_restore_denied（严格路径的设计事件；
  vm_restore_tenant_denied 只在显式关闭严格校验时出现）；
  legacy 警告放行测试改走显式 CHV_SNAPSHOT_STRICT_TENANT=0。
- test_chv_v3_fixes（5 个 restore 测试）：登记带匹配 tenant
  （proj1）的快照，意图不变（盘预检 / source_url / 清旧 VMM
  仍由各自断言覆盖）。
- test_cloud_hypervisor.LifecycleTest：登记 file:///snap/1
  （tenant proj1），restore/resume 顺序断言不变。
- 长函数：B 轨合入把 _run_with_persistent_vm 推到 53 行，超
  50 行架构上限；finally 收尾块抽为 _suspend_persistent_after_task。

针对性 12 项全部通过；全量待重跑。

cloud/autoscaler.py

@@ -62,6 +62,29 @@ from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.queue import ready_tasks
 from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE_SECONDS
 
+# CHV 混合模式每宿主预热数 W（DESIGN-HYBRID.md §1.2 已定值：
+# 2 台/宿主；池目标数 P = W × 在线宿主数）。
+CHV_WARM_POOL_PER_HOST = 2
+
+
+def pool_target_for_hosts(online_hosts,
+                          per_host=CHV_WARM_POOL_PER_HOST):
+    """池目标数 P = W × 在线宿主数（§1.2 口径）。
+
+    在线宿主数/每宿主预热数为负或非整数时按 0 计，
+    绝不算出负的池目标。"""
+    try:
+        hosts = int(online_hosts)
+    except (TypeError, ValueError):
+        return 0
+    try:
+        per = int(per_host)
+    except (TypeError, ValueError):
+        return 0
+    if hosts <= 0 or per <= 0:
+        return 0
+    return hosts * per
+
 
 def queue_depth(db_path, executor=None):
     """当前就绪任务数：依赖已满足且状态为 queued 的任务。
@@ -126,19 +149,35 @@ def queue_depth(db_path, executor=None):
 
 
 def desired_workers(depth, per_worker_capacity,
-                    min_workers, max_workers):
+                    min_workers, max_workers, pool_target=0):
     """按深度算目标 worker 数。
 
     depth=0 时也要保留 min_workers（默认 1）：冷启动要快，
     且 scaler 缩到 0 后再拉起仍有进程启动延迟；想缩到 0
     可以把 min_workers 配成 0，scaler 仍在跑所以能再拉起。
+
+    CHV 混合模式（DESIGN-HYBRID.md §1.2，C 轨）：期望值
+    作用域扩展——队列深度算出的 fleet 需要量之外，再加
+    池目标数 pool_target（= W × 在线宿主数，W=2，见
+    pool_target_for_hosts）。pool_target=0 时与旧公式
+    一字不差。池目标数计入 max_workers 上限：max_workers
+    须配得下 min_workers + pool_target，否则期望值会被
+    上限截断、池子补不满（配错时 state 文件里的
+    pool_target 与 desired 可见此截断，不静默多扩）。
     """
     if per_worker_capacity <= 0:
         per_worker_capacity = 1
+    try:
+        pool_target = int(pool_target) if pool_target is not None else 0
+    except (TypeError, ValueError):
+        pool_target = 0
+    if pool_target < 0:
+        pool_target = 0
     if depth <= 0:
         want = min_workers
     else:
         want = math.ceil(depth / per_worker_capacity)
+    want += pool_target
     return max(min_workers, min(max_workers, want))
 
 
@@ -264,7 +303,31 @@ class Autoscaler:
                  per_worker_capacity=2, interval=30, scale_down_after=3,
                  drain_timeout=60, state_file=None, log_dir=None,
                  worker_cmd=None, executor=None,
-                 guard_alive_seconds=_WORKER_ALIVE_SECONDS):
+                 guard_alive_seconds=_WORKER_ALIVE_SECONDS,
+                 pool_target=None, online_hosts=None,
+                 warm_pool_per_host=CHV_WARM_POOL_PER_HOST):
+        self.db_path = db_path
+        self.worker_args = list(worker_args)
+        self.min_workers = min_workers
+        self.max_workers = max_workers
+        self.per_worker_capacity = per_worker_capacity
+        self.interval = interval
+        self.scale_down_after = scale_down_after
+        self.drain_timeout = drain_timeout
+        self.guard_alive_seconds = guard_alive_seconds
+        # CHV 混合模式池目标数（§1.2）：显式 pool_target 优先；
+        # 未给时按 online_hosts × warm_pool_per_host（W=2）算；
+        # 两者皆无即 0（旧行为）。L2 仍不自动加主机。
+        if pool_target is not None:
+            try:
+                self.pool_target = max(0, int(pool_target))
+            except (TypeError, ValueError):
+                self.pool_target = 0
+        elif online_hosts is not None:
+            self.pool_target = pool_target_for_hosts(
+                online_hosts, warm_pool_per_host)
+        else:
+            self.pool_target = 0
         self.db_path = db_path
         self.worker_args = list(worker_args)
         self.min_workers = min_workers
@@ -407,7 +470,8 @@ class Autoscaler:
             self._write_state()
             return self.last_depth, self.last_desired, len(self.workers)
         desired = desired_workers(depth, self.per_worker_capacity,
-                                  self.min_workers, self.max_workers)
+                                  self.min_workers, self.max_workers,
+                                  pool_target=self.pool_target)
         self.last_depth = depth
         self.last_desired = desired
         self._reap()
@@ -477,8 +541,9 @@ class Autoscaler:
         if action != "none":
             log_event(self.logger, "autoscaler_reconcile",
                       message="autoscaler reconcile: depth=%s desired=%s "
-                      "actual=%s action=%s"
-                      % (depth, desired, len(self.workers), action))
+                      "actual=%s action=%s pool_target=%s"
+                      % (depth, desired, len(self.workers), action,
+                         self.pool_target))
         self._write_state()
         return depth, desired, len(self.workers)
 
@@ -489,6 +554,7 @@ class Autoscaler:
             "timestamp": time.time(),
             "depth": self.last_depth,
             "desired": self.last_desired,
+            "pool_target": self.pool_target,
             "actual": len(self.workers),
             "workers": [{"id": wid, "pid": entry["proc"].pid,
                          "started_at": entry["started_at"]}
@@ -571,6 +637,16 @@ def main(argv=None):
                              "已显式给 --executor 时以显式为准）；"
                              "不给时按全部任务算（可能轻微多扩，"
                              "受 max-workers 封顶）")
+    parser.add_argument("--pool-target", type=int, default=None,
+                        help="CHV 混合模式池目标数（§1.2）：期望值"
+                             " = 队列深度算出的 fleet 需要量 + 该数；"
+                             "不给时按 --online-hosts × "
+                             "--warm-pool-per-host 算；两者皆无即 0")
+    parser.add_argument("--online-hosts", type=int, default=None,
+                        help="在线 CHV 宿主数：与 --warm-pool-per-host"
+                             " 相乘得池目标数")
+    parser.add_argument("--warm-pool-per-host", type=int, default=2,
+                        help="每宿主预热数 W，已定值 2 台/宿主（§1.2）")
     parser.add_argument("worker_args", nargs=argparse.REMAINDER,
                         help="传给 cloud.worker 的参数（写在 -- 之后）")
     args = parser.parse_args(argv)
@@ -595,7 +671,9 @@ def main(argv=None):
         per_worker_capacity=args.per_worker_capacity,
         interval=args.interval, scale_down_after=args.scale_down_after,
         drain_timeout=args.drain_timeout, state_file=args.state_file,
-        log_dir=args.log_dir, executor=args.executor)
+        log_dir=args.log_dir, executor=args.executor,
+        pool_target=args.pool_target, online_hosts=args.online_hosts,
+        warm_pool_per_host=args.warm_pool_per_host)
 
     def _on_signal(signum, frame):
         scaler.request_stop()

cloud/chv/hybrid_events.py

@@ -0,0 +1,146 @@
+"""CHV 混合模式审计事件目录（C 轨，feat-chv-hybrid-c）。
+
+唯一依据：~/workspace/chv-multi-host/DESIGN-HYBRID.md
+§7 与 §9.1 第 13 项。事件名字符串与设计原文一字不差，
+供 A/B 两轨的发射点 import（A/B 工作树里还没有本文件，
+他们先按设计原文的事件名字符串直接发射，协调员集成时
+统一收到本模块名下）。
+
+规则（设计 §7）：
+- 全部事件带 host_id（宿主名，默认 socket.gethostname()）。
+- 绝不带密钥值：字段名命中密钥名（见 SECRET_FIELD_NAMES）
+  即拒绝，错误信息只含字段名、不含值；调用方仍须遵守
+  executor._emit 的既有契约（detail 只含 VM 名、路径、
+  URL、kind，不含任何 key 值）。
+- 本模块只定名 + 必填字段 + 校验/格式化，不发事件、
+  不读库、不碰 VM；发射仍走各轨现有的 _emit / 日志路径。
+
+另有一枚 C 轨事件 vm_recycle_sync_failed：不在 §9.1-13
+的名单里，但 §3.3 要求“同步失败 → 任务标同步失败
+（审计）”，回收同步失败的审计须有个名字，故在此单列
+并注明来源，不混进设计名单。
+"""
+
+import socket
+
+# -- 设计原文事件名（§9.1 第 13 项 / §7，一字不差） ------------
+WARM_POOL_ACQUIRE = "warm_pool_acquire"
+WARM_POOL_MISS = "warm_pool_miss"
+WARM_POOL_RELEASE = "warm_pool_release"
+VM_RESIDENT_SUSPENDED = "vm_resident_suspended"
+VM_RESIDENT_DEMOTED = "vm_resident_demoted"
+VM_RESIDENT_PROMOTED = "vm_resident_promoted"
+RESIDENT_UPGRADE_HINT = "resident_upgrade_hint"
+SNAPSHOT_TENANT_MISMATCH = "snapshot_tenant_mismatch"
+VM_RESTORE_DENIED = "vm_restore_denied"
+
+# 设计 §9.1-13 名单一共 9 个事件（acquire/miss/release
+# 是 §7“warm_pool_acquire/miss/release”的三枚展开，
+# miss 的全名 warm_pool_miss 见 §7“须记 warm_pool_miss
+# 审计”）。
+HYBRID_EVENTS = (
+    WARM_POOL_ACQUIRE,
+    WARM_POOL_MISS,
+    WARM_POOL_RELEASE,
+    VM_RESIDENT_SUSPENDED,
+    VM_RESIDENT_DEMOTED,
+    VM_RESIDENT_PROMOTED,
+    RESIDENT_UPGRADE_HINT,
+    SNAPSHOT_TENANT_MISMATCH,
+    VM_RESTORE_DENIED,
+)
+
+# -- C 轨回收同步审计（§3.3，非 §9.1-13 名单） ---------------
+VM_RECYCLE_SYNC_FAILED = "vm_recycle_sync_failed"
+
+ALL_KNOWN_EVENTS = HYBRID_EVENTS + (VM_RECYCLE_SYNC_FAILED,)
+
+# -- 必填字段：全部事件都须带 host_id -------------------------
+# 其余字段按事件语义：vm = VM 名，tenant_id = 租户，
+# reason = 失败原因短文本（已截断，绝不含密钥值）。
+REQUIRED_FIELDS = {
+    WARM_POOL_ACQUIRE: ("host_id", "vm"),
+    WARM_POOL_MISS: ("host_id",),
+    WARM_POOL_RELEASE: ("host_id", "vm"),
+    VM_RESIDENT_SUSPENDED: ("host_id", "tenant_id", "vm"),
+    VM_RESIDENT_DEMOTED: ("host_id", "tenant_id", "vm"),
+    VM_RESIDENT_PROMOTED: ("host_id", "tenant_id", "vm"),
+    RESIDENT_UPGRADE_HINT: ("host_id", "tenant_id"),
+    SNAPSHOT_TENANT_MISMATCH: ("host_id", "tenant_id", "vm"),
+    VM_RESTORE_DENIED: ("host_id", "tenant_id", "vm"),
+    VM_RECYCLE_SYNC_FAILED: ("host_id", "vm"),
+}
+
+# 密钥字段名：命中即拒绝。exact 匹配 + token/secret/
+# password 子串匹配（host_key 等公开指纹名不在此列，
+# 不可误拦）。
+SECRET_FIELD_NAMES = frozenset({
+    "gitea_token", "token", "secret", "password", "api_key",
+    "apikey", "private_key", "access_key", "session_key",
+    "credential", "credentials", "seed", "auth",
+})
+_SECRET_SUBSTRINGS = ("token", "secret", "password")
+
+
+def _is_secret_field(name: str) -> bool:
+    lowered = str(name).lower()
+    if lowered in SECRET_FIELD_NAMES:
+        return True
+    return any(part in lowered for part in _SECRET_SUBSTRINGS)
+
+
+def default_host_id() -> str:
+    """宿主名：审计事件的 host_id 缺省值。"""
+    return socket.gethostname()
+
+
+def validate_event_fields(event: str, fields: dict) -> dict:
+    """校验一个事件的字段，返回清洗后的副本（去 None 值）。
+
+    未知事件名、缺必填字段、host_id 为空、字段名命中密钥
+    名时一律 ValueError；错误信息只含事件名/字段名，
+    绝不含字段值。"""
+    if event not in ALL_KNOWN_EVENTS:
+        raise ValueError("未知审计事件：%s" % event)
+    cleaned = {k: v for k, v in dict(fields or {}).items()
+               if v is not None}
+    secret = sorted(k for k in cleaned if _is_secret_field(k))
+    if secret:
+        raise ValueError(
+            "审计事件 %s 不得带密钥字段：%s" % (event, ",".join(secret)))
+    missing = [k for k in REQUIRED_FIELDS[event]
+               if k not in cleaned
+               or (isinstance(cleaned[k], str)
+                   and not cleaned[k].strip())]
+    if missing:
+        raise ValueError(
+            "审计事件 %s 缺必填字段：%s" % (event, ",".join(missing)))
+    return cleaned
+
+
+def build_event(event: str, fields: dict | None = None,
+                host_id: str | None = None) -> dict:
+    """组装一个事件的字段 dict：host_id 未给时取本机宿主名，
+    再过 validate_event_fields。返回的 dict 可直接用于
+    format_detail 或交审计回调。"""
+    merged = dict(fields or {})
+    if host_id is not None:
+        merged["host_id"] = host_id
+    elif not merged.get("host_id"):
+        merged["host_id"] = default_host_id()
+    return validate_event_fields(event, merged)
+
+
+def format_detail(event: str, fields: dict | None = None,
+                  host_id: str | None = None) -> str:
+    """把事件字段格式化成 executor._emit 的 detail 字符串
+    （"k=v k=v"，键排序，值截断 200 字符、去换行）。
+    先过 build_event 校验，所以带 host_id、不带密钥字段。"""
+    cleaned = build_event(event, fields, host_id=host_id)
+    parts = []
+    for key in sorted(cleaned):
+        value = str(cleaned[key]).replace("\n", " ").replace("\r", " ")
+        if len(value) > 200:
+            value = value[:200]
+        parts.append("%s=%s" % (key, value))
+    return " ".join(parts)

cloud/tests/test_chv_closeup.py

@@ -26,6 +26,7 @@ from pathlib import Path
 from unittest import mock
 
 try:
+    from cloud.chv.disk import record_snapshot
     from cloud.chv.executor import CloudHypervisorExecutor
     from cloud.chv.hostkeys import dirty_marker_path
     from cloud.chv.worker_adapter import (
@@ -38,6 +39,7 @@ try:
     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.disk import record_snapshot
     from cloud.chv.executor import CloudHypervisorExecutor
     from cloud.chv.hostkeys import dirty_marker_path
     from cloud.chv.worker_adapter import (
@@ -288,8 +290,20 @@ class RestoreSnapshotTest(CloseupCase):
             fh.write(b"S" * 60)
         return "file://" + path, path
 
+    def _register_snapshot(self, url, tenant="proj1", vm="chv-abc123"):
+        # 严格快照校验（设计 §2.2，默认开启）：恢复须登记带
+        # tenant_id 且与请求方一致；_task() 的 project_id 是
+        # proj1，即请求租户。登记用 record_snapshot（与
+        # snapshot_vm 同款登记口径）。vm 参数只定登记落在哪
+        # 个 chv-* 目录（find_snapshot_entry 全盘扫描），默认
+        # chv-abc123；net_register 失败测试须用别的 vm 名，
+        # 否则登记建的目录会撞它同名 blocker 文件。
+        record_snapshot(self.state_dir, vm, url, 20.0,
+                        tenant_id=tenant)
+
     def test_restore_failure_closes_vm(self):
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         ex = self.make_exec()
 
         def runner(cmd, timeout=None):
@@ -305,6 +319,7 @@ class RestoreSnapshotTest(CloseupCase):
 
     def test_restore_success_and_wake_from_hibernate(self):
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         ex = self.make_exec()
         out = ex.restore_vm(self._task(), url)
         self.assertEqual(out["mode"], "restored")
@@ -327,6 +342,7 @@ class RestoreSnapshotTest(CloseupCase):
         # 已登记 VM 静默复用 tap/mac/ip，且自身不登记、后
         # 来的 wake 也看不到它，两边都漏判。
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         self._register_other_vm_net()
         ex = self.make_exec(env_extra={
             "CHV_NET_JSON": '[{"tap": "tap0", '
@@ -343,6 +359,7 @@ class RestoreSnapshotTest(CloseupCase):
         # feat-chv-cover100：恢复成功后登记 net.json，后来
         # 的 wake 才能看到恢复出的 VM 占用的 tap/mac/ip。
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         ex = self.make_exec(env_extra={
             "CHV_NET_JSON": '[{"tap": "tap0", '
                             '"mac": "aa:bb:cc:dd:ee:01", '
@@ -361,6 +378,7 @@ class RestoreSnapshotTest(CloseupCase):
         #（P2-19 语义），restore_vm 的 except 会 stop_vm
         # 清掉半恢复 VM（shutdown 已发出）。
         url, _ = self._snap_url()
+        self._register_snapshot(url, vm="chv-zzz")
         blocker = os.path.join(self.state_dir, "chv-abc123")
         os.makedirs(self.state_dir, exist_ok=True)
         with open(blocker, "w") as fh:
@@ -377,6 +395,7 @@ class RestoreSnapshotTest(CloseupCase):
 
     def test_restore_without_net_json_writes_no_registry(self):
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         ex = self.make_exec()
         out = ex.restore_vm(self._task(), url)
         self.assertEqual(out["mode"], "restored")
@@ -385,6 +404,7 @@ class RestoreSnapshotTest(CloseupCase):
 
     def test_restore_provisioned_missing_known_hosts(self):
         url, _ = self._snap_url()
+        self._register_snapshot(url)
         ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"})
         with self.assertRaises(ChvFailure) as ctx:
             ex.restore_vm(self._task(), url)

cloud/tests/test_chv_fix.py

@@ -21,7 +21,9 @@ from types import SimpleNamespace
 
 try:
     from cloud.chv.api_client import _unix_socket_api
-    from cloud.chv.disk import check_disk_quota, provision_task_disk
+    from cloud.chv.disk import (check_disk_quota, provision_task_disk,
+                                record_snapshot)
+    from cloud.chv.executor import write_persistent_vm
     from cloud.chv.hostkeys import build_seed_disk, seed_disk_path
     from cloud.chv.experimental_pool import WarmPool
     from cloud.sandbox import (
@@ -32,7 +34,9 @@ try:
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud.chv.api_client import _unix_socket_api
-    from cloud.chv.disk import check_disk_quota, provision_task_disk
+    from cloud.chv.disk import (check_disk_quota, provision_task_disk,
+                                record_snapshot)
+    from cloud.chv.executor import write_persistent_vm
     from cloud.chv.hostkeys import build_seed_disk, seed_disk_path
     from cloud.chv.experimental_pool import WarmPool
     from cloud.sandbox import (
@@ -227,8 +231,13 @@ class RestoreNoFallbackTest(ChvFixCase):
         # feat-chv-secfix（L2）：恢复源须在白名单目录内，
         # 本测试的意图是“起 VMM 前失败”，用 state_dir 下路径。
         snap_url = "file://" + os.path.join(self.state_dir, "snap", "x")
+        # 严格快照校验（设计 §2.2，默认开启）：恢复须登记带
+        # tenant_id 且与请求方一致；本测试意图是 known-hosts
+        # 门，登记与请求方同租户 p1 后流程才走到该门。
+        record_snapshot(self.state_dir, "chv-r1", snap_url, 20.0,
+                        tenant_id="p1")
         with self.assertRaises(ChvFailure) as ctx:
-            ex.restore_vm({"id": "r1"}, snap_url)
+            ex.restore_vm({"id": "r1", "project_id": "p1"}, snap_url)
         self.assertEqual(ctx.exception.kind, "host_key_not_pinned")
         self.assertEqual(launches, [])
 
@@ -304,6 +313,14 @@ class GatewayExtraEnvTest(ChvFixCase):
         db_path = os.path.join(self.dir, "control.db")
         init_db(db_path)
         ex = self.make_exec()
+        # 混合模式网关分流（设计 §6）：网关任务只接常驻 VM。
+        # 本测试的租户 p1（project_id → tenant）须有常驻有效态
+        # 登记 + Paid 档位，否则 _check_gateway_vm_class 以
+        # gateway_pool_denied 拒绝；池 VM 拒绝腿见下一测试。
+        write_persistent_vm(self.state_dir, "p1", {
+            "vm": "chv-persist-p1", "tenant_id": "p1",
+            "state": "resident-active", "tier": "Paid",
+            "origin": "task", "gitea_synced": False})
         worker = Worker(db_path, "/fake/host.py", executor=ex,
                         env={"XCM_API_KEY": "test-placeholder"})
         task = {"id": "gw1", "project_id": "p1", "prompt": "做目标",
@@ -330,6 +347,19 @@ class GatewayExtraEnvTest(ChvFixCase):
             os.path.join(self.state_dir, "chv-gw1",
                          "ssh_host_ed25519_key")))
 
+    def test_worker_gateway_guard_denies_pool_vm(self):
+        # 混合模式网关分流（设计 §6）：无常驻租约的网关任务
+        # 走池即以 gateway_pool_denied 明确拒绝，不偷偷放行。
+        ex = self.make_exec()
+        task = {"id": "gw2", "project_id": "p1", "prompt": "做目标",
+                "goal_mode": 1,
+                "_gateway_provider": "sub2api",
+                "_gateway_env": {"SUB2API_API_KEY": "gwkey",
+                                 "SUB2API_BASE_URL": "http://gw/"}}
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task(task)
+        self.assertEqual(ctx.exception.kind, "gateway_pool_denied")
+
 
 # -- P1-2 取消/清理/看门狗 ----------------------------------------------
 class CancelCleanupTest(ChvFixCase):

cloud/tests/test_chv_hybrid_a.py

@@ -0,0 +1,718 @@
+"""CHV 混合模式 A 轨（feat-chv-hybrid-a）测试。
+
+只用标准库与 mock runner/launcher/api_client，绝不启动真实
+VM（与 test_chv_opt_stage1.py 同款边界）。覆盖 DESIGN-HYBRID.md
+§9.1 第 1–4 项：
+- 混合路由分流（常驻 / 池命中 / 池空冷启 / 排队拒绝）
+- 干净池预热不带租户（无数据盘、无 seed、无同步等待）
+- 数据盘挂载归属 + “任务盘来源”不变量
+- snapshot_tenant_mismatch（含无 tenant_id 登记拒绝）
+- 已定值常量（每宿主预热 2 台、池上限 50%、常驻上限 50%、
+  排队超时 300 秒、热恢复 ≤2 秒、冷恢复 ≤30 秒、快照 1 个/30 天）
+
+不改其它测试文件；阶段 1 的旧语义（prewarm_persistent_vm /
+acquire_for_task / dispatch_task_persistent / _validate_restore
+的 legacy 警告放行）原样保留，只测本轨新增行为。
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv.disk import (
+        read_persistent_vm,
+        read_vm_tenant,
+        record_snapshot,
+        write_persistent_vm,
+    )
+    from cloud.chv.executor import (
+        HYBRID_COLD_RESTORE_TARGET_SECONDS,
+        HYBRID_QUEUE_TIMEOUT_SECONDS,
+        HYBRID_SNAPSHOT_PER_TENANT,
+        HYBRID_SNAPSHOT_RETENTION_DAYS,
+        HYBRID_WARM_POOL_PER_HOST,
+        HYBRID_WARM_RESUME_TARGET_SECONDS,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv.disk import (
+        read_persistent_vm,
+        read_vm_tenant,
+        record_snapshot,
+        write_persistent_vm,
+    )
+    from cloud.chv.executor import (
+        HYBRID_COLD_RESTORE_TARGET_SECONDS,
+        HYBRID_QUEUE_TIMEOUT_SECONDS,
+        HYBRID_SNAPSHOT_PER_TENANT,
+        HYBRID_SNAPSHOT_RETENTION_DAYS,
+        HYBRID_WARM_POOL_PER_HOST,
+        HYBRID_WARM_RESUME_TARGET_SECONDS,
+    )
+
+
+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
+
+    def poll(self):
+        return None if self._alive else 1
+
+    def terminate(self):
+        self._alive = False
+
+    def wait(self, timeout=None):
+        return 0
+
+    def kill(self):
+        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 HybridACase(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, "base.qcow2"),
+                             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.vm_states = {}
+        self.events = []
+
+    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",
+               "CHV_WARM_POOL_SIZE": "2"}
+        if env_extra:
+            env.update(env_extra)
+        if runner is None:
+            runner = self._default_runner
+        if api_client is None:
+            api_client = self._default_api
+        if launcher is None:
+            launcher = self._default_launcher
+        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, launcher=launcher, api_client=api_client,
+            sleeper=lambda s: None, env=env)
+        kwargs.update(overrides)
+        ex = CloudHypervisorExecutor(**kwargs)
+        ex.audit_callback = lambda e, t, d, k: self.events.append(
+            (e, d, k))
+        return ex
+
+    def _default_runner(self, cmd, timeout=None):
+        self.calls.append(list(cmd))
+        if cmd[0] == "ssh-keyscan":
+            return FakeResult(
+                0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "")
+        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"}), "")
+            return FakeResult(0, "", "")
+        return FakeResult(0, "", "")
+
+    def _default_api(self, socket_path, method, path, payload=None):
+        self.api_calls.append((method, path, payload))
+        if path == "/vm.info":
+            state = self.vm_states.get(
+                os.path.basename(socket_path)[:-len(".sock")])
+            if state:
+                return (200, {"state": state})
+            return (200, {})
+        return (200, {})
+
+    def _default_launcher(self, cmd):
+        self.launches.append(list(cmd))
+        return FakeProc(alive=True)
+
+    def _task(self, task_id="abc123", project="proj1", tier=None):
+        task = {"id": task_id, "prompt": "做一个加法函数",
+                "project_id": project}
+        if tier is not None:
+            task["tier"] = tier
+        return task
+
+    def _socket(self, vm):
+        os.makedirs(self.state_dir, exist_ok=True)
+        path = os.path.join(self.state_dir, vm + ".sock")
+        with open(path, "w") as fh:
+            fh.write("")
+        return path
+
+    def _event_names(self):
+        return [e for e, _d, _k in self.events]
+
+    def _event_details(self, name):
+        return [d for e, d, _k in self.events if e == name]
+
+
+# -- 已定值常量 -------------------------------------------------------------
+class SettledValuesTest(HybridACase):
+    def test_hybrid_settled_values(self):
+        ex = self.make_exec()
+        self.assertEqual(HYBRID_WARM_POOL_PER_HOST, 2)
+        self.assertEqual(ex.warm_pool_per_host_target(), 2)
+        self.assertEqual(ex.clean_pool_target(), 2)
+        # 池上限 = 低频并发峰值的 50%：20 → 10，50 → 25。
+        self.assertEqual(ex.warm_pool_cap(20), 10)
+        self.assertEqual(ex.warm_pool_cap(50), 25)
+        self.assertIsNone(ex.warm_pool_cap(None))
+        # 生效目标取 min（配置目标，上限）：目标 2、峰值 20 → 2。
+        self.assertEqual(ex.clean_pool_effective_target(20), 2)
+        ex_big = self.make_exec(env_extra={
+            "CHV_WARM_POOL_SIZE": "30",
+            "CHV_LOW_FREQ_PEAK_CONCURRENCY": "20"})
+        self.assertEqual(ex_big.clean_pool_effective_target(), 10)
+        # 常驻上限 = 槽位 50%：4 槽 → 2 台，48 槽 → 24 台。
+        self.assertEqual(ex.resident_cap(), 2)
+        ex48 = self.make_exec(env_extra={"CHV_MAX_VMS": "48"})
+        self.assertEqual(ex48.resident_cap(), 24)
+        # 排队超时 300 秒；热恢复 ≤2 秒、冷快照恢复 ≤30 秒（目标值，
+        # 只作常量/注释，不在单元测试里硬测时延）；快照 1 个/30 天。
+        self.assertEqual(HYBRID_QUEUE_TIMEOUT_SECONDS, 300)
+        self.assertEqual(HYBRID_WARM_RESUME_TARGET_SECONDS, 2)
+        self.assertEqual(HYBRID_COLD_RESTORE_TARGET_SECONDS, 30)
+        self.assertEqual(HYBRID_SNAPSHOT_PER_TENANT, 1)
+        self.assertEqual(HYBRID_SNAPSHOT_RETENTION_DAYS, 30)
+
+
+# -- 混合路由分流 -----------------------------------------------------------
+class HybridRouteTest(HybridACase):
+    def test_route_by_tier(self):
+        ex = self.make_exec()
+        self.assertEqual(
+            ex._hybrid_route(self._task(tier="paid")), "resident")
+        self.assertEqual(
+            ex._hybrid_route(self._task(tier="heavy")), "resident")
+        self.assertEqual(
+            ex._hybrid_route(self._task(tier="PAID")), "resident")
+        self.assertEqual(
+            ex._hybrid_route(self._task(tier="free")), "pool")
+        self.assertEqual(ex._hybrid_route(self._task()), "pool")
+        # 常驻是权益门：Paid 但无 tenant（无归属）即 fail-closed 走池。
+        task = {"id": "x1", "prompt": "p", "tier": "paid"}
+        self.assertEqual(ex._hybrid_route(task), "pool")
+
+    def test_dispatch_hybrid_validates_task(self):
+        ex = self.make_exec()
+        with self.assertRaises(ValueError):
+            ex.dispatch_task_hybrid({"prompt": "p"})
+        with self.assertRaises(ValueError):
+            ex.dispatch_task_hybrid({"id": "x"})
+
+    def test_dispatch_hybrid_pool_disabled_falls_back(self):
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
+        out = ex.dispatch_task_hybrid(self._task())
+        self.assertEqual(out["status"], "completed")
+        # 退化为阶段 1 旧语义：结果不带 route/allocation。
+        self.assertNotIn("route", out)
+        self.assertEqual(ex._vms, {})
+
+    def test_dispatch_pool_cold_when_pool_empty(self):
+        ex = self.make_exec()
+        out = ex.dispatch_task_hybrid(self._task(task_id="t1"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "pool")
+        self.assertEqual(out["allocation"], "pool-cold")
+        self.assertEqual(out["vm_mode"], "cold")
+        self.assertIn("warm_pool_miss", self._event_names())
+        self.assertEqual(ex._vms, {})
+
+    def test_dispatch_pool_queue_timeout_refuses(self):
+        ex = self.make_exec(env_extra={
+            "CHV_QUEUE_TIMEOUT": "0", "CHV_MAX_VMS": "1"})
+        # 宿主已满（1 台活 VM 占满唯一槽位），池空 → 排队 0 秒即拒绝。
+        ex._vms["chv-busy"] = FakeProc(alive=True)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task_hybrid(self._task(task_id="t1"))
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+        self.assertEqual(self.launches, [])
+        self.assertIn("warm_pool_miss", self._event_names())
+
+    def test_dispatch_resident_cold_when_no_resident_vm(self):
+        ex = self.make_exec()
+        out = ex.dispatch_task_hybrid(
+            self._task(task_id="t1", project="paid1", tier="paid"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "resident")
+        self.assertEqual(out["allocation"], "resident-cold")
+        self.assertEqual(out["vm"], "chv-persist-paid1")
+        rec = read_persistent_vm(self.state_dir, "paid1")
+        self.assertEqual(rec["state"], "resident-active")
+        self.assertEqual(rec["tier"], "paid")
+        self.assertTrue(rec["resident"])
+        self.assertEqual(ex._resident_count(), 1)
+
+    def test_dispatch_resident_resumed_from_prewarm(self):
+        ex = self.make_exec()
+        ex.prewarm_persistent_vm("paid1")
+        self._socket("chv-persist-paid1")
+        self.vm_states["chv-persist-paid1"] = "Paused"
+        out = ex.dispatch_task_hybrid(
+            self._task(task_id="t1", project="paid1", tier="paid"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "resident")
+        self.assertEqual(out["allocation"], "resident-resumed")
+        self.assertEqual(out["vm"], "chv-persist-paid1")
+        self.assertIn("resume_seconds", out)
+        rec = read_persistent_vm(self.state_dir, "paid1")
+        self.assertEqual(rec["state"], "resident-active")
+        self.assertEqual(rec["tier"], "paid")
+        self.assertTrue(rec["resident"])
+
+    def test_resident_cap_blocks_new_resident(self):
+        ex = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"})
+        # 4 槽位 → 常驻上限 2 台：已有 2 个常驻租户，第 3 个排队即拒。
+        for tenant in ("r1", "r2"):
+            write_persistent_vm(self.state_dir, tenant, {
+                "vm": "chv-persist-%s" % tenant,
+                "tenant_id": tenant, "state": "resident-active",
+                "origin": "task", "gitea_synced": False,
+                "tier": "paid", "resident": True})
+        self.assertEqual(ex._resident_count(), 2)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task_hybrid(
+                self._task(task_id="t1", project="r3", tier="paid"))
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+        self.assertEqual(self.launches, [])
+        # 已有常驻的租户不受上限挡（自己的槽位）。
+        ex2 = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"})
+        write_persistent_vm(self.state_dir, "r1", {
+            "vm": "chv-persist-r1", "tenant_id": "r1",
+            "state": "resident-active", "origin": "task",
+            "gitea_synced": False, "tier": "paid", "resident": True})
+        write_persistent_vm(self.state_dir, "r2", {
+            "vm": "chv-persist-r2", "tenant_id": "r2",
+            "state": "resident-active", "origin": "task",
+            "gitea_synced": False, "tier": "paid", "resident": True})
+        self._socket("chv-persist-r1")
+        self.vm_states["chv-persist-r1"] = "Paused"
+        out = ex2.dispatch_task_hybrid(
+            self._task(task_id="t1", project="r1", tier="paid"))
+        self.assertEqual(out["route"], "resident")
+        self.assertEqual(out["allocation"], "resident-resumed")
+
+
+# -- 干净池预热 -------------------------------------------------------------
+class CleanPoolPrewarmTest(HybridACase):
+    def test_prewarm_clean_pool_has_no_tenant_state(self):
+        ex = self.make_exec()
+        out = ex.prewarm_clean_pool()
+        self.assertTrue(out["pooled"])
+        self.assertEqual(out["mode"], "prewarmed")
+        self.assertEqual(out["vm"], "chv-pool-0")
+        self.assertFalse(out["gitea_synced"])
+        self.assertTrue(out["clean"])
+        self.assertIsNone(out["tenant_id"])
+        self.assertEqual(len(self.launches), 1)
+        vm_dir = os.path.join(self.state_dir, "chv-pool-0")
+        # 出生时不带租户：无数据盘、无 tenant 标记、无 seed。
+        self.assertFalse(os.path.exists(
+            os.path.join(vm_dir, "data.raw")))
+        self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0"))
+        self.assertFalse(os.path.exists(
+            os.path.join(vm_dir, "seed.iso")))
+        self.assertFalse(os.path.exists(
+            os.path.join(vm_dir, "seed")))
+        # 无 Gitea 同步等待：预热全程没有任何 ssh test -d 调用。
+        self.assertFalse(any(
+            "test" in call for call in self.calls
+            if call and call[0] == "ssh"))
+        # 登记同样不带租户。
+        record = ex._read_clean_pool_record(0)
+        self.assertEqual(record["state"], "paused")
+        self.assertIsNone(record["tenant_id"])
+        self.assertFalse(record["gitea_synced"])
+        # socket 造出来后第二次即 already-warm，不重建。
+        self._socket("chv-pool-0")
+        second = ex.prewarm_clean_pool()
+        self.assertEqual(second["mode"], "already-warm")
+        self.assertEqual(len(self.launches), 1)
+
+    def test_prewarm_clean_pool_slots_and_range(self):
+        ex = self.make_exec()
+        first = ex.prewarm_clean_pool(slot=1)
+        self.assertEqual(first["vm"], "chv-pool-1")
+        out = ex.prewarm_clean_pool(slot=5)
+        self.assertFalse(out["pooled"])
+        self.assertEqual(out["reason"], "clean_pool_slot_out_of_range")
+        self.assertEqual(len(self.launches), 1)
+
+    def test_prewarm_clean_pool_disabled_when_size_zero(self):
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
+        self.assertEqual(
+            ex.prewarm_clean_pool()["reason"], "warm_pool_disabled")
+        self.assertEqual(
+            ex.acquire_clean_pool(self._task())["mode"], "cold")
+        self.assertEqual(ex.refill_clean_pool(), {"target": 0,
+                                                  "slots": []})
+
+    def test_prewarm_clean_pool_refuses_tenant_seed(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.prewarm_clean_pool(
+                extra_env={"GITEA_TOKEN": "secret-value"})
+        self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.prewarm_clean_pool(
+                extra_env={"GITEA_URL": "https://gitea.tose.top"})
+        self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused")
+        self.assertEqual(self.launches, [])
+
+    def test_prewarm_clean_pool_never_rebuilds_running_slot(self):
+        ex = self.make_exec()
+        ex._write_clean_pool_record(0, {
+            "vm": "chv-pool-0", "slot": 0, "tenant_id": "tenant-a",
+            "state": "running", "origin": "clean-prewarm",
+            "gitea_synced": False, "data_disk": "x",
+            "updated_at": 0})
+        out = ex.prewarm_clean_pool(slot=0)
+        self.assertFalse(out["pooled"])
+        self.assertEqual(out["reason"], "clean_pool_slot_busy")
+        self.assertEqual(self.launches, [])
+        # 自动选槽跳过正跑的 slot 0，建 slot 1。
+        out2 = ex.prewarm_clean_pool()
+        self.assertEqual(out2["vm"], "chv-pool-1")
+        self.assertEqual(len(self.launches), 1)
+
+    def test_refill_clean_pool(self):
+        ex = self.make_exec()
+        ex.prewarm_clean_pool(slot=0)
+        self._socket("chv-pool-0")
+        out = ex.refill_clean_pool()
+        self.assertEqual(out["target"], 2)
+        self.assertEqual(out["slots"][0]["mode"], "already-warm")
+        self.assertEqual(out["slots"][1]["mode"], "prewarmed")
+        self.assertEqual(len(self.launches), 2)
+
+
+# -- 数据盘挂载归属 + 任务盘来源不变量 ---------------------------------------
+class DataDiskAttachTest(HybridACase):
+    def test_verify_task_disk_source(self):
+        ex = self.make_exec()
+        tenant = "tenant-a"
+        disk = ex._provision_tenant_data_disk(tenant)
+        self.assertEqual(disk, ex._tenant_data_disk_path(tenant))
+        self.assertTrue(os.path.exists(disk))
+        # 合法的租户数据盘通过。
+        ex._verify_task_disk_source("chv-pool-0", tenant, disk)
+        # 干净池 VM 自带的盘（干净快照盘）绝不能当任务盘。
+        pool_data = os.path.join(self.state_dir, "chv-pool-0",
+                                 "data.raw")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._verify_task_disk_source("chv-pool-0", tenant, pool_data)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+        # 基础镜像不能当任务盘。
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._verify_task_disk_source("chv-pool-0", tenant,
+                                        self.rootfs)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+        # 其它租户的盘不能当本租户的任务盘。
+        other = ex._provision_tenant_data_disk("tenant-b")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._verify_task_disk_source("chv-pool-0", tenant, other)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+        # 无盘 / 无租户同样拒绝。
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._verify_task_disk_source("chv-pool-0", tenant, None)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._verify_task_disk_source("chv-pool-0", None, disk)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+
+    def test_attach_refuses_wrong_disk_before_api_call(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.attach_tenant_data_disk(
+                self._task(project="tenant-a"), "chv-pool-0",
+                data_disk=self.rootfs)
+        self.assertEqual(ctx.exception.kind, "task_disk_source_invalid")
+        # 不变量在 API 调用之前拦住：一次 add-disk 都没发。
+        self.assertFalse(any(p == "/vm.add-disk"
+                             for _m, p, _pl in self.api_calls))
+
+    def test_attach_tenant_data_disk(self):
+        ex = self.make_exec()
+        out = ex.attach_tenant_data_disk(
+            self._task(project="tenant-a"), "chv-pool-0")
+        self.assertEqual(out["tenant_id"], "tenant-a")
+        self.assertEqual(out["data_disk"],
+                         ex._tenant_data_disk_path("tenant-a"))
+        self.assertTrue(any(p == "/vm.add-disk"
+                            for _m, p, _pl in self.api_calls))
+        _method, _path, payload = next(
+            (m, p, pl) for m, p, pl in self.api_calls
+            if p == "/vm.add-disk")
+        self.assertEqual(payload["path"], out["data_disk"])
+        # 客内挂载脚本被调过（chv-data-mount）。
+        self.assertTrue(any(
+            "/usr/local/bin/chv-data-mount" in call
+            for call in self.calls if call and call[0] == "ssh"))
+        self.assertEqual(
+            read_vm_tenant(self.state_dir, "chv-pool-0"), "tenant-a")
+
+    def test_dispatch_pool_hit_binds_tenant_disk_and_destroys_vm(self):
+        ex = self.make_exec()
+        ex.prewarm_clean_pool(slot=0)
+        self._socket("chv-pool-0")
+        self.vm_states["chv-pool-0"] = "Paused"
+        out = ex.dispatch_task_hybrid(
+            self._task(task_id="t1", project="tenant-a"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "pool")
+        self.assertEqual(out["allocation"], "pool-hit")
+        self.assertEqual(out["vm"], "chv-pool-0")
+        self.assertTrue(out["warm_hit"])
+        self.assertEqual(out["data_disk"],
+                         ex._tenant_data_disk_path("tenant-a"))
+        self.assertIn("resume_seconds", out)
+        self.assertIn("warm_pool_acquire", self._event_names())
+        self.assertIn("warm_pool_release", self._event_names())
+        # 用过的池 VM 已销毁、不回池：登记清掉、VMM 登记空、归属清掉。
+        self.assertIsNone(ex._read_clean_pool_record(0))
+        self.assertEqual(ex._vms, {})
+        self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0"))
+        # 租户数据盘还在（用户数据不随 VM 销毁而删）。
+        self.assertTrue(os.path.exists(
+            ex._tenant_data_disk_path("tenant-a")))
+        # 池 VM 自带的干净盘从未被当成任务盘建出来。
+        self.assertFalse(os.path.exists(os.path.join(
+            self.state_dir, "chv-pool-0", "data.raw")))
+
+    def test_acquire_sync_incomplete_fails_closed_at_allocate(self):
+        # 分配时同步等不到：fail-closed，VM 解绑销毁，不派任务。
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh-keyscan":
+                return FakeResult(
+                    0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "")
+            if cmd[0] == "ssh":
+                if "test" in cmd:
+                    return FakeResult(1, "", "")
+                if cmd[-1] == "true":
+                    return FakeResult(0, "", "")
+                if "dispatch" in cmd:
+                    return FakeResult(0, json.dumps(
+                        {"status": "completed", "text": "ok"}), "")
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner)
+        ex.prewarm_clean_pool(slot=0)
+        self._socket("chv-pool-0")
+        self.vm_states["chv-pool-0"] = "Paused"
+        task = self._task(task_id="t1", project="tenant-a")
+        task["sync_workspace"] = "/home/chv/workspace/repo"
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task_hybrid(task)
+        self.assertEqual(ctx.exception.kind,
+                         "warm_pool_sync_incomplete")
+        # 已绑过租户态的 VM 绝不回干净池：已销毁、登记清掉。
+        self.assertIsNone(ex._read_clean_pool_record(0))
+        self.assertEqual(ex._vms, {})
+        self.assertFalse(any("dispatch" in call
+                             for call in self.calls
+                             if call and call[0] == "ssh"))
+
+    def test_release_sync_failure_keeps_vm(self):
+        # 回收时 chv-sync push 失败：VM 不回收、不复用。
+        def runner(cmd, timeout=None):
+            self.calls.append(list(cmd))
+            if cmd[0] == "ssh-keyscan":
+                return FakeResult(
+                    0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "")
+            if cmd[0] == "ssh":
+                if "chv-sync" in cmd:
+                    return FakeResult(1, "", "push 被拒绝")
+                if cmd[-1] == "true":
+                    return FakeResult(0, "", "")
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+
+        ex = self.make_exec(runner=runner)
+        ex.prewarm_clean_pool(slot=0)
+        self._socket("chv-pool-0")
+        self.vm_states["chv-pool-0"] = "Paused"
+        acquired = ex.acquire_clean_pool(self._task(project="tenant-a"))
+        self.assertEqual(acquired["mode"], "resumed")
+        out = ex.release_pool_vm(
+            self._task(project="tenant-a"), "chv-pool-0",
+            sync_workspace="/home/chv/workspace/repo")
+        self.assertFalse(out["released"])
+        self.assertEqual(out["reason"], "sync_failed")
+        self.assertIn("warm_pool_release_failed", self._event_names())
+        # VM 还在（未销毁），登记仍是 running 绑定态。
+        self.assertIn("chv-pool-0", ex._vms)
+        record = ex._read_clean_pool_record(0)
+        self.assertEqual(record["state"], "running")
+        self.assertEqual(record["tenant_id"], "tenant-a")
+
+
+# -- 快照 tenant 校验 --------------------------------------------------------
+class SnapshotTenantStrictTest(HybridACase):
+    def _snap_url(self, vm, name="snap1"):
+        snap_dir = os.path.join(self.state_dir, vm, name)
+        os.makedirs(snap_dir, exist_ok=True)
+        with open(os.path.join(snap_dir, "config.json"), "w") as fh:
+            fh.write("{}")
+        with open(os.path.join(snap_dir, "state.json"), "w") as fh:
+            fh.write("{}")
+        return "file://" + snap_dir
+
+    def _record(self, vm, url, tenant):
+        record_snapshot(self.state_dir, vm, url, 20.0,
+                        tenant_id=tenant)
+
+    def test_unmarked_snapshot_denied_in_hybrid_mode(self):
+        url = self._snap_url("chv-a")
+        self._record("chv-a", url, None)
+        ex = self.make_exec()  # CHV_WARM_POOL_SIZE=2 → 严格校验生效
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm({"id": "task-a", "project_id": "tenant-a"},
+                          url)
+        self.assertEqual(ctx.exception.kind,
+                         "snapshot_tenant_mismatch")
+        denied = [d for _e, d, _k in self.events
+                  if _e == "vm_restore_denied"]
+        self.assertTrue(denied)
+        # 审计带 host_id + 请求租户，不带快照内容（URL 不得出
+        # 现在 detail 里）。
+        self.assertIn("host_id=", denied[0])
+        self.assertIn("tenant=tenant-a", denied[0])
+        self.assertNotIn(url, denied[0])
+        self.assertNotIn("snap1", denied[0])
+
+    def test_unregistered_snapshot_denied_in_hybrid_mode(self):
+        url = self._snap_url("chv-a")
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm({"id": "task-a", "project_id": "tenant-a"},
+                          url)
+        self.assertEqual(ctx.exception.kind,
+                         "snapshot_tenant_mismatch")
+        self.assertIn("vm_restore_denied", self._event_names())
+
+    def test_mismatched_snapshot_denied_in_hybrid_mode(self):
+        url = self._snap_url("chv-a")
+        self._record("chv-a", url, "tenant-a")
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.restore_vm({"id": "task-b", "project_id": "tenant-b"},
+                          url)
+        self.assertEqual(ctx.exception.kind,
+                         "snapshot_tenant_mismatch")
+        self.assertIn("vm_restore_denied", self._event_names())
+
+    def test_same_tenant_snapshot_restores_in_hybrid_mode(self):
+        url = self._snap_url("chv-a")
+        self._record("chv-a", url, "tenant-a")
+        ex = self.make_exec()
+        out = ex.restore_vm(
+            {"id": "task-a", "project_id": "tenant-a"}, url)
+        self.assertEqual(out["mode"], "restored")
+        self.assertEqual(out["state"], "running")
+
+    def test_explicit_strict_flag_denies_unmarked(self):
+        url = self._snap_url("chv-a")
+        self._record("chv-a", url, None)
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
+        # 显式 strict_tenant=True 时即使池未启用也拒绝。
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._validate_restore(
+                {"id": "task-a", "project_id": "tenant-a"}, url,
+                strict_tenant=True)
+        self.assertEqual(ctx.exception.kind,
+                         "snapshot_tenant_mismatch")
+        self.assertIn("vm_restore_denied", self._event_names())
+
+    def test_legacy_warn_pass_when_strict_opted_out(self):
+        # 严格校验默认开启（设计 §2.2，集成时翻为默认）；
+        # legacy 警告放行只在显式 CHV_SNAPSHOT_STRICT_TENANT=0
+        # 时保留（有未标记旧快照的环境的逃生口）。
+        url = self._snap_url("chv-a")
+        self._record("chv-a", url, None)
+        ex = self.make_exec(env_extra={
+            "CHV_WARM_POOL_SIZE": "0",
+            "CHV_SNAPSHOT_STRICT_TENANT": "0"})
+        snapshot_url, _net = ex._validate_restore(
+            {"id": "task-a", "project_id": "tenant-a"}, url)
+        self.assertEqual(snapshot_url, url)
+        self.assertIn("vm_restore_tenant_unmarked",
+                      self._event_names())
+        self.assertNotIn("vm_restore_denied", self._event_names())
+
+
+# -- dispatch 入口接线（集成时补）--------------------------------------
+class DispatchWiringTest(HybridACase):
+    def test_dispatch_task_routes_to_hybrid_when_pool_on(self):
+        # warm_pool_size>=1 时 dispatch_task 走混合路由。
+        ex = self.make_exec()
+        out = ex.dispatch_task(self._task(task_id="t1"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "pool")
+        self.assertEqual(out["allocation"], "pool-cold")
+
+    def test_dispatch_task_routes_resident_when_paid(self):
+        ex = self.make_exec()
+        out = ex.dispatch_task(
+            self._task(task_id="t1", project="paid1", tier="paid"))
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["route"], "resident")
+        self.assertEqual(out["allocation"], "resident-cold")
+
+    def test_dispatch_task_stays_cold_when_pool_off(self):
+        # warm_pool_size==0 时 dispatch_task 保持旧冷启语义，
+        # 不进混合路由（结果无 route 字段）。
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"})
+        out = ex.dispatch_task(self._task(task_id="t1"))
+        self.assertEqual(out["status"], "completed")
+        self.assertNotIn("route", out)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_chv_hybrid_b.py

@@ -0,0 +1,710 @@
+"""CHV 混合模式 B 轨（feat-chv-hybrid-b）测试。
+
+只用标准库与 mock runner/launcher/api_client，绝不启动真实
+VM（与 test_chv_opt_stage1.py 同款边界）。覆盖（设计
+DESIGN-HYBRID.md §9.1 第 5–9 项）：
+- 登记字段扩展 + tier 校验（含无 tier 行拒绝常驻）
+- 租约挂起/降级计时（环境变量注入短阈值测状态机转换，
+  含降级同步失败中止）
+- 排队超时拒绝（host_capacity_exceeded，不插队、不抢占、
+  不升降档）
+- 升级提示点去重（一周内同一租户最多一次）
+- 网关分流（常驻可接、池不接）
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv.disk import (
+        read_persistent_vm,
+        write_persistent_vm,
+        write_vm_tenant,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv import ChvFailure
+    from cloud.chv import CloudHypervisorExecutor
+    from cloud.chv.disk import (
+        read_persistent_vm,
+        write_persistent_vm,
+        write_vm_tenant,
+    )
+
+
+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
+
+    def poll(self):
+        return None if self._alive else 1
+
+    def terminate(self):
+        self._alive = False
+
+    def wait(self, timeout=None):
+        return 0
+
+    def kill(self):
+        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 HybridBCase(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, "base.qcow2"),
+                             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.vm_states = {}
+        self.sleeps = []
+        self.events = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def make_exec(self, runner=None, api_client=None, launcher=None,
+                  env_extra=None, sleeper=None, **overrides):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_HOST_KEY_MODE": "keyscan",
+               "CHV_SSH_FORWARD": "1",
+               "CHV_HOST_ID": "host-test-1"}
+        if env_extra:
+            env.update(env_extra)
+        if runner is None:
+            runner = self._default_runner
+        if api_client is None:
+            api_client = self._default_api
+        if launcher is None:
+            launcher = self._default_launcher
+        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, launcher=launcher, api_client=api_client,
+            sleeper=sleeper if sleeper is not None
+            else (lambda s: self.sleeps.append(s)),
+            env=env)
+        kwargs.update(overrides)
+        ex = CloudHypervisorExecutor(**kwargs)
+        ex.audit_callback = lambda event, task, detail, kind=None: \
+            self.events.append((event, detail, kind))
+        return ex
+
+    def _default_runner(self, cmd, timeout=None):
+        self.calls.append(list(cmd))
+        if cmd[0] == "ssh-keyscan":
+            return FakeResult(
+                0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "")
+        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"}), "")
+            return FakeResult(0, "", "")
+        if cmd and os.path.basename(str(cmd[0])) == "ch-remote":
+            if "snapshot" in cmd:
+                url = cmd[cmd.index("snapshot") + 1]
+                if url.startswith("file://"):
+                    snap_dir = url[len("file://"):]
+                    os.makedirs(snap_dir, exist_ok=True)
+                    with open(os.path.join(snap_dir, "state.json"),
+                              "w") as fh:
+                        fh.write("{}")
+                return FakeResult(0, "", "")
+            return FakeResult(0, "", "")
+        return FakeResult(0, "", "")
+
+    def _default_api(self, socket_path, method, path, payload=None):
+        self.api_calls.append((method, path, payload))
+        if path == "/vm.info":
+            state = self.vm_states.get(
+                os.path.basename(socket_path)[:-len(".sock")])
+            if state:
+                return (200, {"state": state})
+            return (200, {})
+        return (200, {})
+
+    def _default_launcher(self, cmd):
+        self.launches.append(list(cmd))
+        return FakeProc(alive=True)
+
+    def _task(self, task_id="abc123", project="proj1"):
+        return {"id": task_id, "prompt": "做一个加法函数",
+                "project_id": project}
+
+    def _event_names(self):
+        return [e[0] for e in self.events]
+
+    def _write_resident(self, ex, tenant, state, tier, last_active,
+                        vm=None):
+        vm = vm or ex.persistent_vm_name(tenant)
+        write_persistent_vm(ex.state_dir, tenant, {
+            "vm": vm, "tenant_id": tenant, "tier": tier,
+            "state": state, "last_active_at": last_active,
+            "lease_expires_at": last_active + 1800 + 86400,
+            "origin": "task", "gitea_synced": True,
+            "updated_at": last_active})
+        return vm
+
+    def _live_vm(self, ex, vm):
+        """给一台 VM 造“活着”的外形：socket 文件 + vm.info。"""
+        sock = os.path.join(ex.state_dir, vm + ".sock")
+        os.makedirs(ex.state_dir, exist_ok=True)
+        with open(sock, "w") as fh:
+            fh.write("")
+        return sock
+
+
+# -- 登记字段扩展 + tier 校验（第 5 项）-------------------------------
+class ResidentRegistryTest(HybridBCase):
+    def test_promote_writes_extended_fields(self):
+        ex = self.make_exec()
+        out = ex.promote_to_resident("t1", "Paid")
+        self.assertTrue(out["promoted"])
+        self.assertFalse(out["queued"])
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["tier"], "Paid")
+        self.assertEqual(record["state"], "resident-active")
+        self.assertIn("last_active_at", record)
+        self.assertIn("lease_expires_at", record)
+        self.assertGreater(record["lease_expires_at"],
+                           record["last_active_at"])
+        self.assertIn("vm_resident_promoted", self._event_names())
+        detail = [d for e, d, _k in self.events
+                  if e == "vm_resident_promoted"][0]
+        self.assertIn("host_id=host-test-1", detail)
+        self.assertIn("tenant=t1", detail)
+
+    def test_promote_free_tier_denied(self):
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.promote_to_resident("tfree", "Free")
+        self.assertEqual(ctx.exception.kind, "tier_denied")
+        self.assertIsNone(read_persistent_vm(ex.state_dir, "tfree"))
+
+    def test_promote_no_tier_row_denied(self):
+        # 无 tier 行 = 无常驻资格（fail-closed，走池）：登记里
+        # 没有 tier 字段、也不传 tier 参数时同样拒绝。
+        ex = self.make_exec()
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.promote_to_resident("tnope")
+        self.assertEqual(ctx.exception.kind, "tier_denied")
+
+    def test_tier_eligible_matrix(self):
+        ex = self.make_exec()
+        self.assertTrue(ex.resident_tier_eligible("x", "Paid"))
+        self.assertTrue(ex.resident_tier_eligible("x", "Heavy"))
+        self.assertFalse(ex.resident_tier_eligible("x", "Free"))
+        self.assertFalse(ex.resident_tier_eligible("x", None))
+        self.assertFalse(ex.resident_tier_eligible("nobody"))
+
+    def test_promote_over_cap_queued_not_preempt(self):
+        # max_vms 默认 4 → 常驻上限 2；第 3 个排队，不抢占前两个。
+        ex = self.make_exec()
+        self.assertTrue(ex.promote_to_resident("t1", "Paid")["promoted"])
+        self.assertTrue(ex.promote_to_resident("t2", "Heavy")["promoted"])
+        out = ex.promote_to_resident("t3", "Paid")
+        self.assertFalse(out["promoted"])
+        self.assertTrue(out["queued"])
+        self.assertEqual(out["reason"], "resident_capacity_exceeded")
+        self.assertEqual(ex._resident_count(), 2)
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-active")
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t2")["state"],
+            "resident-active")
+        self.assertIsNone(read_persistent_vm(ex.state_dir, "t3"))
+
+    def test_promote_idempotent_no_double_slot(self):
+        ex = self.make_exec()
+        ex.promote_to_resident("t1", "Paid")
+        out = ex.promote_to_resident("t1", "Paid")
+        self.assertTrue(out["promoted"])
+        self.assertEqual(ex._resident_count(), 1)
+
+    def test_touch_renews_only_resident(self):
+        ex = self.make_exec(env_extra={
+            "CHV_RESIDENT_IDLE_SUSPEND_SECONDS": "60",
+            "CHV_RESIDENT_IDLE_DEMOTE_SECONDS": "120"})
+        now = 1800000000.0
+        # legacy paused 登记：touch 不写、不造租约字段。
+        write_persistent_vm(ex.state_dir, "tleg", {
+            "vm": ex.persistent_vm_name("tleg"), "tenant_id": "tleg",
+            "state": "paused", "origin": "task",
+            "gitea_synced": True, "updated_at": now - 100})
+        self.assertFalse(ex.touch_resident_activity("tleg", now=now))
+        self.assertNotIn("last_active_at",
+                         read_persistent_vm(ex.state_dir, "tleg"))
+        # 常驻登记：touch 续期并重算 lease_expires_at。
+        vm = self._write_resident(ex, "t2", "resident-active",
+                                  "Paid", now - 100)
+        self.assertTrue(ex.touch_resident_activity("t2", now=now))
+        record = read_persistent_vm(ex.state_dir, "t2")
+        self.assertEqual(record["last_active_at"], now)
+        self.assertEqual(record["lease_expires_at"], now + 60 + 120)
+        self.assertEqual(record["state"], "resident-active")
+        self.assertEqual(record["tier"], "Paid")
+
+    def test_suspend_after_task_keeps_resident_state_and_tier(self):
+        ex = self.make_exec()
+        ex.promote_to_resident("t1", "Paid")
+        vm = ex.persistent_vm_name("t1")
+        self.assertTrue(ex.suspend_after_task({"id": vm},
+                                              tenant_id="t1"))
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["state"], "resident-active")
+        self.assertEqual(record["tier"], "Paid")
+        self.assertIn("last_active_at", record)
+
+    def test_acquire_resident_paused_paid_tier(self):
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"})
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Paid", now - 10)
+        write_vm_tenant(ex.state_dir, vm, "t1")
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.acquire_for_task(self._task(project="t1"))
+        self.assertEqual(out["mode"], "resumed")
+        self.assertEqual(out["vm"], vm)
+        record = read_persistent_vm(ex.state_dir, "t1")
+        # resume 后仍是常驻态（不写回 legacy running），tier 保留。
+        self.assertEqual(record["state"], "resident-paused")
+        self.assertEqual(record["tier"], "Paid")
+
+    def test_acquire_resident_paused_free_tier_denied(self):
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"})
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Free", now - 10)
+        write_vm_tenant(ex.state_dir, vm, "t1")
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.acquire_for_task(self._task(project="t1"))
+        self.assertEqual(out["mode"], "cold")
+        self.assertIn("resident_tier_denied", self._event_names())
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["state"], "resident-paused")
+        self.assertEqual(record["tier"], "Free")
+
+    def test_acquire_legacy_paused_without_tier_still_works(self):
+        # legacy paused 登记无 tier 字段：走旧语义，仍可认领。
+        ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "1"})
+        vm = ex.persistent_vm_name("tleg")
+        write_persistent_vm(ex.state_dir, "tleg", {
+            "vm": vm, "tenant_id": "tleg", "state": "paused",
+            "origin": "prewarm", "gitea_synced": True,
+            "updated_at": 1800000000.0})
+        write_vm_tenant(ex.state_dir, vm, "tleg")
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.acquire_for_task(self._task(project="tleg"))
+        self.assertEqual(out["mode"], "resumed")
+
+
+# -- 租约挂起/降级计时（第 6 项）--------------------------------------
+class LeaseTimerTest(HybridBCase):
+    def _lease_exec(self):
+        return self.make_exec(env_extra={
+            "CHV_RESIDENT_IDLE_SUSPEND_SECONDS": "60",
+            "CHV_RESIDENT_IDLE_DEMOTE_SECONDS": "120"})
+
+    def test_sweep_suspends_idle_active_already_paused(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-active",
+                                  "Paid", now - 61)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.sweep_resident_leases(now=now)
+        self.assertEqual(out["suspended"], ["t1"])
+        self.assertEqual(out["demoted"], [])
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["state"], "resident-paused")
+        self.assertEqual(record["tier"], "Paid")
+        self.assertIn("vm_resident_suspended", self._event_names())
+        detail = [d for e, d, _k in self.events
+                  if e == "vm_resident_suspended"][0]
+        self.assertIn("host_id=host-test-1", detail)
+        self.assertIn("tenant=t1", detail)
+        # VM 本已 Paused：sweep 不应再发 pause。
+        self.assertFalse(any(c[-1] == "pause" for c in self.calls
+                             if c and "ch-remote" in c[0]))
+
+    def test_sweep_suspends_running_vm_calls_pause(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-active",
+                                  "Paid", now - 61)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Running"
+        out = ex.sweep_resident_leases(now=now)
+        self.assertEqual(out["suspended"], ["t1"])
+        self.assertTrue(any(c[-1] == "pause" for c in self.calls
+                            if c and "ch-remote" in c[0]))
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-paused")
+
+    def test_sweep_no_suspend_before_threshold(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        self._write_resident(ex, "t1", "resident-active",
+                             "Paid", now - 30)
+        out = ex.sweep_resident_leases(now=now)
+        self.assertEqual(out, {"suspended": [], "demoted": [],
+                               "aborted": []})
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-active")
+
+    def test_sweep_demotes_after_24h(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Paid", now - 61 - 121)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        # 用户数据盘与任务盘：降级后须保留（快照输入 + 用户数据）。
+        vm_dir = os.path.join(ex.state_dir, vm)
+        os.makedirs(vm_dir, exist_ok=True)
+        with open(os.path.join(vm_dir, "data.raw"), "wb") as fh:
+            fh.write(b"D" * 10)
+        with open(os.path.join(vm_dir, "disk.raw"), "wb") as fh:
+            fh.write(b"K" * 10)
+        out = ex.sweep_resident_leases(now=now)
+        self.assertEqual(out["demoted"], ["t1"])
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["state"], "demoted")
+        self.assertEqual(record["last_active_at"], now - 61 - 121)
+        self.assertIn("snapshot_url", record)
+        self.assertIn("vm_resident_demoted", self._event_names())
+        detail = [d for e, d, _k in self.events
+                  if e == "vm_resident_demoted"][0]
+        self.assertIn("host_id=host-test-1", detail)
+        self.assertIn("tenant=t1", detail)
+        self.assertTrue(os.path.exists(
+            os.path.join(vm_dir, "data.raw")))
+        self.assertTrue(os.path.exists(
+            os.path.join(vm_dir, "disk.raw")))
+        # 槽位已释放：可再 promote 别的租户。
+        self.assertEqual(ex._resident_count(), 0)
+
+    def test_demote_sync_failure_aborts(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Paid", now - 61 - 121)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.demote_resident(
+            "t1", sync_fn=lambda tenant, v: False, now=now)
+        self.assertFalse(out["demoted"])
+        self.assertEqual(out["reason"], "sync_failed")
+        record = read_persistent_vm(ex.state_dir, "t1")
+        self.assertEqual(record["state"], "resident-paused")
+        self.assertNotIn("snapshot_url", record)
+        # VM 保留：socket 还在，未销毁。
+        self.assertTrue(os.path.exists(
+            os.path.join(ex.state_dir, vm + ".sock")))
+        self.assertIn("vm_resident_demote_aborted",
+                      self._event_names())
+
+    def test_demote_snapshot_failure_aborts(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Paid", now - 61 - 121)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        with mock.patch.object(
+                ex, "snapshot_vm",
+                side_effect=ChvFailure("snapshot_failed", "boom")):
+            out = ex.demote_resident("t1", now=now)
+        self.assertFalse(out["demoted"])
+        self.assertEqual(out["reason"], "snapshot_failed")
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-paused")
+
+    def test_demote_not_demotable_states(self):
+        ex = self._lease_exec()
+        now = 1800000000.0
+        self._write_resident(ex, "t1", "resident-active",
+                             "Paid", now - 100000)
+        out = ex.demote_resident("t1", now=now)
+        self.assertFalse(out["demoted"])
+        self.assertEqual(out["reason"], "not_demotable")
+        out = ex.demote_resident("ghost", now=now)
+        self.assertFalse(out["demoted"])
+        self.assertEqual(out["reason"], "not_demotable")
+
+    def test_sweep_demote_aborted_on_sync_hook_failure(self):
+        ex = self._lease_exec()
+        ex.resident_sync_hook = lambda tenant, vm: False
+        now = 1800000000.0
+        vm = self._write_resident(ex, "t1", "resident-paused",
+                                  "Paid", now - 61 - 121)
+        self._live_vm(ex, vm)
+        self.vm_states[vm] = "Paused"
+        out = ex.sweep_resident_leases(now=now)
+        self.assertEqual(out["demoted"], [])
+        self.assertEqual(out["aborted"], ["t1"])
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-paused")
+
+
+# -- 排队超时拒绝（第 7 项）--------------------------------------------
+class QueueTest(HybridBCase):
+    def _queue_exec(self):
+        return self.make_exec(env_extra={
+            "CHV_MAX_VMS": "1",
+            "CHV_QUEUE_TIMEOUT_SECONDS": "10",
+            "CHV_QUEUE_POLL_SECONDS": "2"})
+
+    def test_wait_for_capacity_timeout_rejects(self):
+        ex = self._queue_exec()
+        occupant = FakeProc(alive=True)
+        ex._vms["chv-other"] = occupant
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._wait_for_capacity(self._task(), exclude="chv-abc123")
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+        # 轮询 10/2 = 5 次 sleeper 调用后才拒绝。
+        self.assertEqual(len(self.sleeps), 5)
+        # 不抢占：占位 VM 仍活着；队列计数已归零。
+        self.assertIsNone(occupant.poll())
+        self.assertEqual(ex.queue_depth(), 0)
+
+    def test_wait_for_capacity_free_passes_immediately(self):
+        ex = self._queue_exec()
+        ex._wait_for_capacity(self._task(), exclude="chv-abc123")
+        self.assertEqual(self.sleeps, [])
+
+    def test_run_with_vm_queues_before_wake(self):
+        ex = self.make_exec(env_extra={
+            "CHV_MAX_VMS": "2",
+            "CHV_QUEUE_TIMEOUT_SECONDS": "10",
+            "CHV_QUEUE_POLL_SECONDS": "2"})
+        ex._vms["chv-other1"] = FakeProc(alive=True)
+        ex._vms["chv-other2"] = FakeProc(alive=True)
+        # 常驻租户排队：档位与租约原样保留，不升降档。
+        # （max_vms=2 → 常驻上限 1，promote 恰好占满 1 槽。）
+        ex.promote_to_resident("tq", "Paid")
+        wake_calls = []
+        ex.wake_vm = lambda *a, **k: wake_calls.append(a)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._run_with_vm(self._task(project="tq"),
+                            ["dispatch", "--name", "tq",
+                             "--task", "hi"])
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+        self.assertEqual(wake_calls, [])
+        record = read_persistent_vm(ex.state_dir, "tq")
+        self.assertEqual(record["tier"], "Paid")
+        self.assertEqual(record["state"], "resident-active")
+
+    def test_wait_for_resident_slot_timeout(self):
+        ex = self.make_exec(env_extra={
+            "CHV_MAX_VMS": "4",
+            "CHV_QUEUE_TIMEOUT_SECONDS": "10",
+            "CHV_QUEUE_POLL_SECONDS": "2"})
+        ex.promote_to_resident("t1", "Paid")
+        ex.promote_to_resident("t2", "Heavy")
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._wait_for_resident_slot("t3")
+        self.assertEqual(ctx.exception.kind, "host_capacity_exceeded")
+        # 不抢占、不挪池：前两个常驻原样。
+        self.assertEqual(ex._resident_count(), 2)
+        self.assertEqual(
+            read_persistent_vm(ex.state_dir, "t1")["state"],
+            "resident-active")
+
+    def test_wait_for_resident_slot_free_passes(self):
+        ex = self.make_exec(env_extra={"CHV_MAX_VMS": "4"})
+        ex.promote_to_resident("t1", "Paid")
+        ex._wait_for_resident_slot("t2")
+        self.assertEqual(self.sleeps, [])
+
+
+# -- 升级提示点去重（第 8 项）------------------------------------------
+class UpgradeHintTest(HybridBCase):
+    def test_hint_at_100_tasks_in_30d_and_dedup(self):
+        ex = self.make_exec()
+        now = 1800000000.0
+        hinted_at = None
+        for i in range(100):
+            out = ex.record_pool_task("tfree", tier="Free",
+                                      now=now - 1000 + i * 10)
+            if out["hinted"]:
+                hinted_at = i
+        self.assertEqual(hinted_at, 99)
+        self.assertEqual(
+            self._event_names().count("resident_upgrade_hint"), 1)
+        detail = [d for e, d, _k in self.events
+                  if e == "resident_upgrade_hint"][0]
+        self.assertIn("host_id=host-test-1", detail)
+        self.assertIn("tenant=tfree", detail)
+        self.assertIn("tier=Free", detail)
+        self.assertIn("30d_tasks=100", detail)
+        # 第 101 次：一周内同一租户不再记第二次。
+        out = ex.record_pool_task("tfree", tier="Free", now=now + 10)
+        self.assertFalse(out["hinted"])
+        self.assertEqual(
+            self._event_names().count("resident_upgrade_hint"), 1)
+        # 一周后：可再记一次。
+        out = ex.record_pool_task("tfree", tier="Free",
+                                  now=now + 8 * 86400)
+        self.assertTrue(out["hinted"])
+        self.assertEqual(
+            self._event_names().count("resident_upgrade_hint"), 2)
+
+    def test_hint_on_7day_streak(self):
+        ex = self.make_exec()
+        now = 1800000000.0
+        today = int(now // 86400)
+        hinted = False
+        for day_offset in range(6, -1, -1):
+            day_start = (today - day_offset) * 86400 + 3600
+            for j in range(3):
+                out = ex.record_pool_task(
+                    "tstreak", tier="Free",
+                    now=day_start + j * 60)
+                hinted = hinted or out["hinted"]
+        self.assertTrue(hinted)
+        self.assertEqual(
+            self._event_names().count("resident_upgrade_hint"), 1)
+        detail = [d for e, d, _k in self.events
+                  if e == "resident_upgrade_hint"][0]
+        self.assertIn("7d_streak_3_per_day", detail)
+
+    def test_no_hint_below_streak(self):
+        ex = self.make_exec()
+        now = 1800000000.0
+        today = int(now // 86400)
+        for day_offset in range(5, -1, -1):
+            day_start = (today - day_offset) * 86400 + 3600
+            for j in range(3):
+                out = ex.record_pool_task(
+                    "tshort", tier="Free", now=day_start + j * 60)
+                self.assertFalse(out["hinted"])
+        self.assertEqual(self._event_names(), [])
+
+    def test_paid_tier_not_recorded(self):
+        ex = self.make_exec()
+        out = ex.record_pool_task("tpaid", tier="Paid",
+                                  now=1800000000.0)
+        self.assertFalse(out["recorded"])
+        self.assertFalse(out["hinted"])
+        self.assertFalse(os.path.exists(
+            ex._pool_task_file("tpaid")))
+
+
+# -- 网关分流（第 9 项）-------------------------------------------------
+class GatewaySplitTest(HybridBCase):
+    def test_gateway_allowed_resident_paid(self):
+        ex = self.make_exec()
+        ex.promote_to_resident("t1", "Paid")
+        task = self._task(project="t1")
+        self.assertTrue(ex.gateway_allowed_for_task(
+            task, model_provider="sub2api"))
+        self.assertTrue(ex.gateway_allowed_for_task(
+            dict(task, _gateway_provider="sub2api")))
+
+    def test_gateway_allowed_hibernated_resident(self):
+        ex = self.make_exec()
+        now = 1800000000.0
+        self._write_resident(ex, "t1", "hibernated", "Paid", now - 10)
+        self.assertTrue(ex.gateway_allowed_for_task(
+            self._task(project="t1"), model_provider="sub2api"))
+
+    def test_gateway_denied_pool_tenants(self):
+        ex = self.make_exec()
+        # 无登记：池。
+        self.assertFalse(ex.gateway_allowed_for_task(
+            self._task(project="tpool"), model_provider="sub2api"))
+        # 常驻态但 tier 是 Free：无常驻资格，按池算。
+        now = 1800000000.0
+        self._write_resident(ex, "tfree", "resident-active",
+                             "Free", now - 10)
+        self.assertFalse(ex.gateway_allowed_for_task(
+            self._task(project="tfree"), model_provider="sub2api"))
+        # demoted：租约已收回，按池算。
+        self._write_resident(ex, "tdem", "demoted", "Paid", now - 10)
+        self.assertFalse(ex.gateway_allowed_for_task(
+            self._task(project="tdem"), model_provider="sub2api"))
+        # legacy paused 登记：不是常驻态，按池算。
+        vm = ex.persistent_vm_name("tleg")
+        write_persistent_vm(ex.state_dir, "tleg", {
+            "vm": vm, "tenant_id": "tleg", "state": "paused",
+            "origin": "task", "gitea_synced": True,
+            "updated_at": now})
+        self.assertFalse(ex.gateway_allowed_for_task(
+            self._task(project="tleg"), model_provider="sub2api"))
+
+    def test_gateway_check_non_gateway_not_blocked(self):
+        ex = self.make_exec()
+        # 非网关任务不由本门拦（model_provider 不是网关）。
+        self.assertTrue(ex.gateway_allowed_for_task(
+            self._task(project="tpool"), model_provider=None))
+        self.assertTrue(ex.gateway_allowed_for_task(
+            self._task(project="tpool"), model_provider="xcmapi"))
+
+    def test_run_with_vm_gateway_pool_denied_before_wake(self):
+        ex = self.make_exec()
+        wake_calls = []
+        ex.wake_vm = lambda *a, **k: wake_calls.append(a)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex._run_with_vm(self._task(project="tpool"),
+                            ["dispatch", "--name", "tpool",
+                             "--task", "hi"],
+                            model_provider="sub2api")
+        self.assertEqual(ctx.exception.kind, "gateway_pool_denied")
+        self.assertEqual(wake_calls, [])
+
+    def test_run_with_vm_gateway_resident_passes_gate(self):
+        ex = self.make_exec()
+        ex.promote_to_resident("t1", "Paid")
+        # 常驻租户的网关任务过门不抛错（全程 dispatch 由既有
+        # 测试覆盖，这里只断言门本身放行）。
+        ex._check_gateway_vm_class(
+            self._task(project="t1"), model_provider="sub2api")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_chv_hybrid_c.py

@@ -0,0 +1,573 @@
+"""CHV 混合模式 C 轨测试（feat-chv-hybrid-c）。
+
+只覆盖本轨的三个改动点（DESIGN-HYBRID.md §9.1 第 10–13 项）：
+- autoscaler desired_workers 作用域扩展（§1.2，W=2）；
+- 回收同步失败即任务标同步失败、VM 不回收（§3.3）；
+- 审计事件目录（§7，事件名与设计原文一致、带 host_id、
+  不带密钥值）与 orphans facade 核实（§1.3/ADR-5/ADR-12）。
+
+只用标准库与 mock runner/launcher/api_client，绝不启动
+真实 VM、不联网、不碰生产库。真机验证（WSL 3 台排布、
+热恢复 ≤2 秒等）不在本文件声称已验证。
+"""
+
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.autoscaler import (
+        CHV_WARM_POOL_PER_HOST,
+        Autoscaler,
+        desired_workers,
+        pool_target_for_hosts,
+    )
+    from cloud.chv import hybrid_events
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.orphans import reconcile_orphans_for_executor
+    from cloud.server import init_db
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.autoscaler import (
+        CHV_WARM_POOL_PER_HOST,
+        Autoscaler,
+        desired_workers,
+        pool_target_for_hosts,
+    )
+    from cloud.chv import hybrid_events
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.orphans import reconcile_orphans_for_executor
+    from cloud.server import init_db
+
+
+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
+
+
+# -- 11. desired_workers + 池目标数 --------------------------------------
+class DesiredWorkersPoolTestCase(unittest.TestCase):
+    def test_pool_target_zero_unchanged(self):
+        # pool_target=0（缺省）时与旧公式一字不差。
+        self.assertEqual(desired_workers(0, 2, 1, 2), 1)
+        self.assertEqual(desired_workers(0, 2, 0, 2), 0)
+        self.assertEqual(desired_workers(1, 2, 1, 2), 1)
+        self.assertEqual(desired_workers(3, 2, 1, 2), 2)
+        self.assertEqual(desired_workers(10, 2, 1, 2), 2)
+        self.assertEqual(desired_workers(5, 1, 1, 3), 3)
+        self.assertEqual(desired_workers(5, 1, 1, 3, pool_target=0), 3)
+
+    def test_pool_target_additive(self):
+        # 期望值 = fleet 需要量 + 池目标数（§1.2）。
+        self.assertEqual(desired_workers(0, 2, 1, 20, pool_target=12), 13)
+        self.assertEqual(desired_workers(4, 2, 1, 20, pool_target=12), 14)
+        self.assertEqual(desired_workers(3, 2, 1, 20, pool_target=2), 4)
+        # depth=0 且 min_workers=0 时池目标仍维持（预热不因
+        # 队列空而归零）。
+        self.assertEqual(desired_workers(0, 2, 0, 20, pool_target=12), 12)
+
+    def test_pool_target_capped_by_max_workers(self):
+        # 池目标数计入 max_workers 上限：配不下即截断，不静默多扩。
+        self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=12), 2)
+        self.assertEqual(desired_workers(10, 2, 1, 3, pool_target=12), 3)
+
+    def test_pool_target_negative_or_none_is_zero(self):
+        self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=-5), 1)
+        self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=None), 1)
+
+
+class PoolTargetForHostsTestCase(unittest.TestCase):
+    def test_w_is_two_per_host(self):
+        self.assertEqual(CHV_WARM_POOL_PER_HOST, 2)
+
+    def test_pool_target_is_w_times_online_hosts(self):
+        self.assertEqual(pool_target_for_hosts(6), 12)
+        self.assertEqual(pool_target_for_hosts(22), 44)
+        self.assertEqual(pool_target_for_hosts(1), 2)
+        self.assertEqual(pool_target_for_hosts(0), 0)
+
+    def test_invalid_hosts_is_zero(self):
+        self.assertEqual(pool_target_for_hosts(-1), 0)
+        self.assertEqual(pool_target_for_hosts(None), 0)
+        self.assertEqual(pool_target_for_hosts("not-a-number"), 0)
+
+
+def _insert_task(db_path, task_id):
+    now = time.time()
+    import sqlite3
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, 'p1', 'x', 'queued', NULL, '[]', 0,"
+            " 'tester', ?, ?, NULL)",
+            (task_id, now, now))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class AutoscalerPoolReconcileTestCase(unittest.TestCase):
+    """L1 其余语义不变：30 秒一轮（interval 缺省）、scale up
+    fast / scale down slow（scale_down_after 冷却）原样复用，
+    只验证期望值多了池目标数。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.state_file = os.path.join(self.tmp.name, "state.json")
+        self.fake_cmd = [sys.executable, "-c",
+                         "import time; time.sleep(120)"]
+
+    def tearDown(self):
+        if hasattr(self, "scaler"):
+            for wid in list(self.scaler.workers):
+                try:
+                    self.scaler._stop_worker(wid)
+                except Exception:
+                    pass
+        self.tmp.cleanup()
+
+    def _scaler(self, **kwargs):
+        kwargs.setdefault("min_workers", 1)
+        kwargs.setdefault("max_workers", 5)
+        kwargs.setdefault("per_worker_capacity", 2)
+        kwargs.setdefault("scale_down_after", 1)
+        kwargs.setdefault("drain_timeout", 5)
+        kwargs.setdefault("state_file", self.state_file)
+        self.scaler = Autoscaler(
+            self.db_path, [], worker_cmd=self.fake_cmd, **kwargs)
+        return self.scaler
+
+    def test_empty_queue_still_holds_pool(self):
+        scaler = self._scaler(pool_target=2)
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (0, 3, 3))
+
+    def test_depth_plus_pool_and_scale_down_slow(self):
+        scaler = self._scaler(pool_target=2, scale_down_after=2)
+        for tid in ("A", "B", "C", "D"):
+            _insert_task(self.db_path, tid)
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (4, 4, 4))
+        # 任务全完成：深度归零但池目标仍在，期望值回落到 3
+        # 而不是 1；缩容仍须等满 2 轮冷却（scale down slow）。
+        import sqlite3
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("UPDATE tasks SET status='completed'")
+            conn.commit()
+        finally:
+            conn.close()
+        scaler.reconcile()
+        self.assertEqual(len(scaler.workers), 4)
+        self.assertEqual(scaler.last_action, "scale_down_pending_1/2")
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (0, 3, 3))
+
+    def test_online_hosts_derives_pool_target(self):
+        scaler = self._scaler(online_hosts=6, max_workers=20)
+        self.assertEqual(scaler.pool_target, 12)
+        depth, desired, actual = scaler.reconcile()
+        self.assertEqual((depth, desired, actual), (0, 13, 13))
+
+    def test_explicit_pool_target_wins_over_hosts(self):
+        scaler = self._scaler(pool_target=2, online_hosts=6)
+        self.assertEqual(scaler.pool_target, 2)
+
+    def test_state_file_records_pool_target(self):
+        scaler = self._scaler(pool_target=2)
+        scaler.reconcile()
+        with open(self.state_file) as fh:
+            state = json.load(fh)
+        self.assertEqual(state["pool_target"], 2)
+        self.assertEqual(state["desired"], 3)
+
+    def test_main_passes_pool_flags(self):
+        import cloud.autoscaler as autoscaler_mod
+        with mock.patch.object(
+                autoscaler_mod, "Autoscaler") as mock_cls, \
+                mock.patch("signal.signal"):
+            mock_cls.return_value.run_loop = mock.Mock()
+            autoscaler_mod.main([
+                "--db", self.db_path, "--pool-target", "12",
+                "--", "--interval", "10"])
+        kwargs = mock_cls.call_args.kwargs
+        self.assertEqual(kwargs["pool_target"], 12)
+        with mock.patch.object(
+                autoscaler_mod, "Autoscaler") as mock_cls, \
+                mock.patch("signal.signal"):
+            mock_cls.return_value.run_loop = mock.Mock()
+            autoscaler_mod.main([
+                "--db", self.db_path, "--online-hosts", "6",
+                "--", "--interval", "10"])
+        kwargs = mock_cls.call_args.kwargs
+        self.assertEqual(kwargs["online_hosts"], 6)
+        self.assertEqual(kwargs["warm_pool_per_host"], 2)
+
+
+# -- 12. 回收同步 ---------------------------------------------------------
+class RecycleSyncCase(unittest.TestCase):
+    """mock runner/launcher/api_client，不启动真实 VM。
+    push_rc 控制客内 `chv-sync push` 的返回值。"""
+
+    TOKEN = "SECRET-TOKEN-XYZ-DO-NOT-LEAK"
+
+    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.audit = []
+        self.proc = FakeProc(alive=True)
+        self.push_rc = 0
+        self.push_stdout = "已 push"
+        self.push_stderr = ""
+        self.raise_on_push = False
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _runner(self, cmd, timeout=None):
+        self.calls.append(list(cmd))
+        joined = " ".join(str(c) for c in cmd)
+        if "chv-sync" in joined:
+            if self.raise_on_push:
+                raise RuntimeError("ssh broken")
+            return FakeResult(self.push_rc, self.push_stdout,
+                              self.push_stderr)
+        return FakeResult(0, "", "")
+
+    def _make_exec(self, sync_configured=True):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_HOST_KEY_MODE": "keyscan",
+               "CHV_SSH_FORWARD": "1"}
+        if sync_configured:
+            env.update({
+                "CHV_GITEA_URL": "https://gitea.tose.top",
+                "CHV_GITEA_USER": "tester",
+                "CHV_GITEA_TOKEN": self.TOKEN,
+                "CHV_SYNC_REPO": "tester/repo",
+            })
+        ex = CloudHypervisorExecutor(
+            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=self._runner,
+            launcher=lambda cmd: self.proc,
+            api_client=lambda *a, **k: (200, {}),
+            sleeper=lambda s: None, env=env)
+        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 _task(self):
+        return {"id": "abc123", "prompt": "x", "project_id": "proj1"}
+
+    def _mark_vm_running(self, ex, vm_name="chv-abc123"):
+        os.makedirs(self.state_dir, exist_ok=True)
+        with open(os.path.join(self.state_dir, vm_name + ".sock"),
+                  "w") as fh:
+            fh.write("")
+        ex._vms[vm_name] = self.proc
+
+    def _audit_events(self):
+        return [e for e, _t, _d, _k in self.audit]
+
+    def _audit_details(self):
+        return [d for _e, _t, d, _k in self.audit]
+
+    def test_no_sync_configured_recycles_as_before(self):
+        # 未配 Gitea 同步：无物可推，走旧回收路径（回归）。
+        ex = self._make_exec(sync_configured=False)
+        self._mark_vm_running(ex)
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "recycled")
+        self.assertTrue(out["sync_ok"])
+        self.assertFalse(out["vm_retained"])
+        self.assertTrue(self.proc.terminated)
+        self.assertIn("vm_recycled", self._audit_events())
+        self.assertFalse(any("chv-sync" in " ".join(c)
+                             for c in self.calls))
+
+    def test_sync_push_ok_then_recycle(self):
+        ex = self._make_exec()
+        self._mark_vm_running(ex)
+        self.push_rc = 0
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "recycled")
+        self.assertTrue(out["sync_ok"])
+        self.assertFalse(out["vm_retained"])
+        self.assertTrue(self.proc.terminated)
+        self.assertIn("vm_recycled", self._audit_events())
+        self.assertTrue(any("chv-sync" in " ".join(c)
+                            for c in self.calls))
+        self.assertNotIn("vm_recycle_sync_failed",
+                         self._audit_events())
+
+    def test_sync_push_failed_marks_sync_failed_and_retains_vm(self):
+        # §3.3 新增语义：同步失败即任务标同步失败、VM 不回收。
+        ex = self._make_exec()
+        self._mark_vm_running(ex)
+        self.push_rc = 1
+        self.push_stderr = "push 被拒绝（远端已有新提交）"
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "sync_failed")
+        self.assertEqual(out["status"], "sync_failed")
+        self.assertFalse(out["sync_ok"])
+        self.assertTrue(out["vm_retained"])
+        self.assertIn("push", out["sync_error"])
+        # VM 保留：进程未停、socket 未删、未发 recycled/shutdown。
+        self.assertFalse(self.proc.terminated)
+        self.assertTrue(os.path.exists(os.path.join(
+            self.state_dir, "chv-abc123.sock")))
+        self.assertNotIn("vm_recycled", self._audit_events())
+        self.assertNotIn("vm_shutdown", self._audit_events())
+        self.assertNotIn("vm_destroy", self._audit_events())
+        # 记审计：事件名正确、带 host_id、不带密钥值。
+        self.assertIn("vm_recycle_sync_failed", self._audit_events())
+        details = " ".join(self._audit_details())
+        self.assertIn("host_id=", details)
+        self.assertNotIn(self.TOKEN, details)
+        self.assertNotIn(self.TOKEN, json.dumps(self.audit))
+
+    def test_sync_ssh_error_also_retains_vm(self):
+        ex = self._make_exec()
+        self._mark_vm_running(ex)
+        self.raise_on_push = True
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "sync_failed")
+        self.assertTrue(out["vm_retained"])
+        self.assertFalse(self.proc.terminated)
+        self.assertIn("vm_recycle_sync_failed", self._audit_events())
+
+    def test_sync_failed_then_retry_after_push_ok(self):
+        # 人工/重试路径：push 修好后再次回收即成功，不卡死。
+        ex = self._make_exec()
+        self._mark_vm_running(ex)
+        self.push_rc = 1
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "sync_failed")
+        self.push_rc = 0
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "recycled")
+        self.assertTrue(self.proc.terminated)
+
+    def test_sync_skip_flag_recycles_despite_push_failure(self):
+        ex = self._make_exec()
+        self._mark_vm_running(ex)
+        self.push_rc = 1
+        out = ex.recycle_vm(self._task(), sync=False)
+        self.assertEqual(out["state"], "recycled")
+        self.assertTrue(self.proc.terminated)
+
+    def test_vm_not_running_does_not_block_cleanup(self):
+        # VM 已死：没有活工作树可推，不拿死 VM 卡清理。
+        ex = self._make_exec()
+        out = ex.recycle_vm(self._task())
+        self.assertEqual(out["state"], "recycled")
+
+    def test_recycle_persistent_vm_sync_failed_keeps_disks(self):
+        ex = self._make_exec()
+        self._mark_vm_running(ex, vm_name="chv-persist-proj1")
+        self.push_rc = 1
+        out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"})
+        self.assertEqual(out["state"], "sync_failed")
+        self.assertEqual(out["status"], "sync_failed")
+        self.assertTrue(out["vm_retained"])
+        self.assertEqual(out["deleted_disks"], [])
+        self.assertFalse(self.proc.terminated)
+        self.assertNotIn("vm_persistent_recycled",
+                         self._audit_events())
+        self.assertIn("vm_recycle_sync_failed", self._audit_events())
+
+    def test_recycle_vm_missing_id_still_raises(self):
+        ex = self._make_exec()
+        with self.assertRaises(ValueError):
+            ex.recycle_vm({})
+
+
+# -- 13. 审计事件目录 ------------------------------------------------------
+class HybridEventsTestCase(unittest.TestCase):
+    def test_design_event_names_verbatim(self):
+        # 事件名与 DESIGN-HYBRID.md §9.1-13 原文一字不差。
+        self.assertEqual(hybrid_events.HYBRID_EVENTS, (
+            "warm_pool_acquire",
+            "warm_pool_miss",
+            "warm_pool_release",
+            "vm_resident_suspended",
+            "vm_resident_demoted",
+            "vm_resident_promoted",
+            "resident_upgrade_hint",
+            "snapshot_tenant_mismatch",
+            "vm_restore_denied",
+        ))
+
+    def test_every_event_requires_host_id(self):
+        for event in hybrid_events.HYBRID_EVENTS:
+            self.assertIn("host_id",
+                          hybrid_events.REQUIRED_FIELDS[event],
+                          event)
+        self.assertIn(
+            "host_id",
+            hybrid_events.REQUIRED_FIELDS[
+                hybrid_events.VM_RECYCLE_SYNC_FAILED])
+
+    def test_build_event_defaults_host_id(self):
+        fields = hybrid_events.build_event(
+            "warm_pool_miss", {"tenant_id": "t1"})
+        self.assertEqual(fields["host_id"], socket.gethostname())
+        self.assertEqual(fields["tenant_id"], "t1")
+
+    def test_build_event_rejects_missing_required(self):
+        with self.assertRaises(ValueError):
+            hybrid_events.build_event("warm_pool_acquire", {})
+        with self.assertRaises(ValueError):
+            hybrid_events.build_event(
+                "vm_resident_suspended",
+                {"host_id": "h1", "tenant_id": "t1"})
+        with self.assertRaises(ValueError):
+            hybrid_events.build_event("no_such_event",
+                                      {"host_id": "h1"})
+
+    def test_secret_field_rejected_without_leaking_value(self):
+        sentinel = "SECRET-TOKEN-XYZ-DO-NOT-LEAK"
+        with self.assertRaises(ValueError) as ctx:
+            hybrid_events.build_event(
+                "warm_pool_acquire",
+                {"vm": "chv-x", "GITEA_TOKEN": sentinel})
+        self.assertNotIn(sentinel, str(ctx.exception))
+        self.assertIn("GITEA_TOKEN", str(ctx.exception))
+
+    def test_format_detail_has_host_id_and_no_secret(self):
+        detail = hybrid_events.format_detail(
+            "vm_resident_promoted",
+            {"tenant_id": "t1", "vm": "chv-persist-t1"})
+        self.assertIn("host_id=%s" % socket.gethostname(), detail)
+        self.assertIn("vm=chv-persist-t1", detail)
+        self.assertIn("tenant_id=t1", detail)
+
+
+# -- 10. orphans facade 核实 ----------------------------------------------
+class OrphansFacadeTestCase(unittest.TestCase):
+    def test_facade_exists_and_cleans_only_orphans(self):
+        # facade 已存在：读 registry → 只清理 warm/assigned
+        # 条目（只清理、不分配），非法条目丢弃不连坐。
+        with tempfile.TemporaryDirectory() as tmp:
+            registry = {
+                "warm-1": {"state": "warm", "vm": "chv-a",
+                           "task_id": "t1"},
+                "asg-1": {"state": "assigned", "vm": "chv-b",
+                          "task_id": "t2"},
+                "bad-1": {"state": "warm"},
+                "dead-1": {"state": "recycled", "vm": "chv-c"},
+            }
+            with open(os.path.join(tmp, "warm-pool.json"),
+                      "w") as fh:
+                json.dump(registry, fh)
+
+            class Exec:
+                state_dir = tmp
+                cleaned = []
+
+                def shutdown_vm(self, vm):
+                    self.cleaned.append(vm)
+                    return True
+
+                def _emit(self, event, task, detail, kind=None):
+                    pass
+
+            ex = Exec()
+            cleaned = reconcile_orphans_for_executor(ex)
+            self.assertEqual(cleaned, ["chv-a", "chv-b"])
+            self.assertEqual(ex.cleaned, ["chv-a", "chv-b"])
+            with open(os.path.join(tmp, "warm-pool.json")) as fh:
+                self.assertEqual(json.load(fh), {})
+
+    def test_facade_quarantines_corrupt_registry(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "warm-pool.json")
+            with open(path, "w") as fh:
+                fh.write("{not json")
+
+            class Exec:
+                state_dir = tmp
+
+                def shutdown_vm(self, vm):
+                    raise AssertionError("must not be called")
+
+            self.assertEqual(reconcile_orphans_for_executor(Exec()), [])
+            self.assertTrue(os.path.exists(path + ".corrupt"))
+
+    def test_warm_pool_still_not_wired_to_dispatch(self):
+        from cloud.chv.experimental_pool import WarmPool
+        self.assertIs(WarmPool.wired_to_dispatch, False)
+
+    def test_orphans_module_does_not_import_experimental_pool(self):
+        # 生产 facade 不 import 实验模块（ADR-12）：全新解释器
+        # 里只 import orphans，experimental_pool 不得被加载。
+        code = (
+            "import cloud.chv.orphans, sys;"
+            " assert 'cloud.chv.experimental_pool' not in sys.modules;"
+            " print('orphans-clean')"
+        )
+        proc = subprocess.run(
+            [sys.executable, "-c", code],
+            capture_output=True, text=True, cwd=str(
+                Path(__file__).resolve().parents[2]))
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("orphans-clean", proc.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_chv_opt_stage1.py

@@ -412,8 +412,11 @@ class CrossTenantRestoreTest(ChvOpt1Case):
             ex._validate_restore(task_b, url)
         self.assertEqual(ctx.exception.kind,
                          "snapshot_tenant_mismatch")
+        # 严格校验默认开启（设计 §2.2）：跨租户拒绝走设计事件
+        # vm_restore_denied；legacy 的 vm_restore_tenant_denied
+        # 只在显式关闭严格校验时才出现。
         self.assertTrue(any(
-            e == "vm_restore_tenant_denied" for e, _ in events))
+            e == "vm_restore_denied" for e, _ in events))
 
     def test_same_tenant_restore_passes_validation(self):
         url = self._snap_url("chv-a")
@@ -426,7 +429,11 @@ class CrossTenantRestoreTest(ChvOpt1Case):
     def test_legacy_unmarked_snapshot_warns_but_passes(self):
         url = self._snap_url("chv-a", "old")
         self._record_snapshot("chv-a", url, None)
-        ex = self.make_exec()
+        # legacy 警告放行只在显式关闭严格校验时保留
+        # （CHV_SNAPSHOT_STRICT_TENANT=0；默认开启时未标记
+        # 快照即拒绝，见 test_unmarked_snapshot_denied_in_hybrid_mode）。
+        ex = self.make_exec(env_extra={
+            "CHV_SNAPSHOT_STRICT_TENANT": "0"})
         events = []
         ex.audit_callback = lambda e, t, d, k: events.append((e, k))
         snapshot_url, _net = ex._validate_restore(

cloud/tests/test_chv_v3_fixes.py

@@ -38,12 +38,14 @@ from pathlib import Path
 from unittest import mock
 
 try:
+    from cloud.chv.disk import record_snapshot
     from cloud.chv.executor import CloudHypervisorExecutor
     from cloud.chv.hostkeys import write_seed
     from cloud.chv.ssh_transport import remote_command
     from cloud.sandbox import ChvFailure
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.disk import record_snapshot
     from cloud.chv.executor import CloudHypervisorExecutor
     from cloud.chv.hostkeys import write_seed
     from cloud.chv.ssh_transport import remote_command
@@ -195,6 +197,14 @@ class V3Case(unittest.TestCase):
                 fh.write(content)
         return "file://" + path, path
 
+    def _register_snapshot(self, url, tenant="proj1"):
+        # 严格快照校验（设计 §2.2，默认开启）：restore 测试的
+        # 快照须登记带 tenant 且与 _task() 的 project_id 一致；
+        # 本 helper 只登记，不改变测试本身的意图（盘预检 /
+        # source_url / 清旧 VMM 仍由各自断言覆盖）。
+        record_snapshot(self.state_dir, "chv-abc123", url, 20.0,
+                        tenant_id=tenant)
+
     def _spawn_fake_vmm(self, sock_path, argv0="cloud-hypervisor"):
         """真实 spawn 一个长眠进程，argv[0] 伪装成 VMM，
         命令行带 --api-socket <sock>（F2/V-3③ 的扫描目标）。"""
@@ -223,6 +233,7 @@ class SnapshotV3Test(V3Case):
 
     def test_restore_still_passes_source_url(self):
         url, _ = self._snap_dir()
+        self._register_snapshot(url)
         ex = self.make_exec()
         out = ex.restore_vm(self._task(), url)
         self.assertEqual(out["mode"], "restored")
@@ -248,6 +259,7 @@ class SnapshotV3Test(V3Case):
 class RestorePrepareV3Test(V3Case):
     def test_restore_kills_registered_old_vmm_and_socket(self):
         url, _ = self._snap_dir()
+        self._register_snapshot(url)
         ex = self.make_exec()
         ex.wake_vm(self._task())
         old_proc = self.proc
@@ -265,6 +277,7 @@ class RestorePrepareV3Test(V3Case):
 
     def test_restore_kills_unregistered_vmm_by_socket(self):
         url, _ = self._snap_dir()
+        self._register_snapshot(url)
         ex = self.make_exec()
         sock = os.path.join(self.state_dir, "chv-abc123.sock")
         os.makedirs(self.state_dir, exist_ok=True)
@@ -314,6 +327,7 @@ class HibernateDiskV3Test(V3Case):
             "config.json": json.dumps(
                 {"disks": [{"path": missing_disk}]}).encode(),
             "state.json": b"{}"})
+        self._register_snapshot(url)
         with self.assertRaises(ChvFailure) as ctx:
             ex.restore_vm(self._task(), url)
         self.assertEqual(ctx.exception.kind, "snapshot_disk_missing")
@@ -327,6 +341,7 @@ class HibernateDiskV3Test(V3Case):
             "config.json": json.dumps(
                 {"disks": [{"path": disk}]}).encode(),
             "state.json": b"{}"})
+        self._register_snapshot(url)
         out = ex.restore_vm(self._task(), url)
         self.assertEqual(out["mode"], "restored")
 

cloud/tests/test_cloud_hypervisor.py

@@ -20,6 +20,7 @@ from pathlib import Path
 from unittest import mock
 
 try:
+    from cloud.chv.disk import record_snapshot
     from cloud.chv.experimental_pool import WarmPool
     from cloud.sandbox import (
         ChvFailure,
@@ -28,6 +29,7 @@ try:
     )
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.disk import record_snapshot
     from cloud.chv.experimental_pool import WarmPool
     from cloud.sandbox import (
         ChvFailure,
@@ -406,6 +408,10 @@ class LifecycleTest(ChvCase):
         # 路径 + 声明快照目录（生产默认 provisioned + 白名单
         # 目录，旧 file:///snap/1 不再直通）。
         ex = self.make_exec(env_extra={"CHV_SNAPSHOT_DIR": "/snap"})
+        # 严格快照校验（设计 §2.2，默认开启）：恢复须登记带
+        # tenant 且与 _task() 的 project_id（proj1）一致。
+        record_snapshot(self.state_dir, "chv-abc123",
+                        "file:///snap/1", 20.0, tenant_id="proj1")
         out = ex.restore_vm(self._task(), "file:///snap/1")
         self.assertEqual(out["mode"], "restored")
         self.assertTrue(out["ready"])