feat(chv): 混合模式 B 轨——租约 + 排队 + 升级提示 + 网关门

- 登记扩展 tier/state/last_active_at/lease_expires_at；认领三重核对
  原样保留，新增 tier 校验（无 tier 行 = 无常驻资格，fail-closed；
  与 P11 用量门不对称已写进注释）
- 租约计时器：闲置 30 分钟挂起、再 24 小时降级；阈值经环境变量注入，
  生产默认值不变；worker 心跳轮询驱动 sweep_resident_leases；
  降级同步失败即中止、VM 保留
- 排队：300 秒超时后 host_capacity_exceeded 明确拒绝；不插队、
  不抢占常驻槽位、不升降档；常驻任务仍等常驻槽位；队列深度供 L1 读
- 升级提示点 resident_upgrade_hint：30 天 ≥100 次或连续 7 天每天 ≥3 次，
  一周内同一租户最多一次；频率只用于提示，绝不改路由
- 网关分流：常驻 VM 可接网关、池 VM 以 gateway_pool_denied 拒绝
  （worker 侧守卫自 P1-1 起已放行 CHV，分流 enforcement 落执行器）

cloud/chv/executor.py

@@ -31,6 +31,7 @@ import logging
 import os
 import re
 import shutil
+import socket
 import subprocess
 import threading
 import time
@@ -81,6 +82,46 @@ from cloud.executor_base import Executor
 _LOGGER = logging.getLogger("cloud.cloud_hypervisor")
 
 
+# -- 混合模式 B 轨（feat-chv-hybrid-b，设计 DESIGN-HYBRID.md §9.1
+#    第 5–9 项；已定值直接用，不许改回提案口径）-------------------
+# 常驻资格档位（§5）：只有 Paid/Heavy 有常驻资格，Free 无论多
+# 高频都没有。
+RESIDENT_TIERS = ("Paid", "Heavy")
+# 常驻租约状态（§4.1）：登记 state 字段取值。legacy 的
+# paused/running 是阶段 1 持久 VM 的旧状态，不是常驻态。
+RESIDENT_STATES = ("resident-active", "resident-paused",
+                   "hibernated", "demoted")
+# 常驻有效态（占槽、续期、认领、网关放行都只认这三个）：
+# demoted 已收回租约，不占槽、不续期、不复活。
+RESIDENT_ACTIVE_STATES = ("resident-active", "resident-paused",
+                          "hibernated")
+# 认领时可 resume 的登记状态：legacy paused + 常驻态里 VM 已
+# 暂停的两种（resident-active 的 VM 在任务间隙同样被
+# suspend_after_task 暂停；hibernated 的 VM 已停机，须走
+# restore，不进 resume 认领；demoted 已收回租约，不认领）。
+RESIDENT_CLAIMABLE_STATES = ("paused", "resident-active",
+                             "resident-paused")
+# 租约阈值（已定值，§4.2）：闲置 30 分钟挂起，挂起后再闲置
+# 24 小时降级（累计 24.5 小时无活动）。生产默认值不变；
+# 测试用环境变量注入短阈值（见 _resident_thresholds）。
+RESIDENT_IDLE_SUSPEND_SECONDS = 30 * 60
+RESIDENT_IDLE_DEMOTE_SECONDS = 24 * 3600
+# 每宿主常驻上限（已定值，§4.1）：宿主 VM 槽位的 50%。
+RESIDENT_MAX_FRACTION = 0.5
+# 排队超时（已定值，§3.4）：无可用宿主时等 300 秒再明确拒绝。
+QUEUE_TIMEOUT_SECONDS = 300
+QUEUE_POLL_SECONDS = 5
+# 升级提示点（已定值，§5）：30 天内池任务 ≥100 次，或连续 7 天
+# 每天 ≥3 次；一周内同一租户最多记一次。
+UPGRADE_HINT_WINDOW_DAYS = 30
+UPGRADE_HINT_MIN_TASKS = 100
+UPGRADE_HINT_STREAK_DAYS = 7
+UPGRADE_HINT_MIN_PER_DAY = 3
+UPGRADE_HINT_DEDUP_DAYS = 7
+# 网关 provider（§6）：目前唯一的长连接网关是 sub2api。
+GATEWAY_PROVIDERS = ("sub2api",)
+
+
 def _pid_gone(pid):
     """进程是否已不在（/proc 消失，或已成僵尸：父进程未收，
     但内存/tap 等资源已释放，等同已死）。"""
@@ -365,12 +406,20 @@ class CloudHypervisorExecutor(Executor):
     - worker 层现状：通用路径已能走 dispatch_task/goal_task；
       断点续跑（goal-resume-breakpoint）已由补缺口轨接线：
       worker 通用分支对任何提供该方法的执行器调用它，
-      Local 仍走原路径；网关任务仍被 worker 侧 loud fail
+      Local 仍走原路径；网关任务的 worker 侧守卫（P1-1 起）
-      守卫拦住，不放宽网关规则。
+      已放行 CHV，网关分流改由本执行器按 VM 类 enforcement
+      （feat-chv-hybrid-b，见 gateway_allowed_for_task）：
+      常驻 VM 可接网关、池 VM 不接（ChvFailure
+      gateway_pool_denied），改之前网关不上 CHV 的前置
+      依赖即由此落地。
     - 网关现状：worker 通用路径在 sub2api 网关下对非 Local
       执行器直接 loud fail（与 CloudExecutor 同款），v1 不
       改这条门，所以 CHV + sub2api 组合目前在 worker 层就
-      明确失败，不宣称生产可用。
+      明确失败，不宣称生产可用。——本段是 v1 旧口径，已被
+      feat-chv-hybrid-b 取代：worker 守卫自 P1-1 起已放行
+      CHV，网关分流现由本执行器的 gateway_allowed_for_task
+      按常驻/池分流（常驻可接、池不接）；旧文案留此作
+      演变记录，执行以 gateway_allowed_for_task 为准。
 
     配置（只走构造参数或环境变量，不写文件；参数优先）：
     - CHV_BINARY：cloud-hypervisor 二进制（默认同名 PATH 查找）
@@ -552,6 +601,7 @@ class CloudHypervisorExecutor(Executor):
         self.sleeper = sleeper if sleeper is not None else time.sleep
         self.audit_callback = None
         self.audit_failure_count = 0
+        self._init_hybrid_b_runtime()
         self._vms_lock = threading.Lock()
         if runner is not None:
             self.runner = runner
@@ -566,6 +616,17 @@ class CloudHypervisorExecutor(Executor):
             self.launcher = _default_popen_launcher
         self._vms = {}
 
+    def _init_hybrid_b_runtime(self):
+        """B 轨运行期状态（_init_runtime 的一段，拆出只为
+        满足 50 行函数上限，语义不变）：排队等待计数
+        （_wait_for_capacity 在等待期间 +1，退出时 -1；
+        queue_depth() 供 L1 扩缩读队列深度）与租约降级前
+        的同步钩子（见 _sync_resident_before_demote；
+        None 即默认口径）。"""
+        self._queue_waiters = 0
+        self._queue_lock = threading.Lock()
+        self.resident_sync_hook = None
+
     def _delivered_secrets(self, name: str) -> set:
         """P2-20 唯一读入口：该 VM 已 seed 投递的密钥名集合。"""
         with self._guest_env_lock:
@@ -2227,10 +2288,16 @@ class CloudHypervisorExecutor(Executor):
             tenant_id, extra_env, sync_workspace)
         self.pause_vm(vm_task)
         import time as _time
-        write_persistent_vm(self.state_dir, tenant_id, {
+        # feat-chv-hybrid-b：合并写（tier/last_active_at 等
+        # 不再被字面量覆盖丢掉）；已有常驻态登记时保持其
+        # 状态（预热不降级常驻租约），否则记 legacy paused。
+        _prior = read_persistent_vm(self.state_dir, tenant_id) or {}
+        _state = _prior.get("state") \
+            if _prior.get("state") in RESIDENT_STATES else "paused"
+        self._write_resident_record(tenant_id, {
             "vm": vm, "tenant_id": tenant_id,
-            "state": "paused", "origin": "prewarm",
+            "state": _state, "origin": "prewarm",
-            "gitea_synced": synced, "updated_at": _time.time()})
+            "gitea_synced": synced}, now=_time.time())
         self._emit("warm_pool_prewarm", vm_task,
                    "vm=%s gitea_synced=%s" % (vm, synced))
         return {"pooled": True, "vm": vm, "mode": "prewarmed",
@@ -2259,13 +2326,36 @@ class CloudHypervisorExecutor(Executor):
         (vm, record)；任一不符即清登记、返回 (None, None)
         （调用方回落冷启）。归属不符与 socket 消失是登记
         本身已坏，清掉；状态未知（vm.info 取不到）不在此
-        判——那是认领时的第二道门，见 acquire_for_task。"""
+        判——那是认领时的第二道门，见 acquire_for_task。
+        feat-chv-hybrid-b：可认领态扩展为 legacy paused +
+        常驻态里 VM 已暂停的两种（resident-active /
+        resident-paused，见 RESIDENT_CLAIMABLE_STATES）；
+        常驻态登记新增 tier 校验——tier 不是 Paid/Heavy
+        即不认领（登记本身是坏的/被降档的，留给清理，不
+        静默按旧语义 resume；legacy paused 登记无 tier
+        字段，走旧语义，不因此拒认）。"""
         record = read_persistent_vm(self.state_dir, tenant)
-        if record is None or record.get("state") != "paused":
+        if record is None \
+                or record.get("state") not in RESIDENT_CLAIMABLE_STATES:
             return None, None
         vm = record.get("vm")
         if not vm:
             return None, None
+        if record.get("state") in RESIDENT_ACTIVE_STATES \
+                and not self.resident_tier_eligible(
+                    tenant, record.get("tier")):
+            # 常驻态却无常驻资格：不认领、不 resume；登记留给
+            # 租约计时器/清理处理（不静默清掉——删租约是降级
+            # 流程的职责，不是认领的职责）。
+            self._emit("resident_tier_denied",
+                       {"id": "persist-%s" % tenant,
+                        "project_id": tenant},
+                       "host_id=%s tenant=%s vm=%s tier=%s 常驻态登记"
+                       "无常驻资格，不认领、回落冷启"
+                       % (self._host_id(), tenant, vm,
+                          record.get("tier")),
+                       "tier_denied")
+            return None, None
         owner = read_vm_tenant(self.state_dir, vm)
         if owner is not None and owner != tenant:
             # 盘的归属与请求方不一致：这台不是你的 VM，绝不
@@ -2323,12 +2413,20 @@ class CloudHypervisorExecutor(Executor):
             clear_persistent_vm(self.state_dir, tenant)
             return {"mode": "cold", "vm": None}
         resume_seconds = _time.time() - t0
-        write_persistent_vm(self.state_dir, tenant, {
+        # feat-chv-hybrid-b：合并写——常驻态登记 resume 后仍
-            "vm": vm, "tenant_id": tenant,
+        # 保持常驻态（不写回 legacy running），只续期；legacy
-            "state": "running",
+        # 登记走旧语义（state=running）。tier 等字段不再被
-            "origin": record.get("origin") or "task",
+        # 字面量覆盖丢掉。
-            "gitea_synced": bool(record.get("gitea_synced")),
+        if record.get("state") in RESIDENT_ACTIVE_STATES:
-            "updated_at": _time.time()})
+            self._write_resident_record(
+                tenant, {"last_active_at": _time.time()}, now=_time.time())
+        else:
+            self._write_resident_record(tenant, {
+                "vm": vm, "tenant_id": tenant,
+                "state": "running",
+                "origin": record.get("origin") or "task",
+                "gitea_synced": bool(record.get("gitea_synced"))},
+                now=_time.time())
         self._emit("warm_pool_acquire", task,
                    "vm=%s mode=resumed resume_seconds=%.2f"
                    " warm_hit=%s"
@@ -2358,16 +2456,681 @@ class CloudHypervisorExecutor(Executor):
             if tenant:
                 clear_persistent_vm(self.state_dir, tenant)
             return False
+    def suspend_after_task(self, task_or_name: dict | str,
+                             tenant_id: str | None = None) -> bool:
+        """任务间隙暂停：pause + 写持久登记（state=paused，
+        origin=task）。pause 失败或 VM 已死即清登记并返回
+        False（调用方不得把死 VM 记成可用预热机）。
+        feat-chv-hybrid-b：常驻租户的任务结束只暂停 VM、
+        只续期（last_active_at=now），不改租约状态——
+        resident-active → resident-paused 只由租约计时器
+        在闲置满阈值时转（见 sweep_resident_leases），任务
+        结束不是降级信号；登记写改走 _write_resident_record
+        合并写，tier/last_active_at 等字段不再被字面量覆盖丢掉。"""
+        vm = self.vm_name(task_or_name)
+        tenant = tenant_id or self._tenant_id(
+            task_or_name if isinstance(task_or_name, dict)
+            else {"id": task_or_name})
+        try:
+            self.pause_vm(task_or_name)
+        except Exception as exc:
+            self._emit("vm_suspend_failed",
+                       task_or_name if isinstance(task_or_name, dict)
+                       else {"id": task_or_name},
+                       "vm=%s 暂停失败，清持久登记：%s" % (vm, exc),
+                       "vm_suspend_failed")
+            if tenant:
+                clear_persistent_vm(self.state_dir, tenant)
+            return False
         if tenant:
             import time as _time
+            now = _time.time()
             record = read_persistent_vm(self.state_dir, tenant) or {}
-            write_persistent_vm(self.state_dir, tenant, {
+            if record.get("state") in RESIDENT_STATES:
-                "vm": vm, "tenant_id": tenant,
+                self._write_resident_record(
-                "state": "paused", "origin": "task",
+                    tenant, {"last_active_at": now}, now=now)
-                "gitea_synced": bool(record.get("gitea_synced")),
+            else:
-                "updated_at": _time.time()})
+                self._write_resident_record(tenant, {
+                    "vm": vm, "tenant_id": tenant,
+                    "state": "paused", "origin": "task",
+                    "gitea_synced": bool(record.get("gitea_synced"))},
+                    now=now)
         return True
 
+    # -- 混合模式 B 轨（feat-chv-hybrid-b，DESIGN-HYBRID.md §9.1
+    #    第 5–9 项）-------------------------------------------------
+    # 计时器驱动方式（实施时定，§4.2 二选一）：worker 心跳轮询。
+    # executor 暴露 sweep_resident_leases() 做一轮扫描，worker
+    # 心跳 tick 调一次（接线点见汇报；本轨不改 worker.py，由
+    # 协调员在集成时接一行调用）。不复用 P8 cron 语义：P8 是
+    # 任务级定时调度（每项目每 tick 上限、指数退避重试），租约
+    # 扫描是宿主级状态机轮询、无任务可调度，套 P8 只会把简单
+    # 的扫描复杂化；心跳轮询与 _reconcile_warm_pool 的调用形态
+    # 同款（worker 启动/心跳时调执行器方法）。
+    # 审计：状态转换记 vm_resident_suspended / vm_resident_demoted /
+    # vm_resident_promoted、升级提示记 resident_upgrade_hint，
+    # 全部带 host_id + tenant_id，绝不带密钥值。
+
+    def _hybrid_env_float(self, name, default):
+        """读 B 轨阈值类环境变量（测试用短阈值注入）：
+        非法/缺失/<=0 即回生产默认值，绝不因配错而停表。"""
+        try:
+            value = float(self._env.get(name) or default)
+        except (TypeError, ValueError):
+            return float(default)
+        return value if value > 0 else float(default)
+
+    def _host_id(self):
+        """审计里的 host_id：CHV_HOST_ID 环境变量优先，
+        未设即本机 hostname。"""
+        try:
+            host_id = self._env.get("CHV_HOST_ID")
+        except Exception:
+            host_id = None
+        return host_id or socket.gethostname()
+
+    def _resident_thresholds(self):
+        """租约阈值（秒）：(挂起阈值, 降级阈值)。生产默认值
+        30 分钟 / 24 小时；测试经 CHV_RESIDENT_IDLE_SUSPEND_SECONDS /
+        CHV_RESIDENT_IDLE_DEMOTE_SECONDS 注入短阈值（构造时
+        env 或事后改 self._env 均可，调用时现读）。"""
+        return (
+            self._hybrid_env_float(
+                "CHV_RESIDENT_IDLE_SUSPEND_SECONDS",
+                RESIDENT_IDLE_SUSPEND_SECONDS),
+            self._hybrid_env_float(
+                "CHV_RESIDENT_IDLE_DEMOTE_SECONDS",
+                RESIDENT_IDLE_DEMOTE_SECONDS))
+
+    def resident_tier_eligible(self, tenant_id: str | None,
+                               tier: str | None = None) -> bool:
+        """常驻资格（权益门，fail-closed，设计 §5）：
+        只有 Paid/Heavy 有常驻资格；Free 无论多高频都没有；
+        无 tier 行 = 无常驻资格（拒绝常驻，走池）。
+        不对称说明（有意为之，设计 §5）：P11 的既定语义是
+        “无 tier 行的已有用户不受 tier 门约束（只受原有 USD
+        配额约束）”——那是用量门。常驻是权益门，语义相反：
+        无 tier 行 = 不授予新权益。两者不冲突，因为一个是
+        “不限制已有用量”，一个是“不授予新权益”；实施时不许
+        把 P11 的豁免复用到常驻门上。
+        tier 参数优先，否则读登记的 tier 字段；两者皆无即
+        不合格。"""
+        if tier is None:
+            record = read_persistent_vm(self.state_dir, tenant_id) \
+                if tenant_id else None
+            tier = record.get("tier") if record else None
+        return tier in RESIDENT_TIERS
+
+    def _write_resident_record(self, tenant, updates, now=None):
+        """合并写持久登记：保留旧字段，只更新给定字段 +
+        updated_at；新登记补 tenant_id/vm 默认值。阶段 1 的
+        三处字面量写登记（prewarm/acquire/suspend）至此都
+        经本方法，tier/state/last_active_at/lease_expires_at
+        不再被字面量覆盖丢掉。"""
+        record = read_persistent_vm(self.state_dir, tenant) or {}
+        record.update(updates)
+        record.setdefault("tenant_id", tenant)
+        record.setdefault("vm", self.persistent_vm_name(tenant))
+        record["updated_at"] = time.time() if now is None else now
+        write_persistent_vm(self.state_dir, tenant, record)
+        return record
+
+    def touch_resident_activity(self, tenant_id: str | None,
+                                now: float | None = None) -> bool:
+        """租约续期（设计 §4.1）：每次任务/消息/网关活动即续期
+        （last_active_at=now）。仅当登记存在且为常驻有效态
+        （resident-active/resident-paused/hibernated）时写；
+        无登记不凭空造租约，demoted 不静默复活（复活走
+        promote_to_resident）。lease_expires_at 是闲置规则
+        的派生值（last_active_at + 挂起阈值 + 降级阈值，即
+        再无活动时的降级点），不是日历到期日，续期即重算。
+        返回是否续期。"""
+        if not tenant_id:
+            return False
+        record = read_persistent_vm(self.state_dir, tenant_id)
+        if record is None \
+                or record.get("state") not in RESIDENT_ACTIVE_STATES:
+            return False
+        now = time.time() if now is None else now
+        suspend, demote = self._resident_thresholds()
+        self._write_resident_record(tenant_id, {
+            "last_active_at": now,
+            "lease_expires_at": now + suspend + demote}, now=now)
+        return True
+
+    def _resident_records(self):
+        """扫描 state_dir/persistent/*.json，返回常驻有效态
+        登记 [(tenant, record)]（demoted 已收回租约、不占槽，
+        不在内；损坏/非 dict 的登记跳过）。"""
+        out = []
+        pdir = os.path.join(self.state_dir, "persistent")
+        try:
+            entries = os.listdir(pdir)
+        except OSError:
+            return out
+        for entry in entries:
+            if not entry.endswith(".json"):
+                continue
+            try:
+                with open(os.path.join(pdir, entry), "r") as fh:
+                    record = json.load(fh)
+            except (OSError, ValueError):
+                continue
+            if isinstance(record, dict) \
+                    and record.get("state") in RESIDENT_ACTIVE_STATES:
+                tenant = record.get("tenant_id") \
+                    or entry[:-len(".json")]
+                out.append((tenant, record))
+        return out
+
+    def _resident_count(self):
+        """当前常驻有效租约数（占槽数）。"""
+        return len(self._resident_records())
+
+    def resident_slot_cap(self) -> int:
+        """每宿主常驻上限（已定值，§4.1）：VM 槽位的 50%
+        （CHV_MAX_VMS × 50%，向下取整）。"""
+        return max(0, int(self.max_vms * RESIDENT_MAX_FRACTION))
+
+    def promote_to_resident(self, tenant_id: str,
+                            tier: str | None = None) -> dict:
+        """新建/恢复常驻租约（§4.1/§4.4 执行层语义）：
+        - tier 校验 fail-closed：tier 须显式为 Paid/Heavy；
+          缺失/不符即 ChvFailure tier_denied（平台层路由时
+          已拦一次，这里是执行层 preflight 第二道）。
+        - 常驻上限：已常驻数达上限时不抢占已有常驻，返回
+          {promoted: False, queued: True}（新常驻申请排队）。
+        - 成功：登记 tier/state=resident-active/last_active_at/
+          lease_expires_at，审计 vm_resident_promoted。
+        已是常驻有效态时幂等续期（不重复占槽）。
+        返回 {promoted, queued, vm, tier, state, reason?}。"""
+        if not tenant_id:
+            raise ValueError("tenant_id 不能为空")
+        if tier is None:
+            record = read_persistent_vm(self.state_dir, tenant_id)
+            tier = record.get("tier") if record else None
+        if tier not in RESIDENT_TIERS:
+            raise ChvFailure(
+                "tier_denied",
+                "常驻资格须为 Paid/Heavy 档位：租户 %s tier=%s，"
+                "无常驻资格（Free 无论多高频都没有；无 tier 行"
+                "即无资格，走池）" % (tenant_id, tier))
+        now = time.time()
+        record = read_persistent_vm(self.state_dir, tenant_id)
+        if record is not None \
+                and record.get("state") in RESIDENT_ACTIVE_STATES:
+            self.touch_resident_activity(tenant_id, now=now)
+            return {"promoted": True, "queued": False,
+                    "vm": record.get("vm"), "tier": tier,
+                    "state": record.get("state")}
+        if self._resident_count() >= self.resident_slot_cap():
+            return {"promoted": False, "queued": True,
+                    "reason": "resident_capacity_exceeded",
+                    "tier": tier}
+        suspend, demote = self._resident_thresholds()
+        record = self._write_resident_record(tenant_id, {
+            "tier": tier, "state": "resident-active",
+            "last_active_at": now,
+            "lease_expires_at": now + suspend + demote}, now=now)
+        self._emit("vm_resident_promoted",
+                   {"id": "persist-%s" % tenant_id,
+                    "project_id": tenant_id},
+                   "host_id=%s tenant=%s vm=%s tier=%s"
+                   % (self._host_id(), tenant_id,
+                      record.get("vm"), tier))
+        return {"promoted": True, "queued": False,
+                "vm": record.get("vm"), "tier": tier,
+                "state": "resident-active"}
+
+    def _suspend_resident_vm(self, tenant, record):
+        """租约挂起的 VM 动作（sweep 的一步）：VM 已是 Paused
+        （任务间隙 suspend_after_task 已暂停）即直接转态，
+        不调 pause；Running 即 pause_vm；状态未知且 socket
+        已不在时视为 VM 已不在、无须 pause，直接转态；其余
+        未知时试调 pause_vm，失败即本租户中止（返回 False，
+        租约状态不动）。"""
+        vm = record.get("vm") or self.persistent_vm_name(tenant)
+        state = self._vm_info_state({"id": vm})
+        if state == "Paused":
+            return True
+        if state is None and not os.path.exists(
+                self.api_socket({"id": vm})):
+            return True
+        try:
+            self.pause_vm({"id": vm})
+            return True
+        except Exception as exc:
+            self._emit("vm_suspend_failed",
+                       {"id": vm, "project_id": tenant},
+                       "vm=%s 租约挂起 pause 失败：%s" % (vm, exc),
+                       "vm_suspend_failed")
+            return False
+
+    def sweep_resident_leases(self, now: float | None = None) -> dict:
+        """租约计时器一轮扫描（设计 §4.2，worker 心跳轮询驱动）：
+        - resident-active 且闲置 >= 挂起阈值 → pause_vm（已
+          Paused 的直接转态）→ state=resident-paused，审计
+          vm_resident_suspended（带 host_id + tenant_id）。
+        - resident-paused 且闲置 >= 挂起阈值 + 降级阈值（即
+          累计 24.5 小时无活动）→ demote_resident，审计
+          vm_resident_demoted。
+        单个租户失败（pause 失败/降级中止）不中断整轮，记入
+        aborted。返回 {suspended, demoted, aborted}（租户 id
+        列表）。"""
+        now = time.time() if now is None else now
+        suspend_s, demote_s = self._resident_thresholds()
+        out = {"suspended": [], "demoted": [], "aborted": []}
+        for tenant, record in self._resident_records():
+            try:
+                last = record.get("last_active_at") \
+                    or record.get("updated_at") or now
+                idle = now - float(last)
+                state = record.get("state")
+                if state == "resident-active" and idle >= suspend_s:
+                    if self._suspend_resident_vm(tenant, record):
+                        self._write_resident_record(
+                            tenant, {"state": "resident-paused"},
+                            now=now)
+                        self._emit(
+                            "vm_resident_suspended",
+                            {"id": "persist-%s" % tenant,
+                             "project_id": tenant},
+                            "host_id=%s tenant=%s vm=%s idle_seconds=%.0f"
+                            % (self._host_id(), tenant,
+                               record.get("vm"), idle))
+                        out["suspended"].append(tenant)
+                    else:
+                        out["aborted"].append(tenant)
+                elif state == "resident-paused" \
+                        and idle >= suspend_s + demote_s:
+                    result = self.demote_resident(tenant, now=now)
+                    if result.get("demoted"):
+                        out["demoted"].append(tenant)
+                    else:
+                        out["aborted"].append(tenant)
+            except Exception as exc:
+                _LOGGER.error(
+                    "chv_resident_sweep_failed tenant=%s error=%s",
+                    tenant, exc)
+                out["aborted"].append(tenant)
+        return out
+
+    def _sync_resident_before_demote(self, tenant_id, vm):
+        """降级前的同步步（设计 §4.3 第 1 步）：
+        权威同步点在任务结束时（§3.3：任务完成即 Gitea push +
+        记忆/技能落盘，之后 VM 才暂停），降级时盘上已是同步后
+        状态；因此默认本步直接通过。若部署配了
+        resident_sync_hook（例如先 resume 再跑一次 chv-sync
+        push 的回调），则以钩子返回值为准：False 即同步失败，
+        降级中止。demote_resident 的 sync_fn 参数优先于钩子。"""
+        hook = getattr(self, "resident_sync_hook", None)
+        if hook is None:
+            return True
+        return bool(hook(tenant_id, vm))
+
+    # 降级流程细则（设计 §4.3，注释放方法外、不计入 50 行上限）：
+    # 同步 → 快照 → 清 seed → 卸盘 → 销毁 → 登记 demoted。
+    # - 同步失败即降级中止、VM 保留、记审计
+    #   （vm_resident_demote_aborted），绝不丢改动后降级；
+    #   快照失败同样中止。
+    # - 清 seed：stop_vm 删 seed 目录 + 宿主私钥 + 已投递密钥
+    #   记录；seed.iso 随 keep_disk=True 保留——它是快照
+    #   config.json 引用的 restore 输入（hibernate 的 V-7/D2
+    #   同款），删了快照即不可恢复；“清 seed”指清 live seed，
+    #   不指毁快照输入。
+    # - 卸盘/销毁：stop_vm(keep_disk=True) 关 VM；任务盘与
+    #   seed.iso 留作快照输入，数据盘 data.raw 是用户数据、
+    #   永不随降级删（只走显式回收）。
+    # - 登记：state=demoted，保留 last_active_at 与快照 URL
+    #   （供下次唤醒时从 Gitea 重建或快照恢复），审计
+    #   vm_resident_demoted（带 host_id + tenant_id）。
+    # - VM 已不在时跳过快照（无物可拍），只做登记。
+    def demote_resident(self, tenant_id: str | None,
+                        snapshot_url: str | None = None,
+                        sync_fn: Callable[..., bool] | None = None,
+                        now: float | None = None) -> dict:
+        """降级流程（设计 §4.3）：同步 → 快照 → 清 seed →
+        卸盘 → 销毁 → 登记 demoted。同步/快照失败即中止、VM
+        保留；仅 resident-paused/hibernated 可降级，其余返回
+        {demoted: False, reason: not_demotable}。
+        返回 {demoted, vm, snapshot, reason?}。"""
+        now = time.time() if now is None else now
+        record = read_persistent_vm(self.state_dir, tenant_id)
+        if record is None or record.get("state") not in (
+                "resident-paused", "hibernated"):
+            return {"demoted": False, "reason": "not_demotable",
+                    "state": record.get("state") if record else None}
+        vm = record.get("vm") or self.persistent_vm_name(tenant_id)
+        task = {"id": vm, "project_id": tenant_id}
+        if not self._demote_sync_ok(task, tenant_id, vm, sync_fn):
+            return {"demoted": False, "reason": "sync_failed",
+                    "vm": vm}
+        snapshot, failed = self._demote_snapshot(
+            task, vm, snapshot_url)
+        if failed:
+            return {"demoted": False, "reason": "snapshot_failed",
+                    "vm": vm}
+        try:
+            self.stop_vm(task, keep_disk=True)
+        except Exception:
+            pass
+        updates = {"state": "demoted"}
+        if snapshot:
+            updates["snapshot_url"] = snapshot
+        self._write_resident_record(tenant_id, updates, now=now)
+        self._emit("vm_resident_demoted", task,
+                   "host_id=%s tenant=%s vm=%s snapshot=%s"
+                   % (self._host_id(), tenant_id, vm, snapshot))
+        return {"demoted": True, "vm": vm, "snapshot": snapshot}
+
+    def _demote_sync_ok(self, task, tenant_id, vm, sync_fn):
+        """降级同步步（demote_resident 的一步）：失败记
+        vm_resident_demote_aborted 审计并返回 False（VM 保留，
+        绝不丢改动后降级）；成功返回 True。"""
+        sync_callable = sync_fn if sync_fn is not None \
+            else self._sync_resident_before_demote
+        try:
+            synced = bool(sync_callable(tenant_id, vm))
+        except Exception:
+            synced = False
+        if not synced:
+            self._emit("vm_resident_demote_aborted", task,
+                       "host_id=%s tenant=%s vm=%s 同步失败，降级中止、"
+                       "VM 保留" % (self._host_id(), tenant_id, vm),
+                       "resident_demote_sync_failed")
+        return synced
+
+    def _demote_snapshot(self, task, vm, snapshot_url):
+        """降级快照步（demote_resident 的一步）：VM 活着即
+        snapshot_vm（失败记中止审计）；VM 已不在时跳过快照
+        （无物可拍）。返回 (snapshot_url_or_None, failed)。"""
+        if not (os.path.exists(self.api_socket({"id": vm}))
+                or self._vm_info_state({"id": vm}) is not None):
+            return None, False
+        if snapshot_url is None:
+            snapshot_url = "file://%s" % os.path.join(
+                self.state_dir, vm, "snapshot-demote")
+        try:
+            self.snapshot_vm(task, snapshot_url)
+        except Exception as exc:
+            self._emit("vm_resident_demote_aborted", task,
+                       "host_id=%s tenant=%s vm=%s 快照失败，降级"
+                       "中止、VM 保留：%s"
+                       % (self._host_id(), task.get("project_id"),
+                          vm, exc),
+                       "resident_demote_snapshot_failed")
+            return None, True
+        return snapshot_url, False
+
+    # -- 排队（设计 §3.4）--------------------------------------------
+    def queue_depth(self) -> int:
+        """当前排队等待的任务数（L1 扩缩输入；等待期间计数，
+        退出时减回）。"""
+        with self._queue_lock:
+            return self._queue_waiters
+
+    def _queue_attempts(self):
+        """排队轮次（次数制，与 wait_api_ready 同款口径：fake
+        sleeper 不推进墙钟时墙钟循环会空转，次数循环在单测里
+        是确定性的；真实 sleeper 下总时长约等于超时）。
+        超时经 CHV_QUEUE_TIMEOUT_SECONDS 注入（默认 300 秒），
+        轮询间隔经 CHV_QUEUE_POLL_SECONDS（默认 5 秒）。"""
+        timeout = self._hybrid_env_float(
+            "CHV_QUEUE_TIMEOUT_SECONDS", QUEUE_TIMEOUT_SECONDS)
+        poll = self._hybrid_env_float(
+            "CHV_QUEUE_POLL_SECONDS", QUEUE_POLL_SECONDS)
+        return max(1, int(timeout / poll)), poll
+
+    def _wait_for_capacity(self, task, exclude=None):
+        """排队等宿主容量（设计 §3.4）：容量有空即直接过；
+        满时按到达顺序等（本方法是调用方线程内顺序等待，无
+        优先队列、不插队）、不抢占（只等同一容量门放行，不踢
+        别的 VM）、不升/降档位（只等槽位，不改路由）。
+        超时（默认 300 秒）后 ChvFailure host_capacity_exceeded
+        明确拒绝，不无限积压。等待期间 queue_depth() +1。"""
+        try:
+            self._check_capacity(exclude=exclude)
+            return
+        except ChvFailure:
+            pass
+        attempts, poll = self._queue_attempts()
+        with self._queue_lock:
+            self._queue_waiters += 1
+        try:
+            for _ in range(attempts):
+                self.sleeper(poll)
+                try:
+                    self._check_capacity(exclude=exclude)
+                    return
+                except ChvFailure:
+                    continue
+            raise ChvFailure(
+                "host_capacity_exceeded",
+                "排队超时：%d 秒内无可用宿主（容量仍满），明确拒绝，"
+                "不无限积压" % int(attempts * poll))
+        finally:
+            with self._queue_lock:
+                self._queue_waiters -= 1
+
+    def _wait_for_resident_slot(self, tenant_id):
+        """常驻租户的任务排队时仍等常驻槽位（设计 §3.4）：
+        只等常驻计数 < 上限，不挪去池里跑（挪了等于改隔离
+        与计费口径），不抢占已有常驻。超时同样
+        host_capacity_exceeded。"""
+        if self._resident_count() < self.resident_slot_cap():
+            return
+        attempts, poll = self._queue_attempts()
+        with self._queue_lock:
+            self._queue_waiters += 1
+        try:
+            for _ in range(attempts):
+                self.sleeper(poll)
+                if self._resident_count() < self.resident_slot_cap():
+                    return
+            raise ChvFailure(
+                "host_capacity_exceeded",
+                "排队超时：常驻槽位仍满（%d/%d），明确拒绝"
+                % (self._resident_count(),
+                   self.resident_slot_cap()))
+        finally:
+            with self._queue_lock:
+                self._queue_waiters -= 1
+
+    # -- 升级提示点（设计 §5）----------------------------------------
+    def _pool_task_file(self, tenant_id):
+        """池任务时间戳文件：state_dir/pool-tasks/<tenant>.json
+        （执行层任务登记，不是平台层账本；只存时间戳与上次
+        提示时间，不存任务内容、不存密钥）。"""
+        safe = re.sub(r"[^A-Za-z0-9_.-]", "-",
+                      str(tenant_id or "default"))
+        return os.path.join(self.state_dir, "pool-tasks",
+                            safe + ".json")
+
+    def _load_pool_tasks(self, tenant_id):
+        """读池任务登记：{tasks: [ts...], last_hint_at: ts|None}；
+        无文件/损坏即空登记（不猜）。"""
+        try:
+            with open(self._pool_task_file(tenant_id), "r") as fh:
+                data = json.load(fh)
+        except (OSError, ValueError):
+            return {"tasks": [], "last_hint_at": None}
+        if not isinstance(data, dict):
+            return {"tasks": [], "last_hint_at": None}
+        tasks = data.get("tasks") or []
+        if not isinstance(tasks, list):
+            tasks = []
+        return {"tasks": [t for t in tasks
+                          if isinstance(t, (int, float))],
+                "last_hint_at": data.get("last_hint_at")}
+
+    def _save_pool_tasks(self, tenant_id, data):
+        """写池任务登记（atomic 写，与持久登记同款）。"""
+        path = self._pool_task_file(tenant_id)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        atomic_write_text(path, json.dumps(data))
+
+    def _has_7day_streak(self, tasks, now):
+        """连续 7 天每天 ≥3 次（UTC 日桶；含今天往前 7 天）。"""
+        today = int(now // 86400)
+        days = {}
+        for ts in tasks:
+            day = int(ts // 86400)
+            days[day] = days.get(day, 0) + 1
+        for offset in range(UPGRADE_HINT_STREAK_DAYS):
+            if days.get(today - offset, 0) < UPGRADE_HINT_MIN_PER_DAY:
+                return False
+        return True
+
+    def _upgrade_hint_metric(self, tasks, now):
+        """本次提示的频率度量（进审计 detail，供平台层消费）。"""
+        if len(tasks) >= UPGRADE_HINT_MIN_TASKS:
+            return "30d_tasks=%d" % len(tasks)
+        return "7d_streak_%d_per_day" % UPGRADE_HINT_MIN_PER_DAY
+
+    def _upgrade_hint_due(self, tasks, last_hint_at, now):
+        """是否该记升级提示：先去重（一周内同一租户最多一次），
+        再判高频定义（30 天 ≥100 次，或连续 7 天每天 ≥3 次）。"""
+        if last_hint_at is not None \
+                and now - float(last_hint_at) \
+                < UPGRADE_HINT_DEDUP_DAYS * 86400:
+            return False
+        if len(tasks) >= UPGRADE_HINT_MIN_TASKS:
+            return True
+        return self._has_7day_streak(tasks, now)
+
+    def record_pool_task(self, tenant_id: str | None,
+                         tier: str | None = None,
+                         now: float | None = None) -> dict:
+        """记一次池任务并在达高频定义时发升级提示点（设计 §5）：
+        只对 Free 租户记（Paid/Heavy 已有常驻资格，无须提示；
+        tier 未知时按 Free 记——提示只是事件，平台层消费时
+        再核档位）；频率只用于产生提示点，绝不用于改路由
+        （本方法不返回路由决策、不改任何登记的状态字段）。
+        频率度量从本文件里的任务时间戳算，不另建账本。
+        返回 {recorded, hinted, metric}。"""
+        now = time.time() if now is None else now
+        if tier is None:
+            record = read_persistent_vm(self.state_dir, tenant_id) \
+                if tenant_id else None
+            tier = record.get("tier") if record else None
+        if tier is not None and tier != "Free":
+            return {"recorded": False, "hinted": False,
+                    "reason": "not_free_tier"}
+        data = self._load_pool_tasks(tenant_id)
+        cutoff = now - UPGRADE_HINT_WINDOW_DAYS * 86400
+        tasks = [t for t in data["tasks"] if t >= cutoff]
+        tasks.append(now)
+        hinted = False
+        metric = None
+        if self._upgrade_hint_due(tasks, data.get("last_hint_at"),
+                                  now):
+            metric = self._upgrade_hint_metric(tasks, now)
+            data["last_hint_at"] = now
+            self._emit(
+                "resident_upgrade_hint",
+                {"id": "pool-%s" % tenant_id,
+                 "project_id": tenant_id},
+                "host_id=%s tenant=%s tier=Free metric=%s "
+                "window_days=%d"
+                % (self._host_id(), tenant_id, metric,
+                   UPGRADE_HINT_WINDOW_DAYS))
+            hinted = True
+        data["tasks"] = tasks
+        self._save_pool_tasks(tenant_id, data)
+        return {"recorded": True, "hinted": hinted, "metric": metric}
+
+    # -- 网关分流（设计 §6）------------------------------------------
+    def gateway_allowed_for_task(self, task: dict,
+                                 model_provider: str | None = None) -> bool:
+        """网关分流（设计 §6；worker 侧守卫自 P1-1 起已放行 CHV，
+        分流 enforcement 落在本执行器——改之前网关不上 CHV 的
+        前置依赖即由此落地）：
+        - 非网关任务（model_provider 不是网关 provider）→
+          True（不拦；本方法不作常驻判定用）。
+        - 网关任务：常驻 VM 可接（登记为常驻有效态且 tier
+          校验通过）、池 VM 不接。常驻有效态含 hibernated
+          （可 restore + resume 续跑）；demoted、无登记、
+          legacy 态一律按池算。"""
+        provider = model_provider
+        if provider is None and isinstance(task, dict):
+            provider = task.get("_gateway_provider")
+        if provider not in GATEWAY_PROVIDERS:
+            return True
+        tenant = self._tenant_id(task) if isinstance(task, dict) \
+            else None
+        if not tenant:
+            return False
+        record = read_persistent_vm(self.state_dir, tenant)
+        if record is None \
+                or record.get("state") not in RESIDENT_ACTIVE_STATES:
+            return False
+        return self.resident_tier_eligible(tenant,
+                                           record.get("tier"))
+
+    def _check_gateway_vm_class(self, task, model_provider=None):
+        """网关任务的 VM 类门：池 VM 以 gateway_pool_denied
+        明确拒绝（ChvFailure，worker 现有 except 路径照常
+        写回 failed）；常驻 VM 放行。"""
+        if not self.gateway_allowed_for_task(task, model_provider):
+            tenant = self._tenant_id(task) \
+                if isinstance(task, dict) else None
+            raise ChvFailure(
+                "gateway_pool_denied",
+                "网关任务只接常驻 VM：租户 %s 当前走池（无常驻"
+                "租约或档位无常驻资格），拒绝上 CHV；池 VM 用完"
+                "即销毁，长连接在上面活不过一个任务周期"
+                % tenant)
+
+    def _hybrid_pool_preamble(self, task, model_provider=None):
+        """B 轨池路径前置（_run_with_vm 的开头，拆出只为满足
+        50 行函数上限）：任务到达即活动（续期租约）、网关 VM
+        类门（常驻可接、池不接）、排队等容量（超时
+        host_capacity_exceeded）、池任务记一次（Free 高频
+        提示点原料）。续期与记数失败不挡任务；网关门与排队
+        拒绝直接抛。"""
+        try:
+            _tenant = self._tenant_id(task)
+            if _tenant:
+                self.touch_resident_activity(_tenant)
+        except Exception:
+            pass
+        self._check_gateway_vm_class(task, model_provider)
+        self._wait_for_capacity(task, exclude=self.vm_name(task))
+        try:
+            _tenant = self._tenant_id(task)
+            if _tenant:
+                self.record_pool_task(
+                    _tenant,
+                    task.get("tier") if isinstance(task, dict) else None)
+        except Exception:
+            pass
+
+    def _hybrid_persistent_preamble(self, task, model_provider=None):
+        """B 轨持久路径前置（_run_with_persistent_vm 的开头，
+        拆出只为满足 50 行函数上限）：任务到达即续期租约
+        （仅常驻登记写，失败不挡任务）；网关任务先过 VM 类门
+        （常驻可接、池不接）。"""
+        try:
+            _tenant = self._tenant_id(task)
+            if _tenant:
+                self.touch_resident_activity(_tenant)
+        except Exception:
+            pass
+        self._check_gateway_vm_class(task, model_provider)
+
+    def _wait_for_persistent_slot(self, task, tenant, vm):
+        """持久路径冷启需新占槽位时先排队（_run_with_persistent_vm
+        的一段，拆出只为满足 50 行函数上限）：常驻租户等常驻
+        槽位（不挪去池里跑），其余等通用宿主容量；超时
+        host_capacity_exceeded 明确拒绝。"""
+        _record = read_persistent_vm(self.state_dir, tenant) \
+            if tenant else None
+        if _record is not None and _record.get("state") \
+                in RESIDENT_ACTIVE_STATES:
+            self._wait_for_resident_slot(tenant)
+        else:
+            self._wait_for_capacity(task, exclude=vm)
+
     def _run_guest_on_vm(self, task, vm, inner_args, goal,
                          extra_env=None, **kwargs):
         """在指定 VM 上跑客内命令并做中途死亡分类（
@@ -2412,11 +3175,14 @@ class CloudHypervisorExecutor(Executor):
         _run_guest_on_vm（登记进程已死即 interrupted）。"""
         tenant = self._tenant_id(task)
         vm_task = self._persistent_task(tenant)
+        self._hybrid_persistent_preamble(
+            task, kwargs.get("model_provider"))
         acquired = self.acquire_for_task(task, extra_env=extra_env)
         vm = acquired.get("vm") or self.vm_name(vm_task)
         mode = acquired["mode"]
         resume_seconds = acquired.get("resume_seconds")
         if mode == "cold":
+            self._wait_for_persistent_slot(vm_task, tenant, vm)
             self.wake_vm(vm_task, extra_env=extra_env)
         try:
             result = self._run_guest_on_vm(
@@ -3372,35 +4138,48 @@ class CloudHypervisorExecutor(Executor):
         parse_goal_result），这里不再二次改写 goal 结论。
         feat-chv-secfix（H2）：extra_env 透传给 wake_vm（密钥
         子集在 provisioned 模式下随 seed 投递），再透传给
-        _run_guest（远程命令 source 客内密钥文件）。"""
+        _run_guest（远程命令 source 客内密钥文件）。
+        feat-chv-hybrid-b：wake 之前先走 _hybrid_pool_preamble
+        （续期 + 网关门 + 排队 + 池任务记数）。"""
+        self._hybrid_pool_preamble(task, kwargs.get("model_provider"))
         self.wake_vm(task, extra_env=extra_env)
         try:
-            try:
+            return self._run_guest_with_death_classification(
-                if goal:
+                task, inner_args, goal, extra_env, kwargs)
-                    return self._run_guest_goal(
-                        task, inner_args, extra_env=extra_env, **kwargs)
-                result = self._run_guest(task, inner_args,
-                                         extra_env=extra_env, **kwargs)
-            except Exception as exc:
-                if self._vm_proc_dead(task):
-                    return self._vm_death_result(
-                        task, goal, str(exc)[:500])
-                raise
-            if not goal and result.get("status") == "failed" \
-                    and self._vm_proc_dead(task):
-                result = dict(result)
-                result["status"] = "interrupted"
-                result["vm_dead"] = True
-                result["interrupted_reason"] = (
-                    "VM 中途死亡，任务未完成、断点可续跑："
-                    "客内命令已 failed 且登记的 VMM 进程已死")
-            return result
         finally:
             try:
                 self.shutdown_vm(task)
             except Exception:
                 pass
 
+    def _run_guest_with_death_classification(self, task, inner_args,
+                                             goal, extra_env, kwargs):
+        """客内执行 + 中途死亡分类（_run_with_vm 的一段，拆出
+        只为满足 50 行函数上限）：客内命令抛传输异常、或非
+        goal 结果已 failed 时，若登记的 VMM 进程已死，判
+        interrupted（断点可续跑）；VM 还活着时的异常/失败
+        原样处理，不误判。"""
+        try:
+            if goal:
+                return self._run_guest_goal(
+                    task, inner_args, extra_env=extra_env, **kwargs)
+            result = self._run_guest(task, inner_args,
+                                     extra_env=extra_env, **kwargs)
+        except Exception as exc:
+            if self._vm_proc_dead(task):
+                return self._vm_death_result(
+                    task, goal, str(exc)[:500])
+            raise
+        if not goal and result.get("status") == "failed" \
+                and self._vm_proc_dead(task):
+            result = dict(result)
+            result["status"] = "interrupted"
+            result["vm_dead"] = True
+            result["interrupted_reason"] = (
+                "VM 中途死亡，任务未完成、断点可续跑："
+                "客内命令已 failed 且登记的 VMM 进程已死")
+        return result
+
     def execute(self, task: dict) -> dict:
         """只读执行：客内 host.py status，不直接跑模型回合
         （与 Local/Docker 的 execute 同款语义）。
@@ -3527,8 +4306,10 @@ class CloudHypervisorExecutor(Executor):
         worker 层已接线（补缺口轨）：非网关的 CHV 断点任务
         经 worker 的通用断点分支调到这里，model_provider
         按名透传给客内 host.py（与 Local/Docker 同款口径，
-        首轮 thread 的 provider 必须一致）；网关任务在 worker
+        首轮 thread 的 provider 必须一致）；网关任务的分流
-        侧仍被 loud fail 守卫拦住，不到这里。模型 key
+        由 _run_with_vm 开头的 _check_gateway_vm_class 按 VM
+        类 enforcement（常驻 VM 可接、池 VM 以
+        gateway_pool_denied 拒绝）。模型 key
         本身绝不经本方法传值，只按环境变量名注入。"""
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")

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