feat(desktop-wake): add WeCom UIA warm-tree and ship n2-wake in 1.2.3

Expose ensure_ui_ready via DuiLib GetAccessible inject path; package x86 native helpers for host/skill consumers.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/jiangchang_desktop_wake/wake.py

@@ -0,0 +1,225 @@
+"""软唤醒策略梯 + native helper（N1+）。"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import time
+from pathlib import Path
+from typing import Callable, Optional, Tuple, Union
+
+from .errors import DesktopWakeError, ERR_WAKE_FAILED
+from .probe import probe
+from .profiles import load_profile
+from .types import Profile, WakeStepResult
+from . import uia_backend
+from .window import activate_hwnd, find_hwnd_for_profile, require_windows
+
+_logger = logging.getLogger("jiangchang_desktop_wake.wake")
+
+# strategy -> (detail, status)  status: ok | fail | skipped
+StrategyFn = Callable[[int, Profile], Tuple[str, str]]
+
+
+def _strategy_activate(hwnd: int, profile: Profile) -> Tuple[str, str]:
+    activate_hwnd(hwnd, maximize=True)
+    return "foreground+maximize", "ok"
+
+
+def _strategy_wm_getobject(hwnd: int, _profile: Profile) -> Tuple[str, str]:
+    return uia_backend.send_wm_getobject(hwnd), "ok"
+
+
+def _strategy_msaa(hwnd: int, _profile: Profile) -> Tuple[str, str]:
+    return uia_backend.msaa_from_window(hwnd), "ok"
+
+
+def _strategy_uia_walk(hwnd: int, _profile: Profile) -> Tuple[str, str]:
+    return uia_backend.deep_uia_walk(hwnd), "ok"
+
+
+def native_helper_path(name: str) -> Optional[Path]:
+    here = Path(__file__).resolve().parent / "native" / name
+    if here.is_file():
+        return here
+    env = os.environ.get("JIANGCHANG_DESKTOP_WAKE_NATIVE")
+    if env:
+        p = Path(env)
+        if p.is_file():
+            return p
+        cand = p / name
+        if cand.is_file():
+            return cand
+    app_root = os.environ.get("JIANGCHANG_APP_ROOT") or os.environ.get("JIANGCHANG_SKILLS_ROOT")
+    if app_root:
+        for sub in (
+            Path(app_root) / "resources" / "desktop-wake" / name,
+            Path(app_root) / "desktop-wake" / name,
+        ):
+            if sub.is_file():
+                return sub
+    return None
+
+
+def _strategy_native(hwnd: int, profile: Profile) -> Tuple[str, str]:
+    helper_name = profile.native_helper or "desktop-wake.exe"
+    path = native_helper_path(helper_name)
+    if path is None:
+        return "skipped:native_helper_missing", "skipped"
+    args = []
+    for a in profile.native_args or ["0x{hwnd_hex}"]:
+        args.append(
+            a.replace("{hwnd}", str(int(hwnd)))
+            # 必须带 0x：纯 hex 如 542144 会被 C# 当成十进制，导致 window_invalid
+            .replace("{hwnd_hex}", format(int(hwnd), "x"))
+            .replace("{profile}", profile.id)
+        )
+    cmd = [str(path), *args]
+    mode_n2_wake = "n2-wake" in args
+    # n2-wake：注入 + DuiLib 子类化，需更长超时；暖树由后续 probe 判定
+    timeout_s = 60 if mode_n2_wake else 25
+    _logger.info("native_helper cmd=%s timeout=%s", cmd, timeout_s)
+    try:
+        proc = subprocess.run(
+            cmd, capture_output=True, text=True, timeout=timeout_s, check=False
+        )
+    except subprocess.TimeoutExpired:
+        return "timeout", "fail"
+    except OSError as exc:
+        return f"spawn_error:{exc}", "fail"
+
+    if mode_n2_wake:
+        time.sleep(1.0)
+
+    out = (proc.stdout or "").strip()
+    err = (proc.stderr or "").strip()
+    detail = f"rc={proc.returncode} out={out[:300]!r} err={err[:120]!r}"
+
+    native_ok = False
+    if out:
+        # 取最后一行 JSON
+        line = out.splitlines()[-1].strip()
+        try:
+            payload = json.loads(line)
+            if mode_n2_wake:
+                # n2-wake：ok=注入成功且 UI 仍响应；是否暖树看外层 probe
+                native_ok = bool(payload.get("ok")) and bool(
+                    payload.get("inject_ok", payload.get("ok"))
+                )
+                if payload.get("ui_responsive") is False:
+                    native_ok = False
+            else:
+                # n1/auto：ok 表示已暖
+                native_ok = bool(payload.get("ok"))
+            detail = (
+                f"rc={proc.returncode} ok={native_ok} mode={'n2-wake' if mode_n2_wake else 'other'} "
+                f"inject_ok={payload.get('inject_ok')} ui_responsive={payload.get('ui_responsive')} "
+                f"walk={payload.get('walk_count')} findall={payload.get('findall_count')} "
+                f"markers={payload.get('markers_found')} detail={payload.get('detail')}"
+            )
+        except json.JSONDecodeError:
+            native_ok = proc.returncode == 0
+    else:
+        native_ok = proc.returncode == 0
+
+    return detail, ("ok" if native_ok else "fail")
+
+
+STRATEGIES: dict[str, StrategyFn] = {
+    "activate": _strategy_activate,
+    "wm_getobject": _strategy_wm_getobject,
+    "msaa_from_window": _strategy_msaa,
+    "uia_attach_walk": _strategy_uia_walk,
+    "native_helper": _strategy_native,
+}
+
+
+def wake(
+    profile: Union[str, Profile] = "wecom",
+    *,
+    hwnd: Optional[int] = None,
+    stop_when_warm: bool = True,
+    pause_s: float = 0.4,
+) -> list[WakeStepResult]:
+    """按 profile.wake_strategies 依次尝试唤醒。
+
+    - ``skipped``：策略未真正执行（如缺 exe），不算成功
+    - ``fail``：执行了但 helper/异常失败
+    - 每步后若 ``stop_when_warm`` 且 probe 已暖，则提前结束
+    """
+    require_windows()
+    prof = load_profile(profile) if isinstance(profile, str) else profile
+    h = find_hwnd_for_profile(prof, hwnd=hwnd)
+    steps: list[WakeStepResult] = []
+
+    for name in prof.wake_strategies:
+        fn = STRATEGIES.get(name)
+        if fn is None:
+            steps.append(
+                WakeStepResult(strategy=name, ok=False, status="fail", detail="unknown_strategy")
+            )
+            continue
+        try:
+            detail, status = fn(h, prof)
+            steps.append(
+                WakeStepResult(
+                    strategy=name,
+                    ok=(status == "ok"),
+                    status=status,
+                    detail=detail,
+                )
+            )
+            if status == "skipped":
+                continue
+        except Exception as exc:
+            _logger.debug("strategy %s failed: %s", name, exc)
+            steps.append(
+                WakeStepResult(strategy=name, ok=False, status="fail", detail=str(exc))
+            )
+            continue
+
+        time.sleep(pause_s)
+        if stop_when_warm:
+            try:
+                pr = probe(prof, hwnd=h)
+                if pr.warm:
+                    steps.append(
+                        WakeStepResult(
+                            strategy="probe_check",
+                            ok=True,
+                            status="ok",
+                            detail=f"warm after {name} findall={pr.findall_count}",
+                        )
+                    )
+                    break
+            except Exception as exc:
+                steps.append(
+                    WakeStepResult(
+                        strategy="probe_check",
+                        ok=False,
+                        status="fail",
+                        detail=str(exc),
+                    )
+                )
+
+    return steps
+
+
+def wake_or_raise(
+    profile: Union[str, Profile] = "wecom",
+    *,
+    hwnd: Optional[int] = None,
+) -> list[WakeStepResult]:
+    steps = wake(profile, hwnd=hwnd)
+    prof = load_profile(profile) if isinstance(profile, str) else profile
+    h = find_hwnd_for_profile(prof, hwnd=hwnd)
+    pr = probe(prof, hwnd=h)
+    if not pr.warm:
+        raise DesktopWakeError(
+            ERR_WAKE_FAILED,
+            f"唤醒后仍冷 findall={pr.findall_count} missing={pr.markers_missing} "
+            f"steps={[s.strategy + ':' + s.status for s in steps]}",
+        )
+    return steps