WechatOnCloud/bridge/woc_bridge/ui/watchdog.py

"""WeChat 进程看门狗：10s 检查间隔，连续 2 次失败触发 autofix。

autofix 策略：尝试重启 WeChat 进程（通过 backend.activate_window 或 pgrep+kill）。
失败仅告警不抛异常，避免拖垮 bridge 主流程。
"""

from __future__ import annotations

import asyncio
import logging
import time
from typing import Any, Optional

from woc_bridge.ui.backends.base import BackendProtocol

logger = logging.getLogger("woc-bridge")


class WeChatWatchdog:
    """WeChat 进程看门狗。

    Attributes:
        backend: L1 Backend 实例（用于 pgrep/kill 等子进程操作）
        interval: 检查间隔秒数，默认 10.0
        fail_threshold: 连续失败触发 autofix 的阈值，默认 2
    """

    def __init__(
        self,
        backend: BackendProtocol,
        interval: float = 10.0,
        fail_threshold: int = 2,
        scheduler: Optional[Any] = None,
    ) -> None:
        self.backend = backend
        self.interval = interval
        self.fail_threshold = fail_threshold
        self._scheduler = scheduler  # UIActionScheduler 实例（用于 kill 前 drain + mark_wechat_dead）
        self._fail_count = 0
        self._task: Optional[asyncio.Task] = None
        self._stopped = False

    async def start(self) -> None:
        """启动看门狗后台任务。"""
        if self._task is not None and not self._task.done():
            return
        self._stopped = False
        self._task = asyncio.create_task(self._run())
        logger.info("[watchdog] started (interval=%.1fs)", self.interval)

    async def stop(self) -> None:
        """停止看门狗。"""
        self._stopped = True
        if self._task is not None and not self._task.done():
            self._task.cancel()
            try:
                await asyncio.wait_for(self._task, timeout=2.0)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                pass
        self._task = None
        logger.info("[watchdog] stopped")

    async def _run(self) -> None:
        """主循环：每 interval 秒检查一次。"""
        while not self._stopped:
            try:
                await asyncio.sleep(self.interval)
                await self._check()
            except asyncio.CancelledError:
                break
            except Exception as exc:
                logger.warning("[watchdog] check exception: %s", exc)

    async def _check(self) -> None:
        """执行一次检查：WeChat 运行态 + X11 可用性。"""
        wechat_ok = await self._is_wechat_running()
        x11_ok = await self._is_x11_available()

        if wechat_ok and x11_ok:
            if self._fail_count > 0:
                logger.info(
                    "[watchdog] recovered after %d failures", self._fail_count
                )
            self._fail_count = 0
            return

        self._fail_count += 1
        logger.warning(
            "[watchdog] check failed (wechat=%s x11=%s) fail_count=%d/%d",
            wechat_ok, x11_ok, self._fail_count, self.fail_threshold,
        )

        if self._fail_count >= self.fail_threshold:
            logger.error(
                "[watchdog] fail threshold reached, trigger autofix"
            )
            await self._autofix_wechat()
            # autofix 后重置计数，等下一轮检查
            self._fail_count = 0

    async def _is_wechat_running(self) -> bool:
        """检查 WeChat 进程是否运行（用 pgrep -x wechat）。"""
        try:
            proc = await asyncio.create_subprocess_exec(
                "pgrep", "-x", "wechat",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            try:
                stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=3.0)
            except asyncio.TimeoutError:
                proc.kill()
                await proc.wait()
                return False
            return proc.returncode == 0 and bool(stdout.decode(errors="ignore").strip())
        except Exception as exc:
            logger.debug("[watchdog] pgrep wechat exception: %s", exc)
            return False

    async def _is_x11_available(self) -> bool:
        """检查 X11 是否可用（用 xdpyinfo）。"""
        try:
            proc = await asyncio.create_subprocess_exec(
                "xdpyinfo",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            try:
                await asyncio.wait_for(proc.communicate(), timeout=3.0)
            except asyncio.TimeoutError:
                proc.kill()
                await proc.wait()
                return False
            return proc.returncode == 0
        except Exception as exc:
            logger.debug("[watchdog] xdpyinfo exception: %s", exc)
            return False

    async def _autofix_wechat(self) -> None:
        """自动修复：尝试 kill WeChat 让 autostart 拉起。

        不直接启动 WeChat（避免与 autostart 竞态），仅 kill 旧进程。
        所有子进程超时均执行 proc.kill() 收尸，避免僵尸进程泄漏。

        kill 前若注入了 scheduler，先 drain 等待队列清空，再 mark_wechat_dead
        启动 fast-fail 窗口，避免 kill 后队列内任务形成失败风暴。
        """
        logger.warning("[watchdog] autofix: killing wechat for restart")

        # kill 前等待 UI 操作队列清空（避免 kill 正在执行的 UI 操作）
        # 并启动 fast-fail 窗口拦截 kill 后新入队的非 CRITICAL 任务
        if self._scheduler is not None:
            drained = await self._scheduler.drain(timeout=5.0)
            if not drained:
                try:
                    pending = self._scheduler.pending_count()
                except Exception:
                    pending = -1
                logger.warning(
                    "[watchdog] drain timeout, force kill with %d UI ops pending",
                    pending,
                )
            try:
                self._scheduler.mark_wechat_dead(15.0)
            except Exception as exc:
                logger.warning("[watchdog] mark_wechat_dead failed: %s", exc)

        try:
            # 获取所有 wechat PID
            proc = await asyncio.create_subprocess_exec(
                "pgrep", "-x", "wechat",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            try:
                stdout, _ = await asyncio.wait_for(
                    proc.communicate(), timeout=3.0
                )
            except asyncio.TimeoutError:
                proc.kill()
                await proc.wait()
                logger.warning("[watchdog] autofix pgrep timeout")
                return
            pids = []
            for line in stdout.decode(errors="ignore").splitlines():
                line = line.strip()
                if line:
                    try:
                        pids.append(int(line))
                    except ValueError:
                        pass
            # kill 所有 wechat 进程
            for pid in pids:
                try:
                    kill_proc = await asyncio.create_subprocess_exec(
                        "kill", "-TERM", str(pid),
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE,
                    )
                    try:
                        await asyncio.wait_for(
                            kill_proc.communicate(), timeout=3.0
                        )
                    except asyncio.TimeoutError:
                        kill_proc.kill()
                        await kill_proc.wait()
                    logger.info(
                        "[watchdog] autofix killed wechat pid=%d", pid
                    )
                except Exception as exc:
                    logger.warning(
                        "[watchdog] autofix kill pid=%d failed: %s", pid, exc
                    )
        except Exception as exc:
            logger.error("[watchdog] autofix exception: %s", exc)