Kris 67aab58a00 docs: 新增PRD规范与WechatOnCloud改造、多应用桥接框架文档

新增三份文档：
1. 产品需求文档(PRD)编写规范
2. WechatOnCloud容器化微信改造需求方案
3. 多应用桥接框架整体设计方案

2026-07-18 16:04:25 +08:00

105 KiB

Raw Permalink Blame History

微信 UI 自动化通用架构优化方案

文档版本：v3.0（速度优化版）目标：将当前紧耦合于微信 4.x Linux 单一版本的 xdotool_driver.py 重构为可扩展、可校验、可恢复、高可用、高性能的通用微信 UI 自动化架构。设计原则：

最小可行实现优先：每一阶段都可独立交付并产生可见收益，不允许大爆炸式重构

高可用优先：每一处都考虑并发/竞态/超时/泄漏/降级/熔断/恢复

速度优先：固定 sleep 改自适应等待、会话缓存、DB 校验调优、并行化

失败可观测：所有失败都可被监控、分类、定位

一、背景与现状

1.1 当前实现

当前微信 UI 自动化全部集中在 bridge/woc_bridge/ui/xdotool_driver.py（约 2000+ 行单文件），实现方式：

通过 xdotool search --name "微信" 找窗口（xdotool_driver.py:128）
通过 getwindowgeometry 拿窗口 (X, Y, W, H)
通过硬编码像素偏移计算关键 UI 元素位置（xdotool_driver.py:391-396）
通过 xdotool click 点击，xdotool type 输入文本
每步之间 sleep 1.0s 等待 UI 响应
send_queue 串行化（messaging/send_queue.py），相邻消息最小 3000ms 间隔

1.2 已确认的核心问题（含高可用性复核新增）

1.2.1 功能正确性问题

问题	根因	后果
分辨率/DPI 一变即崩	像素偏移硬编码（xdotool_driver.py:391-396）	1280x720 下点击位置错位
微信更新布局即崩	坐标常量 + 菜单顺序写死	撤回可能变成删除，造成数据丢失
无法校验操作结果	只返回本地生成的 `local_<ts>_<rand>` ID	已记录过"返回 200 但消息未发出"
重试即重发	无幂等键	网络抖动造成重复消息
失败后状态不可知	线性脚本，无状态机	搜索框开着、焦点错位、下一次必败
experimental 功能残留失效代码	forward/add_friend/set_remark 仍用 `ctrl+f`/`ctrl+a`	实际跑不通

1.2.2 高可用性问题（v2.0 新增复核）

问题	严重程度	根因
`asyncio.wait_for` 取消导致 xdotool 子进程泄漏	严重	`_step` 的 deadline 先于 `_run` 内 15s 超时触发，取消时 `communicate` 被停但子进程未 kill
调试截图打满磁盘	严重	`_DEBUG_SCREENSHOTS=True` 每次发送写 5 张 PNG ≈ 5-10MB，一周打满容器磁盘
L1/L2 超时层级冲突	严重	`max(1.0, remaining)` 在剩余 < 15s 时会先于 `_CMD_TIMEOUT_SEC=15s` 触发，导致子进程泄漏
DB 校验失败 ≠ 发送失败	严重	routes/send.py:265 把 `verified=false` 直接抛 `SEND_FAILED`，导致调用方重试即重发
幂等键缺 `client_request_id` 维度	严重	用户主动重发相同内容会被误吞
DB 校验未验证 `talker == to_wxid`	严重	display_name 串号时幂等会"成功"掩盖发错人
bridge 启动时不做全量清场	致命	上次崩溃残留脏状态（搜索框打开/输入框有内容），下一次必败
`_reset_to_idle` 无 post_verify	严重	Esc + BackSpace 无法覆盖 webview/多级菜单/输入法候选框等状态
无错误分类	严重	所有失败都抛 `SEND_FAILED`，调用方无法区分可重试 vs 不可重试
无崩溃 watchdog	严重	微信/Xvnc 崩溃后 bridge 不主动感知，需人工触发 autofix
无 Prometheus metric	严重	纯文本日志无法做趋势分析，无法感知模板失效/失败率突增
无熔断机制	严重	DB 校验/图像匹配连续失败仍每次重试，浪费时间且可能造成重复发送

1.2.3 速度问题（v3.0 新增复核）

通过逐行代码分析 xdotool_driver.py:455-545（_open_session_by_name）与 xdotool_driver.py:550-611（send_text），实测时序拆解如下：

阶段	步骤	耗时	类型
会话定位	find_wechat_window	~50ms	xdotool
	_activate_window_fast	~2s	windowactivate + 轮询
	sleep 1.0s × 7 处	7.0s	固定 sleep
	_click_at × 2	~200ms	xdotool
	_key × 4	~200ms	xdotool
	_paste_via_xclip (name)	~200ms	xdotool type
	_debug_screenshot × 1	~100ms	scrot + 文件
会话定位小计		~10s
发送体	_debug_screenshot × 4	~400ms	scrot + 文件
	_click_input_box	~100ms	xdotool
	_paste_via_xclip (content)	~200ms	xdotool type
	sleep 1.0s × 2 处	2.0s	固定 sleep
	_click_send_button	~100ms	xdotool
发送体小计		~3s
DB 校验	_verify_sent_to_talker	0-10s	轮询 0.5s 间隔
队列延时	send_delay_ms	3.0s	任务后强制 sleep
总计（典型）		~16s
总计（最坏）		~26s	DB 校验 10s + 全部 sleep

核心发现：

固定 sleep 是最大成本：10 处 sleep 1.0s = 10s，占会话定位总耗时的 70%
DB 校验超时过长：10s 超时 + 0.5s 轮询间隔，实际微信 DB 写入通常 < 500ms
调试截图在关键路径上：5 张截图 × ~100ms = 500ms，且写磁盘有 I/O 开销
无会话缓存：每次都重新搜索会话，即使连续发给同一人
_click_at 分两条命令：mousemove + click 分两次 fork，可合并
DB 查询与 UI 操作串行：发送前查 latest_msg_id 与激活窗口串行，可并行

结论：xdotool 命令本身执行很快（毫秒级），慢在固定 sleep 与串行化等待。优化空间巨大，预期可从 16s 降到 3-5s（首条）/ 1-2s（同联系人后续）。

1.3 微信 4.x Linux 技术约束（决定方案边界）

通过项目代码与文档调研确认：

主聊天 UI = Qt 自绘（docker/Dockerfile:28 注释"微信原生版是 Qt 程序"）
WeChatAppEx = Chromium 内核（docker/Dockerfile:41 注释），承载小程序/公众号文章/朋友圈 webview
RadiumWMPF 不响应 X11 修饰键组合（如 Ctrl+V），只响应单字符事件（项目记忆已确认）
不暴露 AT-SPI accessibility 树：Qt 自绘 + Chromium 默认懒启用，dogtail/pyatspi 全部失效
主进程不吃命令行参数（docker/autostart:23-25 注释），无法直接通过 --remote-debugging-port 开启 CDP
send_queue 已实现串行化但无锁、无幂等、无重试（send_queue.py:117）
DB 校验已实现但错误处理不当（routes/send.py:107 _verify_sent_to_talker，10s 轮询）

核心结论：单一方案无法覆盖所有场景。必须采用混合方案——Qt 自绘 UI 用「图像识别 + xdotool」，CEF 部分用 CDP（若可开启），所有操作用状态机编排 + DB 校验兜底，并补齐所有高可用性机制。

二、技术路径选型

2.1 候选方案对比

方案	覆盖范围	元素识别	跨版本	实施成本	风险	决策
A. xdotool + 像素坐标（当前）	全部	无	差	已实现	中	作为兜底保留
B. AT-SPI / dogtail	失效	-	-	-	-	不可用，Qt 自绘无 a11y 树
C. 纯图像识别（OpenCV）	全部	有（模板匹配）	中（需多模板）	中	低	采用
D. CDP 协议	仅 CEF 部分	完整 DOM	强	中高	中	PoC 后决定
E. Frida native hook	全部	函数级	强	极高	高（封号）	不采用
F. pyautogui	与 xdotool 同	同 X11 XTest	同	低	同	不采用，无实质改进

2.2 选型理由

pyautogui 与 xdotool 底层完全相同（都用 X11 XTest 扩展），换皮不解决"微信不响应修饰键"的根本问题
AT-SPI 路线对 Qt 自绘应用完全失效：自绘控件不实现 QAccessibleInterface，a11y 树里只有空 pane，dogtail 拿不到任何可交互节点
CDP 路线可行性未验证：主进程不吃命令行参数，需 LD_PRELOAD 注入或独立启动 WeChatAppEx，ROI 存疑
OpenCV 图像识别 + xdotool 执行是当前最务实路径：保留现有执行链路，仅在定位层加图像校验，最小改动

2.3 最终技术栈

定位层：OpenCV 模板匹配（cv2.matchTemplate 多尺度）+ 窗口几何兜底
执行层：xdotool（保留现有 X11 XTest 链路）
编排层：自写异步状态机（不引入 transitions 库）
校验层：DB 回读（ground truth）+ 截图对比（降级兜底）
去重层：基于 (to_wxid + content_hash + client_request_id) 的 5 分钟幂等窗口
扩展点：CDP 后端（PoC 验证后再实现）
高可用：熔断器 + watchdog + 错误分类 + Prometheus metric + 资源回收

三、目标架构（六层分层 + 高可用横切）

┌─────────────────────────────────────────────────────────────────┐
│ Layer 6: Capability API（对外稳定接口）                          │
│   send_text / send_image / publish_moment / revoke / forward     │
└────────────────────────┬────────────────────────────────────────┘
                         │
┌────────────────────────┴────────────────────────────────────────┐
│ Layer 5: Flow Orchestrator（流程编排 + 幂等 + 重试 + 熔断）      │
│   FlowOrchestrator + IdemCache + CircuitBreaker + SendQueue      │
└────────────────────────┬────────────────────────────────────────┘
                         │
┌────────────────────────┴────────────────────────────────────────┐
│ Layer 4: Flow State Machine（流程状态机 + 回滚 + 启动清场）     │
│   Flow 基类 / SendTextFlow / PublishMomentFlow / ...             │
│   每 transition: pre_guard → action → post_verify → on_failure   │
└────────────────────────┬────────────────────────────────────────┘
                         │
┌────────────────────────┴────────────────────────────────────────┐
│ Layer 3: Action Library（原子动作，可组合）                      │
│   click_element / type_text / key_press / wait_for / screenshot   │
└────────────────────────┬────────────────────────────────────────┘
                         │
┌────────────────────────┴────────────────────────────────────────┐
│ Layer 2: Locator Strategies（多策略定位，可插拔）                │
│   ImageLocator（OpenCV）/ GeomLocator（像素兜底）/ CDPLocator   │
│   统一 Selector → ElementHandle 接口                             │
└────────────────────────┬────────────────────────────────────────┘
                         │
┌────────────────────────┴────────────────────────────────────────┐
│ Layer 1: Backend Adapters（执行后端，多驱动并存）                │
│   XdotoolBackend / OpenCVBackend / CDPBackend（可选）            │
└─────────────────────────────────────────────────────────────────┘

横切关注点（贯穿所有层）：
┌─────────────────────────────────────────────────────────────────┐
│ HA: Watchdog（微信/Xvnc 崩溃感知）                              │
│ HA: Metrics（Prometheus 端点）                                   │
│ HA: ResourceReaper（截图清理 / 模板 LRU / 子进程收尸）          │
│ HA: ErrorClassifier（错误码分类 + 重试策略）                    │
│ HA: TraceID（贯穿单次请求的所有日志行）                          │
└─────────────────────────────────────────────────────────────────┘

3.1 分层职责

层	职责	不做什么
L1 Backend	执行原子操作（click/type/screenshot）	不做元素定位，不做流程编排
L2 Locator	找元素位置，返回 ElementHandle	不执行点击，不决定流程顺序
L3 Action	组合 L1+L2 完成有语义的原子动作（如"点击并校验"）	不跨步骤
L4 Flow	编排多步骤完成业务流程（如"发文本"）	不直接调 xdotool
L5 Orchestrator	幂等检查、入队、重试策略、熔断	不关心 UI 细节
L6 Capability	对外稳定的业务 API	不关心实现后端

3.2 依赖方向

L6 → L5 → L4 → L3 → L2 → L1
L4 可直接调 L3，不绕道 L2（Action 内部组合 L1+L2）
L5 可直接访问 DB（用于幂等与校验），不绕道 L4
横切 HA 模块可被任意层调用

3.3 并发模型（关键决策）

决策：串行化只在 send_queue 一层，Flow 内部不再加锁

理由：send_queue 的单 worker + asyncio.Queue 已经保证了"同一时刻只有一个 Flow 在跑"（send_queue.py:117 的 while True 循环）。如果在 Flow 内再加 asyncio.Lock 会形成"双重串行"，不死锁但语义混乱、排障困难。

只读探测的并发处理：

detect_login_state / is_wechat_running / get_window_geometry 等只读探测走队列外快速路径
但需共享一把轻量 asyncio.Lock 与发送操作互斥，避免探测期间被发送操作打断
探测用 with lock，发送也用 with lock，但发送额外受队列节流

# bridge/ui/backends/xdotool.py
class XdotoolBackend:
    def __init__(self, display: str = ":1"):
        self.display = display
        self._ui_lock = asyncio.Lock()  # UI 单焦点保护
    
    async def click(self, x: int, y: int) -> None:
        async with self._ui_lock:
            # 实际点击
            ...

注意：这把锁的粒度是"单次原子操作"，不是"整个 Flow"。Flow 的串行性由 send_queue 保证，锁只防止探测与发送的并发冲突。

3.4 多实例隔离（MVP 不涉及，但架构必须预留）

问题：当前 config.py:126 display: str = ":1" 是全局单例。两个微信实例共享同一 X server 时，xdotool search --name "微信" 跨实例匹配，鼠标键盘事件跨实例串扰。

架构预留：

XdotoolBackend 必须接受 display 参数（当前已做到，xdotool_driver.py:39）
xdotool search 改用 --pid <wechat_pid> 精确匹配本实例进程的窗口
AppState 改为 per-instance 而非全局单例（属于阶段 5 多应用框架）
MVP 阶段单实例，不触发此问题

四、核心模块详细设计

4.1 Layer 1 - Backend Adapters

职责：把现有的 xdotool 调用、scrot 截图、getwindowgeometry 等封装为统一后端接口。不做元素定位。

4.1.1 BackendProtocol 定义

# bridge/ui/backends/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
import asyncio

@dataclass
class WindowGeometry:
    x: int; y: int; width: int; height: int

class BackendProtocol(ABC):
    """所有执行后端必须实现的原子能力。"""
    
    @abstractmethod
    async def click(self, x: int, y: int) -> None:
        """在屏幕绝对坐标点击。"""
    
    @abstractmethod
    async def type_text(self, text: str) -> None:
        """逐字符输入文本（xdotool type，绕过修饰键问题）。"""
    
    @abstractmethod
    async def key_press(self, key: str, repeat: int = 1) -> None:
        """按键，支持 repeat。"""
    
    @abstractmethod
    async def screenshot(self) -> bytes:
        """截全屏，返回 PNG bytes。"""
    
    @abstractmethod
    async def get_window_geometry(self, window_title: str) -> WindowGeometry:
        """按窗口标题找窗口并返回几何。"""
    
    @abstractmethod
    async def activate_window(self, window_title: str) -> None:
        """激活指定窗口。"""
    
    @property
    def capabilities(self) -> set[str]:
        """该后端支持的能力集合。"""
        return {"click", "type", "key", "screenshot", "window"}

4.1.2 XdotoolBackend 实现（含超时与取消安全）

关键高可用性设计：_run 改为上下文管理器，确保任何取消场景下子进程都被 kill + wait 收尸。

# bridge/ui/backends/xdotool.py
import asyncio
import os
import logging
from contextlib import asynccontextmanager
from typing import Optional, AsyncIterator

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

# 分层超时（关键决策，解决 L1/L2 冲突）
_CMD_TIMEOUT_SEC = 5.0           # L1: 单命令超时，防子进程挂死（从 15s 降到 5s）
_STEP_MIN_TIMEOUT_SEC = 7.0       # L2: 单步骤超时下限 = L1 + 2s，避免 L2 先于 L1 触发
_FLOW_TIMEOUT_SEC = 30.0          # L3: 整流程超时
_HTTP_TIMEOUT_SEC = 60.0          # L4: HTTP 请求超时 = L3 + DB 校验 10s + 余量

class XdotoolBackend(BackendProtocol):
    def __init__(self, display: str = ":1"):
        self.display = display
        self._ui_lock = asyncio.Lock()  # UI 单焦点保护（与只读探测共享）
    
    def _env(self) -> dict:
        env = dict(os.environ)
        env["DISPLAY"] = self.display
        return env
    
    @asynccontextmanager
    async def _run_managed(self, args: list[str]) -> AsyncIterator[tuple[int, bytes, bytes]]:
        """安全的子进程执行上下文管理器。
        
        保证：
        - 无论协程如何退出（正常返回/超时/取消/异常），子进程都会被 kill + wait
        - 超时不重试，退避交给上层
        """
        proc = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=self._env(),
        )
        try:
            try:
                stdout, stderr = await asyncio.wait_for(
                    proc.communicate(),
                    timeout=_CMD_TIMEOUT_SEC,
                )
                yield proc.returncode, stdout, stderr
            except (asyncio.TimeoutError, asyncio.CancelledError):
                # 超时或取消时，显式 kill + wait 收尸，防孤儿进程
                proc.kill()
                try:
                    await asyncio.wait_for(proc.wait(), timeout=5.0)
                except asyncio.TimeoutError:
                    logger.error("[backend] proc.wait() 超时，孤儿进程残留: %s", args[0])
                raise
        except Exception:
            # 兜底：任何异常都确保 kill
            if proc.returncode is None:
                proc.kill()
                try:
                    await proc.wait()
                except Exception:
                    pass
            raise
    
    async def _run(self, args: list[str]) -> tuple[int, bytes, bytes]:
        """执行一条命令并返回 (returncode, stdout, stderr)。"""
        async with self._run_managed(args) as result:
            return result
    
    async def click(self, x: int, y: int) -> None:
        async with self._ui_lock:
            await self._run(["xdotool", "mousemove", "--sync", str(x), str(y)])
            await self._run(["xdotool", "click", "1"])
    
    async def type_text(self, text: str) -> None:
        async with self._ui_lock:
            await self._run(["xdotool", "type", "--", text])
    
    async def key_press(self, key: str, repeat: int = 1) -> None:
        async with self._ui_lock:
            args = ["xdotool", "key"]
            if repeat > 1:
                args.extend(["--repeat", str(repeat)])
            args.append(key)
            await self._run(args)
    
    async def screenshot(self) -> bytes:
        """截图并返回 PNG bytes（不写文件，避免磁盘泄漏）。"""
        async with self._ui_lock:
            proc = await asyncio.create_subprocess_exec(
                ["scrot", "-o", "-"],  # 输出到 stdout
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
                env=self._env(),
            )
            try:
                stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5.0)
                return stdout
            except (asyncio.TimeoutError, asyncio.CancelledError):
                proc.kill()
                await proc.wait()
                raise

关键改进：

_CMD_TIMEOUT_SEC 从 15s 降到 5s（xdotool 命令本身是毫秒级，5s 足够兜底）
_run_managed 上下文管理器确保任何取消场景都 kill + wait
截图用 scrot -o - 输出到 stdout，不写文件，避免磁盘泄漏
_ui_lock 保护 UI 单焦点

4.1.3 OpenCVBackend（图像识别专用）

# bridge/ui/backends/opencv.py
import cv2
import numpy as np
from collections import OrderedDict
from typing import Optional

class OpenCVBackend:
    """图像识别后端，只做 find_element，不执行点击。"""
    
    def __init__(self, template_dir: str, max_cache_size: int = 20):
        self.template_dir = template_dir
        # LRU 模板缓存，上限 20 个（避免内存泄漏）
        self._cache: OrderedDict[str, np.ndarray] = OrderedDict()
        self._max_cache_size = max_cache_size
    
    def _load_template(self, name: str) -> np.ndarray:
        if name in self._cache:
            self._cache.move_to_end(name)  # LRU 更新
            return self._cache[name]
        path = os.path.join(self.template_dir, name)
        img = cv2.imread(path, cv2.IMREAD_GRAYSCALE)
        if img is None:
            raise FileNotFoundError(f"模板不存在: {path}")
        self._cache[name] = img
        if len(self._cache) > self._max_cache_size:
            self._cache.popitem(last=False)  # LRU 淘汰
        return img
    
    async def find_template(
        self, screenshot_png: bytes, template_name: str,
        threshold: float = 0.80,
        scales: list[float] = None,
        roi: Optional[tuple[int, int, int, int]] = None,
    ) -> Optional[tuple[int, int, int, int, float]]:
        """在截图中找模板，返回 (x, y, w, h, confidence) 或 None。
        
        Args:
            screenshot_png: PNG bytes
            template_name: 模板文件相对路径
            threshold: 匹配阈值（默认 0.80）
            scales: 多尺度匹配列表（默认 [1.0, 0.75, 1.25, 1.5]）
            roi: (x, y, w, h) 限制搜索区域（窗口几何内），None 表示全屏
        """
        scales = scales or [1.0, 0.75, 1.25, 1.5]
        # PNG bytes → np.ndarray（灰度）
        img_arr = np.frombuffer(screenshot_png, np.uint8)
        img = cv2.imdecode(img_arr, cv2.IMREAD_GRAYSCALE)
        if roi:
            x, y, w, h = roi
            img = img[y:y+h, x:x+w]
        tpl = self._load_template(template_name)
        
        best_val, best_loc, best_size = 0.0, None, (0, 0)
        # 逐个尺度处理，避免并行加载多个缩放图占内存
        for scale in scales:
            scaled = cv2.resize(tpl, None, fx=scale, fy=scale)
            res = cv2.matchTemplate(img, scaled, cv2.TM_CCOEFF_NORMED)
            _, max_val, _, max_loc = cv2.minMaxLoc(res)
            if max_val > best_val:
                best_val = max_val
                best_loc = max_loc
                best_size = (scaled.shape[1], scaled.shape[0])
            del scaled, res  # 显式释放
        
        if best_val < threshold:
            return None
        # 如果用了 ROI，坐标要加上 ROI 偏移
        if roi:
            best_loc = (best_loc[0] + roi[0], best_loc[1] + roi[1])
        return (best_loc[0], best_loc[1], best_size[0], best_size[1], best_val)

关键改进：

LRU 模板缓存上限 20 个（每个约 2MB，总上限 40MB）
多尺度匹配逐个处理，显式 del 释放，避免并行占内存
支持 ROI 限制（只在窗口区域内搜），性能提升约 4 倍
返回置信度，供熔断决策

4.2 Layer 2 - Locator Strategies

职责：把"找元素"与"点元素"分离。

# bridge/ui/locators/base.py
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class GeomSpec:
    """窗口几何相对坐标兜底定位。"""
    relative_to: str = "window"  # window / window_top_left / window_bottom_right
    x_ratio: float = 0.0
    y_ratio: float = 0.0
    x_offset: int = 0
    y_offset: int = 0

@dataclass
class Selector:
    """声明式元素选择器，可同时配置多种定位策略。"""
    kind: str                          # 'search_box' / 'send_button' / 'input_box'
    by_image: Optional[str] = None     # 模板图文件名
    by_geom: Optional[GeomSpec] = None # 几何兜底
    threshold: float = 0.80            # 图像匹配阈值
    require_image: bool = False         # 高风险操作设 True，图像失败即拒绝
    description: str = ""

@dataclass
class ElementHandle:
    """统一的元素句柄，由 Locator 产出。"""
    selector: Selector
    x: int; y: int
    w: int = 0; h: int = 0
    confidence: float = 1.0
    strategy: str = "geom"              # 'image' / 'geom'
    
    @property
    def center(self) -> tuple[int, int]:
        return (self.x, self.y)

LocatorRegistry 加载 YAML profile，按 (app_version, resolution) 选择：

# bridge/ui/locators/registry.py
class LocatorRegistry:
    """加载 UI profile YAML，提供 selector 查询。
    
    支持深色/浅色模式切换（通过运行时 set_theme）。
    """
    
    def __init__(self, profile_dir: str, opencv: OpenCVBackend):
        self.profile_dir = profile_dir
        self.opencv = opencv
        self.selectors: dict[str, Selector] = {}
        self.theme: str = "light"  # 'light' / 'dark'
        self._fail_counts: dict[str, int] = {}  # per-kind 失败计数（熔断用）
        self._circuit_open_until: dict[str, float] = {}  # per-kind 熔断截止时间
    
    def load(self, app_version: str, resolution: tuple[int, int]) -> None:
        """按 (版本, 分辨率) 加载最匹配的 profile。"""
        candidates = [
            f"wechat_{app_version}_{resolution[0]}x{resolution[1]}.yaml",
            f"wechat_{app_version}.yaml",
            "default.yaml",
        ]
        for fname in candidates:
            path = os.path.join(self.profile_dir, fname)
            if os.path.exists(path):
                self._load_yaml(path)
                logger.info("[locator] 加载 profile: %s", fname)
                return
        logger.warning("[locator] 无可用 UI profile，回退到硬编码默认值")
        self._load_defaults()
    
    def set_theme(self, theme: str) -> None:
        """切换深色/浅色模式模板目录。"""
        if theme != self.theme:
            self.theme = theme
            logger.info("[locator] 切换主题: %s", theme)
    
    def get(self, kind: str) -> Selector:
        return self.selectors.get(kind) or self._default_selector(kind)
    
    def is_circuited(self, kind: str) -> bool:
        """该 kind 的图像匹配是否被熔断。"""
        deadline = self._circuit_open_until.get(kind, 0)
        return time.monotonic() < deadline
    
    def record_image_failure(self, kind: str) -> None:
        """记录图像匹配失败，连续 10 次触发熔断 5 分钟。"""
        self._fail_counts[kind] = self._fail_counts.get(kind, 0) + 1
        if self._fail_counts[kind] >= 10:
            self._circuit_open_until[kind] = time.monotonic() + 300
            logger.warning("[locator] kind=%s 图像匹配熔断 5 分钟", kind)
    
    def record_image_success(self, kind: str) -> None:
        """记录图像匹配成功，重置失败计数。"""
        self._fail_counts[kind] = 0
        self._circuit_open_until.pop(kind, None)

Profile YAML 示例（bridge/ui/profiles/wechat_4.0_1920x1080.yaml）：

以下坐标基于用户实际截图估算：

容器分辨率 1920x1080，DPI 96，浅色模式
左侧会话栏宽度约 300px
搜索框位于左栏顶部，未激活状态为空白输入框
发送按钮位于右下角，激活态为绿色

# 微信 4.0 Linux, 1920x1080 分辨率，浅色模式
# 截图参考：bridge/ui/profiles/templates/screenshot/03-主界面.png
search_box:
  by_image: "wechat_4.0/light/search_box.png"
  # 裁剪建议：从 03-主界面.png 截取 x=40~270, y=32~68
  by_geom:
    relative_to: window_top_left
    x_ratio: 0.08     # 左栏中部，约 150px @ 1920w
    y_ratio: 0.05     # 约 55px
    width_ratio: 0.12 # 约 230px
    height_ratio: 0.04 # 约 40px
  threshold: 0.80
  description: "左栏顶部搜索框（未激活状态）"

# 截图参考：02-聊天界面.png
send_button:
  by_image: "wechat_4.0/light/send_button.png"
  # 裁剪建议：截取右下角绿色"发送"按钮，约 60x30
  by_geom:
    relative_to: window_bottom_right
    x_offset: -60
    y_offset: -30
  threshold: 0.85
  require_image: false  # 发送按钮不强制图像（geom 兜底足够）
  description: "右下角发送按钮"

# 截图参考：02-聊天界面.png
input_box:
  by_image: "wechat_4.0/light/input_box.png"
  # 裁剪建议：截取底部输入框区域，约 400x50
  by_geom:
    relative_to: window
    x_ratio: 0.65     # 右侧面板中部偏左
    y_ratio: 0.93     # 底部上方
  threshold: 0.75
  description: "聊天输入框"

# 用于发送后校验输入框是否清空
input_box_empty:
  by_image: "wechat_4.0/light/input_box_empty.png"
  by_geom:
    relative_to: window
    x_ratio: 0.65
    y_ratio: 0.93
  threshold: 0.80
  description: "空输入框（发送后校验用）"

# 搜索结果第一项，约搜索框下方 70px
search_result_first:
  by_image: "wechat_4.0/light/search_result_first.png"
  by_geom:
    relative_to: window_top_left
    x_ratio: 0.08
    y_ratio: 0.11    # 约 120px，会话项高度约 70px
  threshold: 0.80
  description: "搜索结果第一项"

# 主界面模板（校验是否回到初始空白态）
# 截图参考：03-主界面.png
main_view:
  by_image: "wechat_4.0/light/main_view.png"
  # 裁剪建议：截取右侧主区域空白+微信 logo，约 400x300
  by_geom:
    relative_to: window
    x_ratio: 0.65
    y_ratio: 0.50
  threshold: 0.75
  description: "主界面空白态（校验用）"

# 高风险操作强制图像命中（阶段 2 后补模板）
revoke_menu_item:
  by_image: "wechat_4.0/light/revoke_menu_item.png"
  threshold: 0.85
  require_image: true

4.3 Layer 3 - Action Library

# bridge/ui/actions.py
class Actions:
    def __init__(self, backend: BackendProtocol, locator: LocatorRegistry):
        self.backend = backend
        self.locator = locator
    
    async def find_element(
        self, kind: str, win_geom: WindowGeometry,
    ) -> ElementHandle:
        """按 selector 找元素。优先图像，失败回退几何。
        
        高风险操作（require_image=True）图像失败即抛异常。
        """
        sel = self.locator.get(kind)
        
        # 策略1：图像匹配（未熔断时）
        if sel.by_image and not self.locator.is_circuited(kind):
            shot = await self.backend.screenshot()
            roi = (win_geom.x, win_geom.y, win_geom.width, win_geom.height)
            result = await self.backend.find_template(
                shot, sel.by_image, threshold=sel.threshold, roi=roi,
            )
            if result:
                x, y, w, h, conf = result
                self.locator.record_image_success(kind)
                return ElementHandle(
                    selector=sel, x=x+w//2, y=y+h//2, w=w, h=h,
                    confidence=conf, strategy="image",
                )
            # 图像失败
            self.locator.record_image_failure(kind)
            logger.warning("[action] 图像匹配失败 kind=%s conf=%.3f", kind, 0.0)
            
            # 高风险操作：图像失败即拒绝
            if sel.require_image:
                raise ElementNotFoundError(
                    kind=kind, reason="image_match_failed_require_image",
                )
            logger.warning("[action] 图像匹配失败 kind=%s，回退几何", kind)
        
        # 策略2：几何兜底
        return self._geom_find(sel, win_geom)
    
    async def click_element(self, kind: str, win_geom: WindowGeometry) -> ElementHandle:
        """点击元素，返回元素句柄。"""
        elem = await self.find_element(kind, win_geom)
        await self.backend.click(elem.x, elem.y)
        logger.info("[action] click %s at (%d,%d) strategy=%s conf=%.3f",
                    kind, elem.x, elem.y, elem.strategy, elem.confidence)
        return elem
    
    async def type_into(self, kind: str, text: str, win_geom: WindowGeometry,
                       clear_first: bool = True) -> None:
        """点击元素 + 清空 + 输入文本。"""
        await self.click_element(kind, win_geom)
        if clear_first:
            await self.backend.key_press("BackSpace", repeat=30)
        await self.backend.type_text(text)
    
    async def wait_for(self, kind: str, win_geom: WindowGeometry,
                      timeout: float = 5.0, interval: float = 0.3) -> Optional[ElementHandle]:
        """轮询等待某元素出现。"""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                elem = await self.find_element(kind, win_geom)
                if elem:
                    return elem
            except ElementNotFoundError:
                pass
            await asyncio.sleep(interval)
        return None
    
    async def screenshot(self) -> bytes:
        return await self.backend.screenshot()

4.4 Layer 4 - Flow State Machine

最关键的架构升级。含启动清场、post_verify、安全重置。

4.4.1 状态定义

# bridge/ui/flow.py
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional

class FlowState(Enum):
    INITIAL = "initial"
    WINDOW_ACTIVATED = "window_activated"
    SEARCH_BOX_CLICKED = "search_box_clicked"
    QUERY_TYPED = "query_typed"
    SESSION_OPENED = "session_opened"
    INPUT_FOCUSED = "input_focused"
    TEXT_TYPED = "text_typed"
    SENT = "sent"
    CLEANED_UP = "cleaned_up"
    FAILED = "failed"

@dataclass
class FlowContext:
    """流程上下文。"""
    to_wxid: str
    content: str
    display_name: str
    client_request_id: str = ""         # 幂等键维度
    win_geom: Optional[WindowGeometry] = None
    before_msg_id: int = 0
    local_send_id: str = ""
    verified: bool = False
    image_confidence: float = 0.0       # 供 metric 上报

@dataclass
class FlowResult:
    success: bool
    local_id: str = ""
    verified: bool = False
    skipped: bool = False
    error: str = ""
    error_code: str = ""
    state: str = ""
    image_confidence: float = 0.0
    duration_ms: int = 0

class FlowError(Exception):
    """流程错误基类。"""
    def __init__(self, code: str, message: str, retryable: bool = False):
        self.code = code
        self.message = message
        self.retryable = retryable
        super().__init__(message)

class TransientError(FlowError):
    """瞬时错误，可重试。"""
    def __init__(self, code: str, message: str):
        super().__init__(code, message, retryable=True)

class PermanentError(FlowError):
    """永久错误，不可重试。"""
    def __init__(self, code: str, message: str):
        super().__init__(code, message, retryable=False)

4.4.2 Flow 基类（含全量清场 + post_verify）

# bridge/ui/flow.py
class Flow:
    """流程基类。"""
    
    def __init__(self, actions: Actions, db_reader=None):
        self.actions = actions
        self.db = db_reader
        # 注意：不加 asyncio.Lock，串行性由 send_queue 保证
    
    async def _reset_to_idle(self, win_geom: Optional[WindowGeometry]) -> bool:
        """安全重置到 IDLE 状态。
        
        多策略清场：
        1. Esc×3（关多级菜单）
        2. 点击窗口空白区域（退 webview 焦点）
        3. BackSpace×30（清搜索框）
        4. Esc×2（关残留弹窗）
        5. 截图校验"主界面"，失败返回 False
        
        幂等：重复调用无副作用。
        失败不抛异常（吞掉原始错误），返回 False 表示未恢复。
        """
        try:
            for _ in range(3):
                await self.actions.backend.key_press("Escape")
                await asyncio.sleep(0.3)
            # 点击窗口左上角空白区域（退 webview 焦点）
            if win_geom:
                await self.actions.backend.click(win_geom.x + 10, win_geom.y + 10)
                await asyncio.sleep(0.3)
            await self.actions.backend.key_press("BackSpace", repeat=30)
            await asyncio.sleep(0.3)
            for _ in range(2):
                await self.actions.backend.key_press("Escape")
                await asyncio.sleep(0.3)
            # 截图校验：是否回到主界面
            # 如果有 main_view.png 模板，匹配它
            return await self._verify_main_view(win_geom)
        except Exception as e:
            logger.error("[flow] _reset_to_idle 异常: %s", e)
            return False
    
    async def _verify_main_view(self, win_geom: Optional[WindowGeometry]) -> bool:
        """校验是否回到主界面（匹配 main_view 模板）。"""
        try:
            elem = await self.actions.wait_for("main_view", win_geom, timeout=2.0)
            return elem is not None
        except Exception:
            # 无模板时降级为 True（不阻塞流程）
            return True
    
    async def _full_cleanup_on_startup(self) -> bool:
        """bridge 启动时的全量清场。
        
        上次崩溃可能残留脏状态。连续重置 3 次，仍失败则告警。
        """
        win_geom = None
        try:
            win_geom = await self.actions.backend.get_window_geometry("微信")
        except Exception:
            logger.warning("[flow] 启动清场：微信窗口未找到，跳过")
            return True  # 窗口都没有，无需清场
        
        for attempt in range(3):
            if await self._reset_to_idle(win_geom):
                logger.info("[flow] 启动清场成功 (attempt=%d)", attempt + 1)
                return True
            logger.warning("[flow] 启动清场失败 (attempt=%d)", attempt + 1)
        
        logger.error("[flow] 启动清场 3 次失败，可能需要人工 VNC 接入")
        return False

4.4.3 SendTextFlow 实现

# bridge/ui/flows/send_text.py
class SendTextFlow(Flow):
    """发文本流程的状态机。"""
    
    async def run(self, ctx: FlowContext) -> FlowResult:
        t_start = time.perf_counter()
        state = FlowState.INITIAL
        try:
            # 整流程超时保护（L3）
            async with asyncio.timeout(_FLOW_TIMEOUT_SEC):
                state = await self._activate(ctx)
                state = await self._click_search_box(ctx)
                state = await self._type_query(ctx)
                state = await self._open_session(ctx)
                state = await self._focus_input(ctx)
                state = await self._type_content(ctx)
                state = await self._send(ctx)
                state = await self._cleanup(ctx)
            
            return FlowResult(
                success=True,
                local_id=ctx.local_send_id,
                verified=ctx.verified,
                state=state.value,
                image_confidence=ctx.image_confidence,
                duration_ms=int((time.perf_counter() - t_start) * 1000),
            )
        except FlowError as e:
            logger.error("[flow] 失败 state=%s code=%s err=%s", state.value, e.code, e.message)
            # 失败后重置（不阻塞原始错误传播）
            await self._reset_to_idle(ctx.win_geom)
            return FlowResult(
                success=False,
                error=e.message,
                error_code=e.code,
                state=state.value,
                duration_ms=int((time.perf_counter() - t_start) * 1000),
            )
        except asyncio.TimeoutError:
            logger.error("[flow] 整流程超时 state=%s", state.value)
            await self._reset_to_idle(ctx.win_geom)
            return FlowResult(
                success=False,
                error="整流程超时",
                error_code="SEND_TIMEOUT",
                state=state.value,
                duration_ms=int((time.perf_counter() - t_start) * 1000),
            )
    
    async def _activate(self, ctx: FlowContext) -> FlowState:
        await self.actions.backend.activate_window("微信")
        await asyncio.sleep(1.0)
        ctx.win_geom = await self.actions.backend.get_window_geometry("微信")
        await self.actions.backend.key_press("Escape")
        await asyncio.sleep(0.5)
        return FlowState.WINDOW_ACTIVATED
    
    async def _click_search_box(self, ctx: FlowContext) -> FlowState:
        await self.actions.click_element("search_box", ctx.win_geom)
        await asyncio.sleep(1.0)
        return FlowState.SEARCH_BOX_CLICKED
    
    async def _type_query(self, ctx: FlowContext) -> FlowState:
        await self.actions.backend.key_press("BackSpace", repeat=30)
        await asyncio.sleep(0.5)
        await self.actions.backend.type_text(ctx.display_name)
        await asyncio.sleep(1.0)
        return FlowState.QUERY_TYPED
    
    async def _open_session(self, ctx: FlowContext) -> FlowState:
        # 点击搜索结果第一项（用 geom 兜底，图像匹配不强求）
        result_y = ctx.win_geom.y + 120
        await self.actions.backend.click(ctx.win_geom.x + 150, result_y)
        await asyncio.sleep(1.0)
        await self.actions.backend.key_press("Escape")
        await asyncio.sleep(0.5)
        return FlowState.SESSION_OPENED
    
    async def _focus_input(self, ctx: FlowContext) -> FlowState:
        await self.actions.click_element("input_box", ctx.win_geom)
        await asyncio.sleep(0.3)
        return FlowState.INPUT_FOCUSED
    
    async def _type_content(self, ctx: FlowContext) -> FlowState:
        await self.actions.backend.type_text(ctx.content)
        await asyncio.sleep(1.0)
        return FlowState.TEXT_TYPED
    
    async def _send(self, ctx: FlowContext) -> FlowState:
        # 记录发送前的 DB 最新 msg_id
        if self.db:
            try:
                ctx.before_msg_id = await self.db.get_latest_msg_id(ctx.to_wxid)
            except Exception:
                ctx.before_msg_id = 0
        
        # 点击发送按钮
        elem = await self.actions.click_element("send_button", ctx.win_geom)
        ctx.image_confidence = max(ctx.image_confidence, elem.confidence)
        await asyncio.sleep(1.0)
        
        # 校验：DB 是否出现新消息
        ctx.verified = await self._verify_sent(ctx)
        ctx.local_send_id = f"local_{int(time.time())}_{random.randint(0, 0xFFFFFF):06x}"
        return FlowState.SENT
    
    async def _verify_sent(self, ctx: FlowContext) -> bool:
        """发送后校验，5 秒窗口内 DB 必须出现新消息。
        
        关键：必须验证 talker == to_wxid，防止 display_name 串号。
        """
        if not self.db or ctx.before_msg_id == 0:
            return await self._verify_by_screenshot(ctx)
        
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            msgs = await self.db.get_messages(
                ctx.to_wxid, limit=1, after_local_id=ctx.before_msg_id,
            )
            if msgs:
                msg = msgs[0]
                # 关键校验：talker 必须匹配 + 内容必须匹配 + is_sender=True
                if (msg.talker == ctx.to_wxid 
                    and msg.content == ctx.content 
                    and msg.is_sender):
                    return True
            await asyncio.sleep(0.5)
        return False
    
    async def _verify_by_screenshot(self, ctx: FlowContext) -> bool:
        """降级校验：截图匹配空输入框模板。"""
        try:
            elem = await self.actions.wait_for(
                "input_box_empty", ctx.win_geom, timeout=2.0,
            )
            return elem is not None
        except Exception:
            return False
    
    async def _cleanup(self, ctx: FlowContext) -> FlowState:
        await self.actions.backend.key_press("Escape")
        await asyncio.sleep(0.5)
        return FlowState.CLEANED_UP

4.5 Layer 5 - Flow Orchestrator（含幂等 + 熔断 + 重试）

4.5.1 幂等缓存（含 client_request_id）

# bridge/ui/idem_cache.py
import hashlib
import time
from typing import Any, Optional
from collections import OrderedDict

class IdemCache:
    """内存短时去重，TTL=300s（5 分钟）。
    
    幂等键 = (flow_name, to_wxid, content_hash, client_request_id)
    - client_request_id 非空时：调用方换 id 即可重发
    - client_request_id 为空时：纯 to_wxid + content_hash 去重（旧调用方兼容）
    """
    
    def __init__(self, ttl: int = 300, max_size: int = 1000):
        self.ttl = ttl
        self._cache: OrderedDict[str, tuple[Any, float]] = OrderedDict()
        self._max_size = max_size
    
    def _make_key(self, flow_name: str, to_wxid: str, content: str,
                  client_request_id: str = "") -> str:
        # 用完整 SHA256，不截断（避免短消息碰撞）
        content_hash = hashlib.sha256(content.encode()).hexdigest()
        return f"{flow_name}:{to_wxid}:{content_hash}:{client_request_id}"
    
    def get(self, key: str) -> Optional[Any]:
        if key in self._cache:
            val, ts = self._cache[key]
            if time.monotonic() - ts < self.ttl:
                self._cache.move_to_end(key)
                return val
            del self._cache[key]
        return None
    
    def set(self, key: str, val: Any) -> None:
        self._cache[key] = (val, time.monotonic())
        if len(self._cache) > self._max_size:
            self._cache.popitem(last=False)  # LRU 淘汰
    
    def clear(self) -> None:
        """bridge 重启时清空（内存缓存不持久化）。"""
        self._cache.clear()

4.5.2 熔断器

# bridge/ui/circuit_breaker.py
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 熔断，拒绝请求
    HALF_OPEN = "half_open" # 半开，试探

class CircuitBreaker:
    """通用熔断器。
    
    连续失败 N 次开熔断，冷却期后半开试探，成功则关闭。
    """
    
    def __init__(self, name: str, failure_threshold: int = 5,
                 recovery_timeout: int = 30):
        self.name = name
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self._failures = 0
        self._state = CircuitState.CLOSED
        self._opened_at = 0.0
    
    @property
    def state(self) -> CircuitState:
        if self._state == CircuitState.OPEN:
            if time.monotonic() - self._opened_at >= self.recovery_timeout:
                self._state = CircuitState.HALF_OPEN
                logger.info("[circuit] %s 进入半开状态", self.name)
        return self._state
    
    def allow(self) -> bool:
        """是否允许通过。"""
        return self.state in (CircuitState.CLOSED, CircuitState.HALF_OPEN)
    
    def record_success(self) -> None:
        if self._state == CircuitState.HALF_OPEN:
            logger.info("[circuit] %s 半开试探成功，关闭熔断", self.name)
        self._failures = 0
        self._state = CircuitState.CLOSED
    
    def record_failure(self) -> None:
        self._failures += 1
        if self._failures >= self.failure_threshold:
            self._state = CircuitState.OPEN
            self._opened_at = time.monotonic()
            logger.warning("[circuit] %s 熔断开启（失败 %d 次）", self.name, self._failures)

4.5.3 FlowOrchestrator

# bridge/ui/orchestrator.py
class FlowOrchestrator:
    """管理所有 UI 流程的编排、幂等、重试、熔断。"""
    
    def __init__(self, send_queue: SendQueue, idem_cache: IdemCache,
                 actions: Actions, db_reader=None):
        self.send_queue = send_queue
        self.idem_cache = idem_cache
        self.actions = actions
        self.db = db_reader
        # DB 校验熔断器：连续失败 5 次熔断 30s
        self.db_verify_breaker = CircuitBreaker("db_verify", 5, 30)
    
    async def send_text(self, to_wxid: str, content: str,
                       display_name: str = None,
                       client_request_id: str = "") -> FlowResult:
        # 1. 幂等检查
        idem_key = self.idem_cache._make_key(
            "send_text", to_wxid, content, client_request_id,
        )
        cached = self.idem_cache.get(idem_key)
        if cached:
            logger.info("[orch] 跳过重复发送 idem_key=%s", idem_key[:32])
            return FlowResult(
                success=True, local_id=cached, verified=True,
                skipped=True, state="idempotent_skip",
            )
        
        # 2. 入队串行执行
        async def _run():
            flow = SendTextFlow(self.actions, self.db)
            ctx = FlowContext(
                to_wxid=to_wxid, content=content,
                display_name=display_name or to_wxid,
                client_request_id=client_request_id,
            )
            return await flow.run(ctx)
        
        try:
            result = await self.send_queue.enqueue(_run)
        except BridgeError as e:
            if e.code == "RATE_LIMITED":
                return FlowResult(
                    success=False, error=e.message, error_code="RATE_LIMITED",
                    state="rate_limited",
                )
            raise
        
        # 3. 成功才写幂等缓存
        if result.success and result.verified:
            self.idem_cache.set(idem_key, result.local_id)
        
        # 4. DB 校验熔断
        if not result.verified:
            self.db_verify_breaker.record_failure()
        else:
            self.db_verify_breaker.record_success()
        
        return result

4.6 Layer 6 - Capability API

# bridge/ui/capabilities.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class SendResult:
    """对调用方的稳定响应。"""
    success: bool                # 发送动作是否完成
    local_id: str = ""
    verified: bool = False       # DB 校验是否通过
    skipped: bool = False        # 是否因幂等跳过
    error: str = ""
    error_code: str = ""         # 错误分类（见错误码表）
    retryable: bool = False      # 是否可重试
    diagnostics: dict = None     # 置信度、耗时等诊断信息
    
class WeChatCapabilities:
    """对外稳定的业务接口，与底层后端完全解耦。"""
    
    def __init__(self, orchestrator: FlowOrchestrator):
        self.orch = orchestrator
    
    async def send_text(self, to_wxid: str, content: str,
                       display_name: str = None,
                       client_request_id: str = "") -> SendResult:
        result = await self.orch.send_text(
            to_wxid, content, display_name, client_request_id,
        )
        return SendResult(
            success=result.success,
            local_id=result.local_id,
            verified=result.verified,
            skipped=result.skipped,
            error=result.error,
            error_code=result.error_code,
            retryable=result.error_code in RETRYABLE_CODES,
            diagnostics={
                "image_confidence": result.image_confidence,
                "duration_ms": result.duration_ms,
                "state": result.state,
            },
        )

五、横切高可用模块

5.1 错误码分类表

# bridge/ui/errors.py

# 可重试错误（瞬时性，重试可能成功）
RETRYABLE_CODES = {
    "RATE_LIMITED",       # 队列限流
    "SEND_TIMEOUT",       # xdotool 超时
    "STATE_DIRTY",        # 状态机脏状态，重置后可重试
    "X11_TEMP_UNAVAILABLE",  # X11 临时不可用
}

# 不可重试错误（永久性，重试无意义）
NON_RETRYABLE_CODES = {
    "WECHAT_NOT_LOGGED_IN",   # 需人工扫码
    "WINDOW_NOT_FOUND",        # 微信未运行
    "ELEMENT_NOT_FOUND",      # 模板失效，需更新模板
    "DB_VERIFY_FAILED",       # 消息可能已发到错会话，需人工排查
    "AMBIGUOUS_CONTACT",      # 重名联系人，需指定更精确 display_name
    "X11_UNAVAILABLE",        # Xvnc 崩溃，需重启容器
    "BRIDGE_CIRCUITED",       # 熔断中
}

# 特殊：DB_VERIFY_FAILED 不可重试
# 理由：消息可能已发出，重试就是重复。调用方应人工排查或等幂等窗口过期。

5.2 重试策略

# bridge/ui/retry.py
import random

class RetryPolicy:
    """重试策略：指数退避 + 抖动。"""
    
    def __init__(self, max_attempts: int = 2, base_delay: float = 1.0,
                 max_delay: float = 30.0):
        self.max_attempts = max_attempts  # UI 自动化最多重试 1 次（不含首次）
        self.base_delay = base_delay
        self.max_delay = max_delay
    
    def get_delay(self, attempt: int) -> float:
        """指数退避 + 抖动。"""
        delay = min(self.base_delay * (2 ** attempt), self.max_delay)
        # 抖动 ±25%，避免多调用方同步重试
        jitter = delay * 0.25 * random.uniform(-1, 1)
        return delay + jitter
    
    def should_retry(self, error_code: str, attempt: int) -> bool:
        """是否应该重试。"""
        if attempt >= self.max_attempts:
            return False
        return error_code in RETRYABLE_CODES

关键决策：

UI 自动化最多重试 1 次（不是 3 次）——重试多了反而容易把状态搞更乱
重试前必须 _reset_to_idle + 截图校验
DB_VERIFY_FAILED 不可重试，避免重复发送
RATE_LIMITED 按 retry_after 等待，不用指数退避

5.3 Watchdog（崩溃感知）

# bridge/ui/watchdog.py
class WeChatWatchdog:
    """后台监控微信与 Xvnc 进程，崩溃主动恢复。"""
    
    def __init__(self, backend: XdotoolBackend, interval: float = 10.0):
        self.backend = backend
        self.interval = interval
        self._task: Optional[asyncio.Task] = None
        self._consecutive_failures = 0
    
    async def start(self) -> None:
        if self._task is None or self._task.done():
            self._task = asyncio.create_task(self._run())
    
    async def stop(self) -> None:
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
    
    async def _run(self) -> None:
        """每 10s 检查一次。"""
        while True:
            try:
                await asyncio.sleep(self.interval)
                await self._check()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error("[watchdog] 检查异常: %s", e)
    
    async def _check(self) -> None:
        # 1. 检查微信进程
        running = await self._is_wechat_running()
        if not running:
            self._consecutive_failures += 1
            if self._consecutive_failures >= 2:
                logger.warning("[watchdog] 微信连续 2 次检测失败，触发 autofix")
                await self._autofix_wechat()
        else:
            self._consecutive_failures = 0
        
        # 2. 检查 X11 server（Xvnc）
        if not await self._is_x11_available():
            logger.error("[watchdog] X11 不可用，Xvnc 可能崩溃")
            # Xvnc 崩溃 bridge 无法自救，告警
            # 容器层 s6 会自动重启 Xvnc
    
    async def _is_wechat_running(self) -> bool:
        try:
            rc, stdout, _ = await self.backend._run(["pgrep", "-x", "wechat"])
            return rc == 0 and bool(stdout.decode(errors="ignore").strip())
        except Exception:
            return False
    
    async def _is_x11_available(self) -> bool:
        try:
            # 检查 X11 socket + xdpyinfo 探测
            rc, _, _ = await self.backend._run(["xdpyinfo", "-display", self.backend.display])
            return rc == 0
        except Exception:
            return False
    
    async def _autofix_wechat(self) -> None:
        """触发微信 autofix：pkill → autostart 拉起 → bridge 兜底启动。"""
        try:
            await self.backend._run(["pkill", "-x", "wechat"])
            await asyncio.sleep(2.0)
            # autostart 会自动拉起，bridge 兜底
            # 复用现有 start_wechat 逻辑
        except Exception as e:
            logger.error("[watchdog] autofix 失败: %s", e)

5.4 ResourceReaper（资源回收）

# bridge/ui/resource_reaper.py
import os
import time
import glob

class ResourceReaper:
    """定期清理临时资源。"""
    
    def __init__(self, debug_screenshot_dir: str = "/tmp/woc_debug",
                 max_age_hours: int = 24, max_total_mb: int = 100):
        self.debug_dir = debug_screenshot_dir
        self.max_age_hours = max_age_hours
        self.max_total_mb = max_total_mb
        self._task: Optional[asyncio.Task] = None
    
    async def start(self) -> None:
        if self._task is None or self._task.done():
            self._task = asyncio.create_task(self._run())
    
    async def _run(self) -> None:
        """每小时清理一次。"""
        while True:
            try:
                await asyncio.sleep(3600)
                await self.cleanup()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error("[reaper] 清理异常: %s", e)
    
    async def cleanup(self) -> None:
        """清理过期调试截图 + 总量超限时 LRU 删除。"""
        if not os.path.exists(self.debug_dir):
            return
        
        now = time.time()
        max_age_sec = self.max_age_hours * 3600
        
        # 1. 删除超龄文件
        for f in glob.glob(os.path.join(self.debug_dir, "*.png")):
            try:
                mtime = os.path.getmtime(f)
                if now - mtime > max_age_sec:
                    os.remove(f)
                    logger.debug("[reaper] 删除过期文件: %s", f)
            except OSError:
                pass
        
        # 2. 总量超限时 LRU 删除
        files = [(f, os.path.getmtime(f)) 
                 for f in glob.glob(os.path.join(self.debug_dir, "*.png"))]
        files.sort(key=lambda x: x[1])  # 按修改时间排序
        
        total_mb = sum(os.path.getsize(f) for f, _ in files) / (1024 * 1024)
        while total_mb > self.max_total_mb and files:
            f, _ = files.pop(0)  # 删最老的
            try:
                os.remove(f)
                total_mb -= os.path.getsize(f) / (1024 * 1024)
            except OSError:
                pass

5.5 Metrics（Prometheus）

# bridge/ui/metrics.py
from prometheus_client import Counter, Histogram, Gauge

# 发送计数
send_total = Counter(
    "woc_send_total", "Total send operations",
    ["result"],  # success / failed / skipped
)

# 发送耗时
send_duration = Histogram(
    "woc_send_duration_seconds", "Send duration",
    buckets=[0.5, 1, 2, 5, 10, 15, 30, 60],
)

# 失败状态分布
send_failed_by_state = Counter(
    "woc_send_failed_by_state_total", "Failed sends by state",
    ["state"],  # window_activated / search_box_clicked / ...
)

# 图像匹配置信度
image_match_confidence = Histogram(
    "woc_image_match_confidence", "Image match confidence",
    buckets=[0.5, 0.7, 0.8, 0.85, 0.9, 0.95, 1.0],
)

# DB 校验耗时
db_verify_duration = Histogram(
    "woc_db_verify_duration_seconds", "DB verify duration",
    buckets=[0.1, 0.5, 1, 2, 5, 10],
)

# 队列深度
send_queue_pending = Gauge(
    "woc_send_queue_pending", "Send queue pending count",
)

# 微信运行状态
wechat_running = Gauge(
    "woc_wechat_running", "WeChat process running (0/1)",
)

# 熔断状态
circuit_state = Gauge(
    "woc_circuit_state", "Circuit breaker state (0=closed, 1=open, 2=half)",
    ["name"],
)

5.6 TraceID

# bridge/ui/trace.py
import uuid
from contextvars import ContextVar

# 贯穿单次请求的所有日志行
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="")

def new_trace_id() -> str:
    tid = uuid.uuid4().hex[:12]
    trace_id_var.set(tid)
    return tid

def get_trace_id() -> str:
    return trace_id_var.get()

# 日志格式中加入 trace_id
class TraceFilter(logging.Filter):
    def filter(self, record):
        record.trace_id = get_trace_id()
        return True

六、目录结构重构

bridge/woc_bridge/
├── ui/
│   ├── __init__.py
│   ├── backends/                    # Layer 1
│   │   ├── __init__.py
│   │   ├── base.py                  # BackendProtocol + WindowGeometry
│   │   ├── xdotool.py              # XdotoolBackend（含超时/取消安全）
│   │   └── opencv.py               # OpenCVBackend（含 LRU 缓存）
│   ├── locators/                    # Layer 2
│   │   ├── __init__.py
│   │   ├── base.py                  # Selector + ElementHandle + GeomSpec
│   │   └── registry.py             # LocatorRegistry（含熔断）
│   ├── actions.py                   # Layer 3
│   ├── flow.py                      # Layer 4 基类
│   ├── flows/                      # Layer 4 具体流程
│   │   ├── __init__.py
│   │   └── send_text.py
│   ├── orchestrator.py              # Layer 5
│   ├── idem_cache.py                # 幂等缓存
│   ├── circuit_breaker.py           # 熔断器
│   ├── retry.py                     # 重试策略
│   ├── errors.py                    # 错误码分类
│   ├── capabilities.py              # Layer 6 对外接口
│   ├── watchdog.py                  # HA: 崩溃感知
│   ├── resource_reaper.py          # HA: 资源回收
│   ├── metrics.py                   # HA: Prometheus
│   ├── trace.py                     # HA: TraceID
│   ├── profiles/                    # UI 元素配置
│   │   ├── default.yaml
│   │   ├── wechat_4.0_1920x1080.yaml
│   │   └── templates/
│   │       └── wechat_4.0/
│   │           ├── light/           # 浅色模式模板
│   │           │   ├── search_box.png
│   │           │   ├── send_button.png
│   │           │   ├── input_box.png
│   │           │   ├── input_box_empty.png
│   │           │   └── main_view.png  # 主界面模板（校验用）
│   │           └── dark/            # 深色模式模板（阶段 2 补）
│   ├── qr_capture.py               # 保留（不变）
│   └── xdotool_driver.py           # 旧文件保留作兼容层
└── ...

七、最小可行实现（MVP）路径

核心原则：每个阶段都可独立交付，每个阶段都有可见收益，不允许大爆炸式重构。

阶段 0：PoC 验证（先验证假设）

PoC 1：OpenCV 模板匹配可行性

容器内手动截微信主界面（搜索框、发送按钮、输入框各一张）
或直接使用用户已提供的截图：03-主界面.png（搜索框）、02-聊天界面.png（发送按钮/输入框）
写 50 行 Python 脚本，用 cv2.matchTemplate 验证能否稳定命中
测试不同分辨率下的匹配度（1280x720 / 1920x1080）
关键：search_box.png 模板必须从未激活搜索框截取，不可用带下拉覆盖层的图
判定标准：阈值 0.80 时命中率 > 95%，多尺度匹配下命中位置误差 < 5px

PoC 2：状态机回滚可行性

把现有 send_text 拆成 5 个状态
在中间状态故意失败（模拟 click_element 返回 None）
验证 _reset_to_idle 能否把微信 UI 恢复到可用状态
判定标准：回滚后下一次发送能正常进行，无残留弹窗/搜索框/焦点错位

PoC 3（可选）：CDP 端口开启可行性

容器内 ss -tlnp | grep -E "9222|DevTools" 检查
尝试独立启动 WeChatAppEx 加 --remote-debugging-port=9222 --user-data-dir=/tmp/wx-debug
curl http://127.0.0.1:9222/json/version 看是否返回 JSON
判定标准：能拿到 webSocketDebuggerUrl → CDP 路线可行

阶段 1：抽象层落地 + P0 高可用修复（行为不变）

目标：把现有代码包装成六层架构的骨架，同时修复 P0 高可用问题，行为完全不变。

任务清单：

新建 bridge/ui/backends/base.py 定义 BackendProtocol
新建 backends/xdotool.py 移植现有方法，关键改进：
- _CMD_TIMEOUT_SEC 从 15s 降到 5s
- _run 改为 _run_managed 上下文管理器，确保取消时 kill+wait
- 截图用 scrot -o - 输出到 stdout，不写文件
- 加 _ui_lock 保护 UI 单焦点
新建 locators/base.py + registry.py
新建 actions.py
新建 profiles/default.yaml 迁移硬编码常量
新建 capabilities.py
P0 修复：
- 生产环境关闭 _DEBUG_SCREENSHOTS（环境变量 WOC_UI_DEBUG_SHOTS=false）
- 修复 _step 的 max(1.0, remaining) 为 max(_STEP_MIN_TIMEOUT_SEC, remaining)
- 启动时执行 _full_cleanup_on_startup
旧 xdotool_driver.py 保留，新代码通过 capabilities.py 调用

验收标准：

现有 /api/send/text 接口行为完全不变
调试截图不再写磁盘（或写到 /tmp/woc_debug/ 并自动清理）
无子进程泄漏（取消时 kill+wait）
bridge 启动后微信 UI 处于已知态

阶段 2：引入图像校验

目标：在定位层加 OpenCV 图像匹配，点击前校验是否命中正确元素。

任务清单：

新建 backends/opencv.py 实现 find_template（含 LRU 缓存）
容器内截取微信关键 UI 元素模板，保存到 profiles/templates/wechat_4.0/light/
截取 main_view.png 主界面模板（供 _verify_main_view 使用）
修改 Actions.find_element：优先图像匹配，失败回退几何
高风险操作（撤回/删除）强制要求图像命中（require_image: true）
加 per-kind 熔断（连续 10 次失败熔断 5 分钟）

验收标准：

1280x720 分辨率下发送文本成功率 > 90%
图像匹配失败时回退到几何定位，不报错
撤回操作在图像未命中时拒绝执行，返回 ELEMENT_NOT_FOUND
图像匹配置信度通过 Prometheus metric 暴露

阶段 3：引入状态机 + DB 校验 + 幂等

目标：把 send_text 重构为状态机，加 DB 回读校验与幂等去重。

任务清单：

新建 flow.py 定义 Flow 基类 + FlowState 枚举 + FlowContext
新建 flows/send_text.py 实现 SendTextFlow（含 post_verify）
新建 orchestrator.py 编排幂等检查 + 入队 + 重试
新建 idem_cache.py 实现内存去重（含 client_request_id）
新建 circuit_breaker.py 实现 DB 校验熔断
新建 errors.py 错误码分类
新建 retry.py 重试策略（指数退避 + 抖动，最多 1 次）
在 SendTextFlow._send 中加 DB 校验：
- 关键：验证 msg.talker == to_wxid（防 display_name 串号）
- DB 校验失败时区分 success=true, verified=false vs success=false
切换 /api/send/text 路由到新实现
新建 metrics.py 暴露 Prometheus 指标
新建 watchdog.py 微信/Xvnc 崩溃感知
新建 resource_reaper.py 资源回收
新建 trace.py TraceID

验收标准：

5 分钟内重复发送相同内容（同 client_request_id）被跳过，返回 skipped=True
不同 client_request_id 的相同内容可正常发送
发送失败（DB 校验未通过）时返回 verified=False，但不返回 SEND_FAILED（调用方不误重试）
发错人时 DB 校验失败（talker 不匹配），返回 DB_VERIFY_FAILED（不可重试）
失败后状态机回滚到 IDLE，下一次发送正常
Prometheus /metrics 端点可查
微信崩溃后 watchdog 10s 内感知并触发 autofix

阶段 4（可选）：CDP 后端 PoC

目标：若 PoC 3 验证 CDP 可行，实现 CDPBackend 用于朋友圈/小程序/公众号文章。

任务清单：

新建 backends/cdp.py 用 Playwright connect_over_cdp 连接
实现 async with 上下文管理，__aexit__ 必 browser.close()
加健康检查：每次操作前 browser.is_connected()，失败重连
CDP 与 xdotool 共享同一个 send_queue 串行（不并发）
page 池上限 5 个，超限拒绝新建
朋友圈发布流程切换到 CDP

验收标准：

朋友圈发布成功率 > 90%
CDP 连接断开后自动重连
无 page 泄漏（cdp_pages_active < 5）

阶段 5：多应用桥接框架对齐

目标：把 WechatDriver 改造为实现多应用桥接框架设计.md 定义的 AppDriver 抽象。

任务清单：

抽象 AppDriver 接口
WechatDriver 实现该接口
落地 AppKindRegistry / AppInstance / BridgeContext
AppState 改为 per-instance（支持多实例 X server 隔离）
新增应用（如 Telegram/Chromium）时复用通用层

八、关键工程决策汇总

8.1 并发模型

串行化只在 send_queue 一层，Flow 内部不加锁（避免双重阻塞）
只读探测走队列外快速路径，但共享 _ui_lock 与发送互斥
多实例需独立 X server（阶段 5）

8.2 超时层级（解决 L1/L2 冲突）

L1 单命令：_CMD_TIMEOUT_SEC=5s（从 15s 降到 5s）
L2 单步骤：max(_STEP_MIN_TIMEOUT_SEC=7s, remaining)（确保 ≥ L1+2s）
L3 整流程：_FLOW_TIMEOUT_SEC=30s
L4 HTTP 请求：_HTTP_TIMEOUT_SEC=60s

8.3 子进程安全

_run_managed 上下文管理器：任何取消/超时/异常都 kill+wait
所有 create_subprocess_exec 必须显式 stdout=PIPE/DEVNULL，禁止默认继承

8.4 资源回收

调试截图：生产环境关闭，调试模式写 /tmp/woc_debug/，每小时清理 + 100MB 上限
模板缓存：LRU 上限 20 个（约 40MB）
OpenCV ndarray：多尺度逐个处理 + 显式 del
Playwright page：async with + 池上限 5

8.5 熔断策略

DB 校验熔断：连续失败 5 次熔断 30s，半开试探
图像匹配熔断：per-kind 连续 10 次失败熔断 5 分钟
熔断后降级：DB → 截图校验；图像 → 几何坐标（高风险操作不降级）

8.6 幂等设计

幂等键 = (flow_name, to_wxid, content_sha256, client_request_id)
用完整 SHA256，不截断（避免短消息碰撞）
5 分钟 TTL
内存缓存不持久化，bridge 重启丢失（可接受，重启不频繁）
关键：DB 校验必须验证 msg.talker == to_wxid（防 display_name 串号）

8.7 状态机恢复

启动时全量清场：_full_cleanup_on_startup 连续重置 3 次 + 截图校验
失败时重置：_reset_to_idle 多策略清场（Esc×3 + 点击空白 + BackSpace×30 + Esc×2 + 校验）
正常路径只做轻清场：Esc×1 + BackSpace×30（约 1s）
重置后必须 post_verify（匹配 main_view 模板）

8.8 错误分类与重试

可重试：RATE_LIMITED / SEND_TIMEOUT / STATE_DIRTY / X11_TEMP_UNAVAILABLE
不可重试：WECHAT_NOT_LOGGED_IN / WINDOW_NOT_FOUND / ELEMENT_NOT_FOUND / DB_VERIFY_FAILED / AMBIGUOUS_CONTACT / X11_UNAVAILABLE / BRIDGE_CIRCUITED
UI 自动化最多重试 1 次（重试前必 _reset_to_idle）
指数退避 + 抖动：base=1s, max=30s
DB_VERIFY_FAILED 不可重试（消息可能已发，重试即重复）

8.9 可观测性

Prometheus /metrics 端点
关键指标：woc_send_total{result} / woc_send_duration_seconds / woc_send_failed_by_state_total{state} / woc_image_match_confidence / woc_db_verify_duration_seconds / woc_send_queue_pending / woc_wechat_running / woc_circuit_state{name}
TraceID 贯穿单次请求所有日志行
调试截图带 trace_id 文件名

8.10 模板维护

深色/浅色模式两套模板（light/ 和 dark/）
当前已提供浅色模式截图素材：02-聊天界面.png、03-主界面.png
01-搜索框.png 含下拉覆盖层，不可作为模板，仅作反面参考
启动时检测主题（截主界面与参考图比对）
微信版本升级监控：woc_image_match_confidence 均值跌至 0.7 以下告警
多尺度匹配 [1.0, 0.75, 1.25, 1.5] 覆盖常见 DPI
采集模板时固定 DPI=96
模板裁剪坐标参考见 9.2 节

九、YAML Profile 规范

9.1 文件命名规则

profiles/
├── default.yaml                         # 兜底默认值
├── wechat_4.0_1920x1080.yaml            # 精确匹配版本+分辨率
├── wechat_4.0_1280x720.yaml
├── wechat_4.0.yaml                      # 仅版本匹配
└── templates/
    └── wechat_4.0/
        ├── light/                       # 浅色模式
        │   ├── search_box.png
        │   ├── send_button.png
        │   ├── input_box.png
        │   ├── input_box_empty.png
        │   └── main_view.png            # 主界面模板（校验用）
        └── dark/                        # 深色模式
            └── ...（同名）

9.2 截图素材参考

用户已提供 3 张微信真实截图，位于 bridge/ui/profiles/templates/screenshot/。阶段 2 模板采集可直接基于这些截图裁剪。

文件名	状态	用途	裁剪建议
`01-搜索框.png`	⚠️ 不可用（含下拉覆盖层）	仅作参考，不可直接作为 `search_box.png` 模板	图中搜索框被点击，出现"搜索网络结果"下拉，模板必须从未激活状态截取
`02-聊天界面.png`	✅ 可用	`send_button.png` / `input_box.png` / `input_box_empty.png`	截取右下角绿色发送按钮、底部输入框、空输入框
`03-主界面.png`	✅ 可用	`search_box.png` / `main_view.png` / `search_result_first.png`	截取左栏顶部未激活搜索框、右侧主界面空白+微信 logo

主题判断：当前为浅色模式（白色背景、绿色气泡），阶段 2 模板保存到 templates/wechat_4.0/light/。深色模式需等切换后重新采集。

裁剪坐标参考（基于 1920x1080 截图）：

模板	来源截图	建议裁剪区域（x, y, w, h）	说明
`search_box.png`	`03-主界面.png`	(40, 32, 230, 36)	左栏顶部搜索框，不包含网络结果下拉
`main_view.png`	`03-主界面.png`	(600, 300, 400, 300)	右侧主区域空白+微信 logo
`search_result_first.png`	`03-主界面.png`	(40, 90, 230, 60)	搜索框下方第一条会话结果
`send_button.png`	`02-聊天界面.png`	(右下角 -80, -45, 70, 35)	绿色激活态"发送"按钮
`input_box.png`	`02-聊天界面.png`	(350, 1010, 500, 40)	底部输入框
`input_box_empty.png`	`02-聊天界面.png`	(350, 1010, 500, 40)	空输入框（灰色提示文字状态）

注意：01-搜索框.png 是错误示例——包含"搜索网络结果"覆盖层。阶段 2 裁剪模板时必须使用 03-主界面.png 中的未激活搜索框。

9.3 YAML 字段规范

<element_kind>:
  by_image: "wechat_4.0/light/send_button.png"
  by_geom:
    relative_to: window
    x_ratio: 0.0
    y_ratio: 0.0
    x_offset: 0
    y_offset: 0
  threshold: 0.80              # 图像匹配阈值
  require_image: false          # 高风险操作设 true
  description: "发送按钮"

9.4 加载优先级

LocatorRegistry.load(app_version, resolution) 按以下顺序查找：

wechat_{app_version}_{W}x{H}.yaml — 精确匹配
wechat_{app_version}.yaml — 版本匹配
default.yaml — 默认兜底

十、迁移与兼容策略

10.1 不破坏现有功能

旧 xdotool_driver.py 保留作为兼容层
新代码通过 capabilities.py 调用
send_queue 保留
DB reader 保留

10.2 渐进式路由切换

路由	阶段 1	阶段 2	阶段 3
`/api/send/text`	旧实现	旧实现	新实现
`/api/send/file`	旧实现	旧实现	新实现
`/api/moments/publish`	旧实现	旧实现	旧实现（CDP 待阶段 4）
`/api/messages/revoke`	旧实现	新实现	新实现

10.3 回滚方案

环境变量 WOC_UI_BACKEND=legacy 强制走旧实现：

# bridge/routes/send.py
if config.ui_backend == "legacy":
    result = await state.xdotool.send_text(...)
else:
    result = await state.capabilities.send_text(...)

十一、预期收益

维度	当前	阶段 1 后	阶段 3 后
跨分辨率	1 个固定布局	同左	多 profile 自适应 + 图像匹配兜底
跨版本	微信更新即崩	同左	模板/profile 单独更新，代码不动
发送成功率	~70%	同左	>95%（DB 校验 + 图像校验）
首条消息耗时	16s	~8s（DB/截图/命令优化）	3-5s（自适应等待 + 并行）
同联系人后续耗时	16s	~8s	1-2s（会话缓存 + 队列自适应）
失败可恢复	不可恢复，半残状态	启动清场 + 重置	状态机重置到 IDLE + 截图校验
重复发送风险	高（重试即重发）	同左	低（5 分钟幂等 + client_request_id）
子进程泄漏	有（取消时不 kill）	无（上下文管理器）	无
磁盘泄漏	有（调试截图堆积）	无（生产关闭 + 自动清理）	无
崩溃感知	无（被动触发）	同左	10s watchdog 主动感知
可观测性	散点日志	同左	Prometheus metric + TraceID
错误分类	全部 SEND_FAILED	DB 校验区分 verified	8 类错误码，区分可重试
熔断保护	无	同左	DB + 图像双熔断

十二、风险与应对

风险	概率	影响	应对
OpenCV 模板匹配在深色模式下失效	中	定位失败	采集深色/浅色两套模板，启动时检测主题
DB 校验延迟过高	低	发送后等 5 秒	降级到截图校验
微信版本升级改变 UI 布局	中	模板失效	`woc_image_match_confidence` 均值监控 + 告警
状态机回滚不彻底	中	下一次发送失败	重置后 post_verify + 3 次重试 + 告警
OpenCV 依赖增加镜像体积	低	镜像变大 ~50MB	可接受
CDP 端口开启后微信不稳定	中	微信崩溃	阶段 4 PoC 充分验证，失败则放弃 CDP
幂等缓存 bridge 重启丢失	低	短窗口内重试重发	可从 DB 重建（远期）
watchdog 误判微信崩溃	低	误 pkill 微信	连续 2 次失败才触发
模板截图包含临时覆盖层导致匹配失败	中	`search_box` 模板命中 0%	模板必须从未激活/稳定态截取；`01-搜索框.png` 已标记为不可用
多 DPI/缩放比例下几何坐标偏移	中	图像未命中且几何不准	多尺度匹配 `[1.0, 0.75, 1.25, 1.5]` + 每分辨率独立 profile

十三、验收标准

13.1 阶段验收

阶段 1 验收：

BackendProtocol 定义完成，XdotoolBackend 实现通过所有现有测试
YAML profile 加载机制工作，default.yaml 与现有硬编码常量一致
capabilities.send_text() 行为与旧 xdotool_driver.send_text() 完全一致
生产环境 _DEBUG_SCREENSHOTS=false
无子进程泄漏（压力测试取消 100 次，无僵尸进程）
bridge 启动后微信 UI 处于已知态（截图校验通过）

阶段 2 验收：

5 个 MVP 模板（搜索框/发送按钮/输入框/空输入框/主界面）采集完成
- 已有截图素材：02-聊天界面.png、03-主界面.png
- search_box.png 必须从 03-主界面.png 中未激活搜索框裁剪（不可用 01-搜索框.png）
1280x720 分辨率下发送文本成功率 > 90%
1920x1080 分辨率下图像匹配命中率 > 95%（基于已有截图验证）
图像匹配失败时回退到几何定位
撤回操作在图像未命中时拒绝执行
woc_image_match_confidence metric 可查

阶段 3 验收：

SendTextFlow 状态机实现完成，9 个状态全部覆盖
5 分钟内重复发送相同内容（同 client_request_id）跳过率 100%
不同 client_request_id 的相同内容可正常发送
发送失败时返回 verified=False，不返回 SEND_FAILED
发错人时返回 DB_VERIFY_FAILED（不可重试）
失败后状态机回滚到 IDLE，下一次发送正常
Prometheus /metrics 端点可查
微信崩溃后 watchdog 10s 内感知

13.2 整体验收

单条文本消息端到端耗时 < 10 秒
发送成功率 > 95%（连续 100 次测试）
5 分钟内重复发送跳过率 100%
无子进程泄漏（压力测试）
无磁盘泄漏（7 天运行 /tmp/woc_debug < 100MB）
状态机日志清晰可读（状态、耗时、置信度、trace_id）
新增 UI 流程开发时间 < 2 小时

十四、附录

14.1 技术约束备忘

约束	来源	影响
微信主 UI 是 Qt 自绘	docker/Dockerfile:28	CDP/AT-SPI 不可用
WeChatAppEx 是 Chromium	docker/Dockerfile:41	CDP 理论可行（需 PoC）
不响应 X11 修饰键组合	项目记忆	必须用 `xdotool type` + 鼠标点击
主进程不吃命令行参数	docker/autostart:23-25	CDP 开启需 LD_PRELOAD
send_queue 最小 3000ms 间隔	config.py:128	单条消息下限 3 秒
DB 密钥在主进程 [heap] 段	项目记忆	key 提取只扫主进程堆
WeChat 4.x 消息分片存储	项目记忆	`Msg_<MD5(username)>` 表
send_queue 已串行但无锁无幂等	send_queue.py:117	需在 L5 补幂等
DB 校验已实现但错误处理不当	routes/send.py:265	需区分 verified vs success

14.2 不采用的方案与原因

方案	不采用原因
pyautogui	与 xdotool 底层相同（X11 XTest），不解决核心问题
AT-SPI / dogtail	Qt 自绘应用无 a11y 树，完全失效
Frida native hook	需逆向微信符号，工作量大，封号风险高
SikuliX	JVM 依赖重，OpenCV 纯 Python 已足够
transitions 库	状态数 < 10，自写更简单
逐级状态机回滚	UI 操作不可逆，直接重置到 IDLE 更可靠
Flow 内加 asyncio.Lock	与 send_queue 双重串行，语义混乱
幂等键用 MD5 截断 8 位	短消息碰撞风险，用完整 SHA256
持久化幂等缓存	引入 SQLite/Redis 复杂度，收益低

14.3 参考项目

项目	方向	借鉴点
wechat-dump-rs	DB 解密	手机号定位 key 思路
wechat-decrypt	DB 解析	4.0 表结构与消息内容解析
WeChatFerry	Hook 架构	Windows only，仅参考架构设计
OpenCV 官方教程	图像识别	模板匹配最佳实践
Playwright CDP	CDP 连接	`connect_over_cdp` 用法
prometheus-client	监控	Python Prometheus 客户端

14.4 术语表

术语	含义
Backend	执行后端，封装 xdotool/OpenCV/CDP
Locator	元素定位策略，图像/几何/CDP
Selector	声明式元素选择器，YAML 配置
ElementHandle	元素句柄，包含坐标与置信度
Action	原子动作，组合 Backend + Locator
Flow	业务流程，状态机编排多个 Action
Orchestrator	流程编排器，管理幂等与重试
Capability	对外业务接口
IdemCache	幂等缓存，5 分钟去重
CircuitBreaker	熔断器，连续失败后熔断
Watchdog	崩溃感知后台任务
ResourceReaper	资源回收后台任务
Profile	UI 元素配置文件
TraceID	贯穿单次请求的日志追踪 ID

十五、接口调用速度优化（v3.0 新增）

15.1 速度瓶颈定位

通过逐行代码分析（见 1.2.3 节），当前 send_text 端到端时序拆解：

当前流程（典型 16s / 最坏 26s）：
├─ 会话定位 _open_session_by_name      ~10s
│  ├─ find_wechat_window               ~50ms
│  ├─ _activate_window_fast            ~2s    ← 已用自适应轮询（100ms 步进）
│  ├─ sleep 1.0s × 7 处                7.0s   ← 【最大瓶颈】固定等待
│  ├─ _click_at × 2                    ~200ms ← 两次 fork 可合并
│  ├─ _key × 4                         ~200ms
│  ├─ _paste_via_xclip (name)          ~200ms ← 方法名误导，实际是 xdotool type
│  └─ _debug_screenshot × 1            ~100ms ← 写磁盘 I/O
├─ 发送体 _send_body                    ~3s
│  ├─ _debug_screenshot × 4            ~400ms ← 写磁盘 I/O
│  ├─ _click_input_box                 ~100ms
│  ├─ _paste_via_xclip (content)       ~200ms
│  ├─ sleep 1.0s × 2 处                2.0s   ← 【瓶颈】
│  └─ _click_send_button               ~100ms
├─ DB 校验 _verify_sent_to_talker      0-10s  ← 【瓶颈】轮询 0.5s 间隔
└─ send_queue 延时                     3.0s   ← 任务后强制 sleep

三大瓶颈（占总耗时 80%+）：

固定 sleep：10 处 × 1.0s = 10s（会话定位 7s + 发送体 2s + 队列 3s 含此延时）
DB 校验超时：10s 超时 + 0.5s 轮询间隔（实际 DB 写入 < 500ms）
无会话缓存：每次重新搜索会话（连续发给同一人也重走 10s）

15.2 优化策略总览

策略	节省耗时	实施成本	阶段
A. 固定 sleep → 自适应轮询	5-7s	中	阶段 3
B. 会话缓存（同联系人复用）	10s（后续）	中	阶段 3
C. DB 校验调优	7-9s	低	阶段 1
D. 调试截图移出关键路径	0.5s	极低	阶段 1
E. 合并 xdotool 命令	0.5-1s	低	阶段 1
F. 并行化 DB 查询与 UI 操作	1-2s	低	阶段 3
G. 队列延时自适应	2s（同联系人）	低	阶段 3

预期效果：

首条消息：16s → 3-5s（省 70%）
同联系人后续：16s → 1-2s（省 90%）
不同联系人切换：16s → 3-5s

15.3 策略 A：固定 sleep → 自适应轮询

问题：当前 10 处 sleep 1.0s 是"保守等待 UI 响应"，但 UI 实际响应通常 < 200ms。

方案：用条件轮询替代固定等待，条件满足即继续，超时兜底。

# bridge/ui/flows/send_text.py
async def _wait_for_condition(
    self, condition_fn, timeout: float = 3.0, interval: float = 0.15,
) -> bool:
    """轮询等待条件成立，满足即返回 True，超时返回 False。"""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if await condition_fn():
            return True
        await asyncio.sleep(interval)
    return False

async def _click_search_box(self, ctx: FlowContext) -> FlowState:
    await self.actions.click_element("search_box", ctx.win_geom)
    # 旧：await asyncio.sleep(1.0)
    # 新：轮询等待搜索框获得焦点（光标出现）
    await self._wait_for_condition(
        lambda: self._is_search_focused(ctx.win_geom),
        timeout=2.0, interval=0.15,
    )
    return FlowState.SEARCH_BOX_CLICKED

async def _type_query(self, ctx: FlowContext) -> FlowState:
    await self.actions.backend.key_press("BackSpace", repeat=30)
    await asyncio.sleep(0.2)  # BackSpace 无需长等
    await self.actions.backend.type_text(ctx.display_name)
    # 旧：await asyncio.sleep(1.0)
    # 新：轮询等待搜索结果出现（图像匹配结果列表）
    await self._wait_for_condition(
        lambda: self.actions.find_element("search_result_list", ctx.win_geom),
        timeout=3.0, interval=0.15,
    )
    return FlowState.QUERY_TYPED

条件检测方式：

搜索框聚焦：xdotool getwindowfocus + 图像匹配光标
搜索结果出现：图像匹配搜索结果列表模板
会话打开：图像匹配右侧聊天面板 / 输入框
输入完成：无需等待（xdotool type 同步返回）
发送完成：DB 校验（见策略 C）

效果：7s sleep → ~1-2s 自适应等待，节省 5-6s。

容错：条件未满足时不报错，用 timeout 兜底，记录到 metric。

15.4 策略 B：会话缓存（同联系人复用）

问题：连续发给同一联系人时，每次都重新搜索会话（10s），浪费 9.8s。

方案：维护"当前已打开会话"状态，匹配则跳过会话定位。

# bridge/ui/orchestrator.py
class SessionCache:
    """当前已打开会话的缓存。
    
    缓存 key = display_name，value = 打开时间戳。
    TTL = 30s（30s 内无操作则认为微信可能切走了，缓存失效）。
    """
    
    def __init__(self, ttl: float = 30.0):
        self._ttl = ttl
        self._current: Optional[tuple[str, float]] = None  # (display_name, opened_at)
    
    def get(self) -> Optional[str]:
        """返回当前已打开会话的 display_name，过期或无则 None。"""
        if self._current is None:
            return None
        name, ts = self._current
        if time.monotonic() - ts > self._ttl:
            self._current = None
            return None
        return name
    
    def set(self, display_name: str) -> None:
        self._current = (display_name, time.monotonic())
    
    def invalidate(self) -> None:
        """显式失效（如发送失败、状态机重置时）。"""
        self._current = None


class FlowOrchestrator:
    def __init__(self, ...):
        self.session_cache = SessionCache(ttl=30.0)
    
    async def send_text(self, to_wxid: str, content: str,
                       display_name: str = None, ...):
        # 幂等检查...
        
        async def _run():
            flow = SendTextFlow(self.actions, self.db, self.session_cache)
            ctx = FlowContext(...)
            return await flow.run(ctx)
        
        result = await self.send_queue.enqueue(_run)
        # 失败时失效会话缓存
        if not result.success:
            self.session_cache.invalidate()
        return result


# bridge/ui/flows/send_text.py
class SendTextFlow(Flow):
    def __init__(self, actions, db, session_cache: SessionCache):
        super().__init__(actions, db)
        self.session_cache = session_cache
    
    async def run(self, ctx: FlowContext) -> FlowResult:
        t_start = time.perf_counter()
        
        # 检查会话缓存
        cached_session = self.session_cache.get()
        session_cached = (cached_session == ctx.display_name)
        
        if session_cached:
            # 跳过会话定位，直接聚焦输入框
            logger.info("[flow] 会话缓存命中，跳过定位")
            state = FlowState.SESSION_OPENED
            # 仅做轻量焦点校验
            await self._verify_session_still_open(ctx)
        else:
            # 完整会话定位流程
            state = await self._activate(ctx)
            state = await self._click_search_box(ctx)
            state = await self._type_query(ctx)
            state = await self._open_session(ctx)
            self.session_cache.set(ctx.display_name)
        
        # 共同的发送流程
        state = await self._focus_input(ctx)
        state = await self._type_content(ctx)
        state = await self._send(ctx)
        state = await self._cleanup(ctx)
        # ...

缓存失效条件：

发送失败（状态机重置）
TTL 过期（30s 无操作）
显式 invalidate（如切换联系人、Esc 退会话）

注意：缓存命中后仍需轻量校验，确认会话还开着（图像匹配输入框），防止微信后台切走。

效果：同联系人后续消息从 16s → 3-4s（省 10s 会话定位 + 部分 sleep）。

15.5 策略 C：DB 校验调优

问题：当前 _verify_sent_to_talker（routes/send.py:107）：

超时 10s（实际 DB 写入 < 500ms）
轮询间隔 0.5s（太慢，第一次命中要等 0.5s）
失败后继续等到 10s（浪费时间）

方案：

# bridge/ui/flows/send_text.py
async def _verify_sent(self, ctx: FlowContext) -> bool:
    """DB 校验调优：超时 3s + 轮询 0.2s + 早退出。"""
    if not self.db or ctx.before_msg_id == 0:
        return await self._verify_by_screenshot(ctx)
    
    deadline = time.monotonic() + 3.0  # 从 10s 降到 3s
    while time.monotonic() < deadline:
        msgs = await self.db.get_messages(
            ctx.to_wxid, limit=1, after_local_id=ctx.before_msg_id,
        )
        if msgs:
            msg = msgs[0]
            if (msg.talker == ctx.to_wxid 
                and msg.content == ctx.content 
                and msg.is_sender):
                return True
            # talker 不匹配 = 发错人，立即返回 False（不继续等）
            logger.error("[verify] talker 不匹配: expected=%s actual=%s",
                         ctx.to_wxid, msg.talker)
            return False
        await asyncio.sleep(0.2)  # 从 0.5s 降到 0.2s
    return False

调优依据：

微信 DB 写入延迟实测 < 500ms（SQLCipher 同步写入）
轮询 0.2s → 典型命中只需 1 次轮询（0.2-0.4s）
超时 3s 覆盖 99% 场景（P99 < 1s）
talker 不匹配时立即退出，不浪费时间

效果：DB 校验从 0-10s → 0.2-3s（典型 0.4s）。

15.6 策略 D：调试截图移出关键路径

问题：当前 _DEBUG_SCREENSHOTS=True 在关键路径上写 5 张 PNG ≈ 500ms（含磁盘 I/O 阻塞）。

方案：

# bridge/ui/backends/xdotool.py
class XdotoolBackend:
    def __init__(self, ...):
        self._debug_shots_enabled = os.environ.get("WOC_UI_DEBUG_SHOTS", "false").lower() == "true"
        self._debug_queue: asyncio.Queue = asyncio.Queue()
        self._debug_task: Optional[asyncio.Task] = None
    
    async def _debug_screenshot_async(self, step: str, trace_id: str = "") -> None:
        """异步截图：不阻塞关键路径，丢到后台队列写磁盘。"""
        if not self._debug_shots_enabled:
            return
        # 只入队，不等待写盘
        await self._debug_queue.put((step, trace_id, time.time()))
    
    async def _debug_writer(self) -> None:
        """后台 worker：从队列取截图写磁盘。"""
        while True:
            step, trace_id, ts = await self._debug_queue.get()
            try:
                path = f"/tmp/woc_debug/{trace_id}_{step}_{int(ts*1000)}.png"
                await self._run(["scrot", path])
            except Exception as e:
                logger.warning("[backend] 调试截图失败: %s", e)

关键改进：

生产环境默认关闭（WOC_UI_DEBUG_SHOTS=false）
开启时用后台队列异步写盘，不阻塞关键路径
文件名带 trace_id，便于排障
配合 ResourceReaper 自动清理

效果：500ms → 0ms（生产）/ < 5ms（调试，入队开销）。

15.7 策略 E：合并 xdotool 命令

问题：当前 _click_at（xdotool_driver.py:399）分两条命令：

await self._run(["xdotool", "mousemove", "--sync", str(x), str(y)])  # fork 1
await self._run(["xdotool", "click", "1"])                            # fork 2

每次 fork+exec 约 30-50ms，两次 = 60-100ms。

方案：合并为单条命令：

# bridge/ui/backends/xdotool.py
async def click(self, x: int, y: int) -> None:
    async with self._ui_lock:
        # 合并 mousemove + click 为单条命令
        await self._run([
            "xdotool", "mousemove", "--sync", str(x), str(y),
            "click", "1",
        ])

xdotool 支持链式命令，单条命令内顺序执行。

效果：每次点击节省 30-50ms，整流程 ~5 次点击 = 节省 150-250ms。

15.8 策略 F：并行化 DB 查询与 UI 操作

问题：发送前查 before_msg_id（routes/send.py:90）与 UI 激活窗口串行，可并行。

方案：

# bridge/ui/flows/send_text.py
async def _activate_and_prepare(self, ctx: FlowContext) -> FlowState:
    """并行：激活窗口 + 查询发送前 DB 最新 msg_id。"""
    # 两个独立任务并行
    activate_task = asyncio.create_task(self._activate(ctx))
    db_task = asyncio.create_task(self._fetch_before_msg_id(ctx))
    
    await asyncio.gather(activate_task, db_task)
    return FlowState.WINDOW_ACTIVATED

async def _fetch_before_msg_id(self, ctx: FlowContext) -> None:
    """查询发送前的 DB 最新 msg_id（并行执行，不阻塞 UI）。"""
    if self.db:
        try:
            ctx.before_msg_id = await self.db.get_latest_msg_id(ctx.to_wxid)
        except Exception:
            ctx.before_msg_id = 0

效果：节省 DB 查询时间（~100-200ms），与窗口激活重叠。

15.9 策略 G：队列延时自适应

问题：当前 send_delay_ms=3000（config.py:128）对所有情况一刀切。

方案：同联系人后续消息延时更短，不同联系人保持原延时。

# bridge/ui/orchestrator.py
class FlowOrchestrator:
    async def send_text(self, ...):
        cached_session = self.session_cache.get()
        session_cached = (cached_session == display_name)
        
        async def _run():
            flow = SendTextFlow(self.actions, self.db, self.session_cache)
            ctx = FlowContext(...)
            return await flow.run(ctx)
        
        # 自适应延时
        if session_cached:
            delay = min(self.config.send_delay_ms, 1000)  # 同联系人 1s
        else:
            delay = self.config.send_delay_ms  # 不同联系人 3s
        
        result = await self.send_queue.enqueue(_run, delay_ms=delay)
        return result


# messaging/send_queue.py
class SendQueue:
    async def enqueue(self, coro_factory: CoroFactory, 
                      delay_ms: Optional[int] = None) -> Any:
        """入队，可选自定义延时（覆盖默认 send_delay_ms）。"""
        ...
    
    async def _run(self):
        while True:
            coro_factory, future, custom_delay = await self._queue.get()
            ...
            finally:
                self._queue.task_done()
                delay = custom_delay if custom_delay is not None else self.send_delay_ms
                await asyncio.sleep(delay / 1000.0)

理由：

同联系人：会话已打开，微信 UI 不需要长缓冲，1s 足够
不同联系人：需要切换会话，UI 需要时间稳定，保持 3s

效果：同联系人后续消息队列延时 3s → 1s。

15.10 优化后预期时序

首条消息（不同联系人）：

├─ 激活窗口 + DB 查询（并行）           ~2s
├─ 点击搜索框（自适应等待）            ~0.3s
├─ 输入搜索词（自适应等待结果）        ~0.5s
├─ 点击搜索结果（自适应等待会话打开）  ~0.5s
├─ 聚焦输入框                          ~0.3s
├─ 输入内容 + 点击发送                 ~0.5s
├─ DB 校验（3s 超时，典型 0.4s）      ~0.4s
├─ 清理（Esc）                         ~0.3s
└─ 队列延时                            ~3s
总计：~4.8s（从 16s 降至 4.8s，省 70%）

同联系人后续消息：

├─ 会话缓存命中（跳过定位）            ~0s
├─ 聚焦输入框                          ~0.3s
├─ 输入内容 + 点击发送                 ~0.5s
├─ DB 校验                             ~0.4s
├─ 清理                                ~0.3s
└─ 队列延时（自适应 1s）               ~1s
总计：~2.5s（从 16s 降至 2.5s，省 84%）

15.11 实施阶段对照

策略	阶段 1	阶段 2	阶段 3
A. 自适应轮询	-	-	✅
B. 会话缓存	-	-	✅
C. DB 校验调优	✅	✅	✅
D. 调试截图异步	✅	✅	✅
E. 合并 xdotool	✅	✅	✅
F. 并行 DB 查询	-	-	✅
G. 队列延时自适应	-	-	✅

阶段 1 即可落地 C/D/E（低成本，无需新架构），立即节省 ~8s（DB 校验 7s + 截图 0.5s + 合并命令 0.2s）。

15.12 速度监控指标

# bridge/ui/metrics.py
# 首条 vs 缓存命中耗时分布
send_duration_first = Histogram(
    "woc_send_duration_first_seconds", "First send (no cache)",
    buckets=[1, 2, 3, 5, 8, 12, 16, 25],
)
send_duration_cached = Histogram(
    "woc_send_duration_cached_seconds", "Cached session send",
    buckets=[0.5, 1, 1.5, 2, 3, 5],
)
# 会话缓存命中率
session_cache_hit = Counter(
    "woc_session_cache_total", "Session cache hit/miss",
    ["result"],  # hit / miss
)
# 自适应等待超时次数
adaptive_wait_timeout = Counter(
    "woc_adaptive_wait_timeout_total", "Adaptive wait timeout",
    ["step"],  # search_box / search_result / session_open
)

告警阈值：

woc_send_duration_first_seconds P95 > 8s → 排查自适应等待是否失效
woc_session_cache_hit{result="miss"} 占比 > 80% → 排查缓存 TTL
woc_adaptive_wait_timeout_total 增长率 > 10/min → 模板可能失效

15.13 速度与可靠性的平衡

核心原则：速度优化不能牺牲可靠性。

场景	速度优先	可靠性优先	决策
会话缓存命中后是否跳过焦点校验	是（省 0.3s）	否（防微信切走）	否，必须轻量校验
DB 校验超时设为多短	1s	3s	3s，覆盖 P99
自适应等待超时是否报错	否（容错继续）	是（严格）	否，timeout 兜底 + 记录 metric
队列延时同联系人设多短	500ms	1s	1s，给微信 UI 刷新时间
合并 xdotool 命令是否影响错误定位	是（单条失败难定位）	否（分步可定位）	合并，日志记录命令内容

关键约束：

自适应等待的 timeout 不小于 2s（防止 UI 偶发慢响应误判）
会话缓存 TTL 不超过 30s（防止微信自动切走后误用）
DB 校验 talker 不匹配时立即退出（不缩短超时，但早退）
队列延时不低于 1s（同联系人）/ 3s（不同联系人）

十六、附录：实施检查清单

16.1 阶段 1 检查清单（P0 高可用 + 速度初步优化）

_CMD_TIMEOUT_SEC 从 15s 降到 5s
_run_managed 上下文管理器实现，取消时 kill+wait
_step 的 max(1.0, remaining) 改为 max(7.0, remaining)
生产环境 WOC_UI_DEBUG_SHOTS=false
调试截图改为异步队列写盘
_click_at 合并 mousemove + click 为单条命令
DB 校验超时从 10s 降到 3s，轮询从 0.5s 降到 0.2s
DB 校验 talker 不匹配时立即退出
bridge 启动时执行 _full_cleanup_on_startup
现有 /api/send/text 接口行为不变
无子进程泄漏（压力测试）

16.2 阶段 3 检查清单（状态机 + 速度全面优化）

SendTextFlow 状态机实现，9 个状态覆盖
固定 sleep 改为自适应轮询（10 处）
会话缓存 SessionCache 实现（TTL 30s）
队列延时自适应（同联系人 1s / 不同 3s）
DB 查询与窗口激活并行
幂等缓存含 client_request_id
错误码分类（8 类）
熔断器（DB + 图像）
Watchdog 微信/Xvnc 崩溃感知
Prometheus metric 暴露
TraceID 贯穿日志
首条消息 P95 < 5s
同联系人后续 P95 < 3s

105 KiB Raw Permalink Blame History Unescape Escape

微信 UI 自动化通用架构优化方案

一、背景与现状

1.1 当前实现

1.2 已确认的核心问题（含高可用性复核新增）

1.2.1 功能正确性问题

1.2.2 高可用性问题（v2.0 新增复核）

1.2.3 速度问题（v3.0 新增复核）

1.3 微信 4.x Linux 技术约束（决定方案边界）

二、技术路径选型

2.1 候选方案对比

2.2 选型理由

2.3 最终技术栈

三、目标架构（六层分层 + 高可用横切）

3.1 分层职责

3.2 依赖方向

3.3 并发模型（关键决策）

3.4 多实例隔离（MVP 不涉及，但架构必须预留）

四、核心模块详细设计

4.1 Layer 1 - Backend Adapters

4.1.1 BackendProtocol 定义

4.1.2 XdotoolBackend 实现（含超时与取消安全）

4.1.3 OpenCVBackend（图像识别专用）

4.2 Layer 2 - Locator Strategies

4.3 Layer 3 - Action Library

4.4 Layer 4 - Flow State Machine

4.4.1 状态定义

4.4.2 Flow 基类（含全量清场 + post_verify）

4.4.3 SendTextFlow 实现

4.5 Layer 5 - Flow Orchestrator（含幂等 + 熔断 + 重试）

4.5.1 幂等缓存（含 client_request_id）

4.5.2 熔断器

4.5.3 FlowOrchestrator

4.6 Layer 6 - Capability API

五、横切高可用模块

5.1 错误码分类表

5.2 重试策略

5.3 Watchdog（崩溃感知）

5.4 ResourceReaper（资源回收）

5.5 Metrics（Prometheus）

5.6 TraceID

六、目录结构重构

七、最小可行实现（MVP）路径

阶段 0：PoC 验证（先验证假设）

PoC 1：OpenCV 模板匹配可行性

PoC 2：状态机回滚可行性

PoC 3（可选）：CDP 端口开启可行性

阶段 1：抽象层落地 + P0 高可用修复（行为不变）

阶段 2：引入图像校验

阶段 3：引入状态机 + DB 校验 + 幂等

阶段 4（可选）：CDP 后端 PoC

阶段 5：多应用桥接框架对齐

八、关键工程决策汇总

8.1 并发模型

8.2 超时层级（解决 L1/L2 冲突）

8.3 子进程安全

8.4 资源回收

8.5 熔断策略

8.6 幂等设计

8.7 状态机恢复

8.8 错误分类与重试

8.9 可观测性

8.10 模板维护

九、YAML Profile 规范

9.1 文件命名规则

9.2 截图素材参考

9.3 YAML 字段规范

9.4 加载优先级

十、迁移与兼容策略

10.1 不破坏现有功能

10.2 渐进式路由切换

10.3 回滚方案

十一、预期收益

十二、风险与应对

十三、验收标准

13.1 阶段验收

13.2 整体验收

十四、附录

14.1 技术约束备忘

14.2 不采用的方案与原因

105 KiB

Raw Permalink Blame History