feat(chv-hybrid-c): autoscaler 池目标数 + 回收同步门 + 审计事件目录

- autoscaler: desired_workers += pool_target（W=2 台/宿主，
  pool_target_for_hosts）；Autoscaler/CLI 支持 --pool-target /
  --online-hosts；L1 其余语义（30s 轮、scale up fast/down slow、
  fail-closed、veto、LIFO）原样复用，L2 不自动加主机
- executor: recycle_vm/recycle_persistent_vm 前加 Gitea 同步门
  （chv-sync push 失败即 state/status=sync_failed、VM 不回收、
  记 vm_recycle_sync_failed 审计）；未配同步/VM 已死走旧路径
- 新增 cloud/chv/hybrid_events.py：9 个设计事件名一字不差，
  全部必填 host_id、密钥字段名拒绝入目录
- orphans facade 已存在且复用 registry/orphan 语义，不动
  experimental_pool（wired_to_dispatch 恒 False）
- 测试：cloud/tests/test_chv_hybrid_c.py 32 项；全量 4611 项
  0 失败 0 错误 15 skipped

cloud/autoscaler.py

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

cloud/chv/executor.py

@@ -2469,16 +2469,33 @@ class CloudHypervisorExecutor(Executor):
             sandbox=("read-only" if task.get("plan") else None))
 
     def recycle_persistent_vm(self, task_or_name: dict | str,
-                              tenant_id: str | None = None) -> dict:
+                              tenant_id: str | None = None,
+                              extra_env: dict | None = None,
+                              sync: bool = True) -> dict:
         """显式回收持久 VM（唯一删用户盘的入口）：先 shutdown
        （stop_vm，keep_disk=False），再删 overlay 与数据盘、
         清 tenant 标记与持久登记。返回 {vm, state: recycled,
         deleted_disks}。与 recycle_vm 的区别：后者只关 VM
-        不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。"""
+        不删持久盘（持久盘须显式回收才删，用户数据不静默丢）。
+
+        C 轨新增（§3.3 同款回收同步门，见 recycle_vm）：sync
+        默认开；配了 Gitea 同步且 push 失败时不 shutdown、
+        不删盘，返回 sync_failed（vm_retained=True，
+        deleted_disks=[]），等人工/重试。
+        """
         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})
+        if sync:
+            task_dict = task_or_name if isinstance(
+                task_or_name, dict) else {"id": task_or_name}
+            ok, sync_detail = self._recycle_sync_gate(
+                task_dict, extra_env=extra_env)
+            if not ok:
+                return self._sync_failed_result(vm, sync_detail, [])
+        else:
+            sync_detail = "sync skipped by caller"
         self.shutdown_vm(task_or_name)
         deleted = []
         if delete_task_disk(self.state_dir, vm):
@@ -2496,7 +2513,10 @@ class CloudHypervisorExecutor(Executor):
                    else {"id": task_or_name},
                    "vm=%s deleted_disks=%s" % (vm, deleted))
         return {"vm": vm, "state": "recycled",
-                "deleted_disks": deleted}
+                "deleted_disks": deleted,
+                "sync_ok": True,
+                "sync_detail": sync_detail,
+                "vm_retained": False}
 
     def snapshot_vm(self, task_or_name: dict | str, snapshot_url: str) -> str:
         """给（已暂停的）VM 做快照：ch-remote snapshot <裸 URL>。
@@ -3152,7 +3172,108 @@ class CloudHypervisorExecutor(Executor):
         只改数据库状态，必须真关 VM。"""
         return self.shutdown_vm(task_or_name)
 
-    def recycle_vm(self, task: dict) -> dict:
+    def sync_guest_to_gitea(self, task: dict,
+                            extra_env: dict | None = None) -> tuple[bool, str]:
+        """回收前把客内已提交的工作区改动 push 回 Gitea
+        （DESIGN-HYBRID.md §3.3 回收流程第 1 步）。
+
+        复用客内 chv-sync 的 push 腿（chv-guest/chv-sync.sh）：
+        经 SSH 跑 `chv-sync push`；非 fast-forward 即失败，
+        绝不 --force、不自动 rebase（与 chv-sync.sh 的
+        fail-closed 语义同款）。返回 (ok, detail)：
+        - 未配 Gitea 同步（extract_gitea_sync_env 为空）→
+          (True, "no gitea sync configured")，无物可推，
+          不拦回收；
+        - VM 已不运行 → (True, "vm not running")：死掉的
+          客机没有可 push 的活工作树，拦回收也推不出去，
+          不拿已死的 VM 卡清理（其盘仍在 state_dir 里，
+          由显式回收/GC 按原口径处理）；
+        - push rc==0 → (True, ...)；rc!=0 或 SSH 本身失败
+          → (False, 末行输出截断 200 字符）。chv-sync 的
+          remote URL 是干净 https、token 只走 credential
+          store 文件，所以输出里没有 token（与 syncd 的
+          _run_sync 同款截断口径）。
+        边界（诚实写明）：push 只推已提交的提交，未提交的
+        工作树改动不参与同步——与客内 syncd 常驻循环同款
+        口径，本方法不自动 commit（自动 commit 是新语义，
+        设计未定）；同步失败的判定只看 push 本身。
+        本方法不重试：push 是幂等的，失败后调用方
+        （recycle_vm / 人工）重新调回收即重试。
+        """
+        # 方法内 import：与 wake_vm 同款，避免顶层循环导入。
+        from cloud.chv.gitea_sync import extract_gitea_sync_env
+        sync = extract_gitea_sync_env(
+            task, extra_env,
+            config_sync=getattr(self, "gitea_sync_env", None))
+        if not sync:
+            return True, "no gitea sync configured"
+        if not self._vm_is_running(task):
+            return True, "vm not running: nothing to sync"
+        try:
+            ssh_cmd = self.build_ssh_command(task, ["chv-sync", "push"])
+            result = self._call_runner(ssh_cmd,
+                                       timeout=self.task_timeout)
+        except Exception as exc:
+            return False, "ssh failed: %s" % str(exc)[:200]
+        if getattr(result, "returncode", 1) == 0:
+            return True, "chv-sync push ok"
+        tail = ((getattr(result, "stdout", "") or "")
+                + (getattr(result, "stderr", "") or "")).strip()
+        tail = tail.splitlines()[-1] if tail else ""
+        return False, "chv-sync push failed: %s" % tail[:200]
+
+    def _emit_recycle_sync_failed(self, task, vm, reason):
+        """回收同步失败的审计（§3.3“同步失败记审计”）：
+        事件 vm_recycle_sync_failed（见 hybrid_events），
+        带 host_id + vm + tenant_id（有则带），绝不带密钥值。
+        审计本身 best-effort：格式化失败时退化成裸 vm 明细，
+        绝不因此掩盖 sync_failed 的返回。"""
+        task_dict = task if isinstance(task, dict) else {"id": task}
+        try:
+            from cloud.chv.hybrid_events import (
+                VM_RECYCLE_SYNC_FAILED, format_detail)
+            fields = {"vm": vm, "reason": reason}
+            tenant = self._tenant_id(task_dict)
+            if tenant:
+                fields["tenant_id"] = tenant
+            detail = format_detail(VM_RECYCLE_SYNC_FAILED, fields)
+        except Exception:
+            detail = "vm=%s reason=%s" % (vm, str(reason)[:200])
+        try:
+            self._emit(VM_RECYCLE_SYNC_FAILED, task_dict, detail,
+                       kind="sync_failed")
+        except Exception:
+            pass
+
+    @staticmethod
+    def _sync_failed_result(vm, detail, deleted_disks=None):
+        """同步失败的回收返回（§3.3）：state/status 均为
+        sync_failed，VM 保留。recycle_vm 与
+        recycle_persistent_vm 共用，免得两处拼同一份 dict；
+        各自再补自己的字段（preserved/lost 或 deleted_disks）。"""
+        out = {"vm": vm, "state": "sync_failed",
+               "status": "sync_failed", "sync_ok": False,
+               "sync_error": detail, "vm_retained": True}
+        if deleted_disks is not None:
+            out["deleted_disks"] = deleted_disks
+        return out
+
+    def _recycle_sync_gate(self, task, extra_env=None):
+        """回收同步门（§3.3 新增语义的共用判定）：
+        同步失败 → (False, detail)；可回收 → (True, detail)。
+        调用方在 False 时必须不 shutdown、不删盘、不发
+        recycled 审计，只记 sync_failed 审计并返回
+        state/status 均为 sync_failed 的结果。"""
+        ok, detail = self.sync_guest_to_gitea(task,
+                                              extra_env=extra_env)
+        if ok:
+            return True, detail
+        self._emit_recycle_sync_failed(task, self.vm_name(task),
+                                       detail)
+        return False, detail
+
+    def recycle_vm(self, task: dict, extra_env: dict | None = None,
+                   sync: bool = True) -> dict:
         """回收 VM（任务结束后的销毁语义）：
         先 shutdown，再返回保留/丢失清单。保留的只有宿主侧
         的 task_checkpoints / task_attempts（数据库里的断点与
@@ -3160,9 +3281,30 @@ class CloudHypervisorExecutor(Executor):
         丢失的是客机内的临时文件与未拷出的 worktree 临时物
         ——recycle 绝不声称“恢复临时文件”。需要拷出的工件
         必须在 shutdown 之前由调用方经 SSH 拷出，本方法不
-        假装替调用方拷了。"""
+        假装替调用方拷了。
+
+        C 轨新增（DESIGN-HYBRID.md §3.3）：sync=True（默认）
+        时先过回收同步门——配了 Gitea 同步且 `chv-sync push`
+        失败时，任务标同步失败（返回 state/status 均为
+        sync_failed，sync_ok=False，vm_retained=True），VM
+        不回收、不复用，等人工/重试，绝不丢改动后静默回收；
+        审计记 vm_recycle_sync_failed。未配同步、VM 已死、
+        或 sync=False（调用方显式跳过）时走旧回收路径。
+        """
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
+        if sync:
+            ok, sync_detail = self._recycle_sync_gate(
+                task, extra_env=extra_env)
+            if not ok:
+                out = self._sync_failed_result(self.vm_name(task),
+                                               sync_detail)
+                out["preserved"] = ["task_checkpoints",
+                                    "task_attempts"]
+                out["lost"] = []
+                return out
+        else:
+            sync_detail = "sync skipped by caller"
         self.shutdown_vm(task)
         self._emit("vm_recycled", task,
                    "vm=%s" % self.vm_name(task))
@@ -3171,6 +3313,9 @@ class CloudHypervisorExecutor(Executor):
             "state": "recycled",
             "preserved": ["task_checkpoints", "task_attempts"],
             "lost": ["guest /tmp", "guest worktree temp files"],
+            "sync_ok": True,
+            "sync_detail": sync_detail,
+            "vm_retained": False,
         }
 
     def _stop_dead_socket_vms(self, entries):

cloud/chv/hybrid_events.py

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

cloud/tests/test_chv_hybrid_c.py

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