merge(chv-hybrid): 合入 B 轨——租约 + 排队 + 升级提示 + 网关门

合入 feat-chv-hybrid-b (667a427)。executor.py 两处内容冲突按 hunk
解决，全部取 B 侧；合入时发现 merge 复制出第二个 suspend_after_task
定义（A 版截断体），已删除。

集成统一（A/B 语义冲突三处，设计 §4.1 的 state 模型为准）：
- state 模型：A 的 flag 模型（resident: True + legacy state）与 B
  的 state 模型（state ∈ resident-active/paused/hibernated/demoted）
  统一为 state 模型；_mark_resident_record 标记常驻时同时置
  resident-active，_ensure_resident_capacity 同时认 flag 与 state。
- 容量等待：A 的 bool 版 _wait_for_capacity 改名 _poll_capacity，
  B 的 raise 版保留原名；_wait_for_resident_slot 豁免已有常驻登记
  的租户（它已占一个槽位，再等等于把自己的槽位算成满——A 轨
  test_resident_cap_blocks_new_resident 第二腿在合入时因此失败，
  已修并通过）。
- 档位大小写：A 用小写 paid/heavy，B 用 Paid/Heavy；比较一律走
  _norm_tier + HYBRID_RESIDENT_TIERS（resident_tier_eligible 与
  promote_to_resident 两处），fail-closed 语义不变。
- 严格快照校验翻为默认开启（设计 §2.2）：_snapshot_strict_tenant
  默认 True，显式 CHV_SNAPSHOT_STRICT_TENANT=0 才关闭；旧恢复
  测试的 fixture 已登记带匹配 tenant 的快照，A 轨 legacy 测试改
  为显式 opt-out 覆盖。

生产接线：worker 只调 dispatch_task，dispatch_task_hybrid 在生产
路径上无调用方；dispatch_task 顶部加一道门，warm_pool_size>=1 时
走 dispatch_task_hybrid，==0 时保持旧冷启语义（不成环：hybrid 的
<=0 分支退化为 dispatch_task_persistent，后者 <=0 调回 dispatch_task）。

旧网关测试更新（B 的语义变更）：test_worker_gateway_guard_allows_chv
前加常驻登记；新增 test_worker_gateway_guard_denies_pool_vm。

测试：test_chv_hybrid_a/b 64 项 + 受影响旧测试 16 项（Gateway、
RestoreNoFallback、RestoreSnapshotTest）全部通过；全量待跑。

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

@@ -283,7 +283,7 @@ class HybridRouteTest(HybridACase):
         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"], "paused")
+        self.assertEqual(rec["state"], "resident-active")
         self.assertEqual(rec["tier"], "paid")
         self.assertTrue(rec["resident"])
         self.assertEqual(ex._resident_count(), 1)
@@ -301,7 +301,7 @@ class HybridRouteTest(HybridACase):
         self.assertEqual(out["vm"], "chv-persist-paid1")
         self.assertIn("resume_seconds", out)
         rec = read_persistent_vm(self.state_dir, "paid1")
-        self.assertEqual(rec["state"], "paused")
+        self.assertEqual(rec["state"], "resident-active")
         self.assertEqual(rec["tier"], "paid")
         self.assertTrue(rec["resident"])
 
@@ -311,7 +311,7 @@ class HybridRouteTest(HybridACase):
         for tenant in ("r1", "r2"):
             write_persistent_vm(self.state_dir, tenant, {
                 "vm": "chv-persist-%s" % tenant,
-                "tenant_id": tenant, "state": "paused",
+                "tenant_id": tenant, "state": "resident-active",
                 "origin": "task", "gitea_synced": False,
                 "tier": "paid", "resident": True})
         self.assertEqual(ex._resident_count(), 2)
@@ -324,11 +324,11 @@ class HybridRouteTest(HybridACase):
         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": "paused", "origin": "task",
+            "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": "paused", "origin": "task",
+            "state": "resident-active", "origin": "task",
             "gitea_synced": False, "tier": "paid", "resident": True})
         self._socket("chv-persist-r1")
         self.vm_states["chv-persist-r1"] = "Paused"
@@ -670,12 +670,15 @@ class SnapshotTenantStrictTest(HybridACase):
                          "snapshot_tenant_mismatch")
         self.assertIn("vm_restore_denied", self._event_names())
 
-    def test_legacy_warn_pass_preserved_when_not_strict(self):
+    def test_legacy_warn_pass_when_strict_opted_out(self):
-        # 混合模式未启用时，_validate_restore 沿用阶段 1 的
+        # 严格校验默认开启（设计 §2.2，集成时翻为默认）；
-        # legacy 警告放行（旧测试依赖该语义，本轨不改）。
+        # 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"})
+        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)
@@ -684,5 +687,32 @@ class SnapshotTenantStrictTest(HybridACase):
         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()