merge(chv-hybrid): 合入 C 轨——autoscaler 池目标数 + 回收同步门 + 审计事件目录

合入 feat-chv-hybrid-c (39c51b7)。executor.py 自动合入，无冲突；
autoscaler.py / hybrid_events.py / test_chv_hybrid_c.py 直接合入。

集成统一：A/B 两轨按事件名字符串直接发射的审计点收到 C 的
hybrid_events.py 模块名下——executor.py 顶层 import 9 个设计事件
常量（WARM_POOL_ACQUIRE/MISS/RELEASE、VM_RESIDENT_SUSPENDED/
DEMOTED/PROMOTED、RESIDENT_UPGRADE_HINT、SNAPSHOT_TENANT_MISMATCH、
VM_RESTORE_DENIED），13 处字面量（_emit 事件名、审计 kind、
ChvFailure kind）全部改走常量；发射路径 _emit 与 detail 字符串
原样不动（hybrid_events 只定名 + 校验，不发事件）。C 的回收
同步失败事件仍走其原有函数内 import + format_detail，不动。
experimental_pool.py 的两处同名字符串不动——那是实验位自身的
事件，不属混合生产路径（§9.1-10）。

C 轨未决项 1 定夺（_run_with_vm 的 finally 仍是无条件 shutdown，
未接回收同步门）：本轮不接，列为跟进项。理由：设计 §3.3 的同步
门是回收路径（recycle_vm / recycle_persistent_vm），冷任务每任务
生命周期不在其内；finally 里接同步门会改全部任务的关机语义
（一次 push 失败即留 VM 占槽），须另立重试/留存上限/告警的
设计，不是本轮一行的事。

测试：test_chv_hybrid_a/b/c 共 96 项全部通过；全量待跑。

cloud/chv/executor.py

@@ -69,6 +69,17 @@ from cloud.chv.hardening import (
     validate_net_entries,
     validate_snapshot_url,
 )
+from cloud.chv.hybrid_events import (
+    RESIDENT_UPGRADE_HINT,
+    SNAPSHOT_TENANT_MISMATCH,
+    VM_RESIDENT_DEMOTED,
+    VM_RESIDENT_PROMOTED,
+    VM_RESIDENT_SUSPENDED,
+    VM_RESTORE_DENIED,
+    WARM_POOL_ACQUIRE,
+    WARM_POOL_MISS,
+    WARM_POOL_RELEASE,
+)
 from cloud.chv.ssh_transport import build_ssh_command as _build_ssh_cmd
 from cloud.chv.ssh_transport import known_hosts_path as _known_hosts_path
 from cloud.chv.ssh_transport import (
@@ -2524,7 +2535,7 @@ class CloudHypervisorExecutor(Executor):
                 "gitea_synced": bool(record.get("gitea_synced"))},
                 now=_time.time())
 
-        self._emit("warm_pool_acquire", task,
+        self._emit(WARM_POOL_ACQUIRE, task,
                    "vm=%s mode=resumed resume_seconds=%.2f"
                    " warm_hit=%s"
                    % (vm, resume_seconds,
@@ -2747,7 +2758,7 @@ class CloudHypervisorExecutor(Executor):
             "tier": tier, "state": "resident-active",
             "last_active_at": now,
             "lease_expires_at": now + suspend + demote}, now=now)
-        self._emit("vm_resident_promoted",
+        self._emit(VM_RESIDENT_PROMOTED,
                    {"id": "persist-%s" % tenant_id,
                     "project_id": tenant_id},
                    "host_id=%s tenant=%s vm=%s tier=%s"
@@ -2807,7 +2818,7 @@ class CloudHypervisorExecutor(Executor):
                             tenant, {"state": "resident-paused"},
                             now=now)
                         self._emit(
-                            "vm_resident_suspended",
+                            VM_RESIDENT_SUSPENDED,
                             {"id": "persist-%s" % tenant,
                              "project_id": tenant},
                             "host_id=%s tenant=%s vm=%s idle_seconds=%.0f"
@@ -2893,7 +2904,7 @@ class CloudHypervisorExecutor(Executor):
         if snapshot:
             updates["snapshot_url"] = snapshot
         self._write_resident_record(tenant_id, updates, now=now)
-        self._emit("vm_resident_demoted", task,
+        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}
@@ -3110,7 +3121,7 @@ class CloudHypervisorExecutor(Executor):
             metric = self._upgrade_hint_metric(tasks, now)
             data["last_hint_at"] = now
             self._emit(
-                "resident_upgrade_hint",
+                RESIDENT_UPGRADE_HINT,
                 {"id": "pool-%s" % tenant_id,
                  "project_id": tenant_id},
                 "host_id=%s tenant=%s tier=Free metric=%s "
@@ -3862,7 +3873,7 @@ class CloudHypervisorExecutor(Executor):
         slot = self._clean_pool_slot_for_vm(vm)
         if slot is not None:
             self._clear_clean_pool_record(slot)
-        self._emit("warm_pool_release", task,
+        self._emit(WARM_POOL_RELEASE, task,
                    "vm=%s host_id=%s tenant=%s"
                    % (vm, self._host_id(), tenant))
         return {"released": True, "vm": vm, "tenant_id": tenant}
@@ -3966,7 +3977,7 @@ class CloudHypervisorExecutor(Executor):
         tenant = self._tenant_id(task)
         candidate = self._clean_pool_candidate()
         if candidate is None or not tenant:
-            self._emit("warm_pool_miss", task,
+            self._emit(WARM_POOL_MISS, task,
                        "host_id=%s tenant=%s reason=%s"
                        % (self._host_id(), tenant,
                           "pool_empty" if candidate is None
@@ -3986,7 +3997,7 @@ class CloudHypervisorExecutor(Executor):
             "gitea_synced": False, "data_disk": data_disk,
             "updated_at": time.time()})
         self._check_clean_pool_sync(vm, slot, sync_workspace)
-        self._emit("warm_pool_acquire", task,
+        self._emit(WARM_POOL_ACQUIRE, task,
                    "vm=%s host_id=%s tenant=%s mode=resumed "
                    "resume_seconds=%.2f warm_hit=true"
                    % (vm, self._host_id(), tenant, resume_seconds))
@@ -4088,16 +4099,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):
@@ -4115,7 +4143,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>。
@@ -4206,12 +4237,12 @@ class CloudHypervisorExecutor(Executor):
         只带 host_id + 请求租户 + 原因，不带快照内容（不带 URL、
         不带归属 tenant 值）；错误码一律 snapshot_tenant_mismatch。"""
         requester = self._tenant_id(task)
-        self._emit("vm_restore_denied", task,
+        self._emit(VM_RESTORE_DENIED, task,
                    "host_id=%s tenant=%s reason=%s"
                    % (self._host_id(), requester, reason),
-                   "snapshot_tenant_mismatch")
+                   SNAPSHOT_TENANT_MISMATCH)
         raise ChvFailure(
-            "snapshot_tenant_mismatch",
+            SNAPSHOT_TENANT_MISMATCH,
             "快照 tenant 校验失败（%s），拒绝恢复：请求方 "
             "tenant=%s" % (reason, requester))
 
@@ -4242,9 +4273,9 @@ class CloudHypervisorExecutor(Executor):
                    " tenant=%s，拒绝跨用户恢复"
                    % (self.vm_name(task), snapshot_url,
                       marked, requester),
-                   "snapshot_tenant_mismatch")
+                   SNAPSHOT_TENANT_MISMATCH)
         raise ChvFailure(
-            "snapshot_tenant_mismatch",
+            SNAPSHOT_TENANT_MISMATCH,
             "快照归属 tenant=%s，与请求方 tenant=%s 不一致，"
             "拒绝跨用户恢复：%s" % (marked, requester,
                                    snapshot_url))
@@ -4833,7 +4864,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（数据库里的断点与
@@ -4841,9 +4973,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))
@@ -4852,6 +5005,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):