feat(screencast): 自动激活窗口 / 多显示器 / 字幕独立时长

基于 SDK 新模块（前两个 commit）的 screencast 工具能力升级：

- runner.py:
  · 新增 activate_window: bool = True 参数，录屏前自动调
    activate_and_maximize_jiangchang_window，避免开头几秒录到 PowerShell/桌面
  · 新增 monitor_index: Optional[int] = None 参数，透传给 ScreenRecorder
  · 自带 SDK 路径兜底 _ensure_sdk_on_path，让 record_screencast.py 直接 python
    跑时（不走 pytest conftest）也能 import jiangchang_desktop_sdk.window_win32

- recorder.py:
  · ScreenRecorder.__init__ 加 monitor_index 参数
  · _capture_loop 用 idx 选 monitor，越界兜底到 0
  · None = 旧行为（monitors[0] 所有屏合并），1 = 主屏，2+ = 第 N 屏
  · 多显示器场景推荐显式传 monitor_index=1 避免录出半边空白

- subtitle.py:
  · SubtitleEngine.process_line 支持二元组（兼容旧脚本）和三元组（每条独立 duration）
  · 长等待场景可以让某条字幕停留更久（如 30s）覆盖中间无日志的"字幕真空"

所有改动向后兼容，全部新增参数有默认值。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

src/jiangchang_desktop_sdk/client.py

@@ -219,63 +219,13 @@ class JiangchangDesktopClient:
         return False
 
     def _activate_and_maximize_windows(self) -> bool:
-        """Windows 专属：用 Win32 API 把主窗口拉到桌面最前并最大化。
+        """Windows 专属：把匠厂主窗口拉到桌面最前并最大化。
 
-        EnumWindows 遍历所有顶层窗口，按标题模糊匹配（含"匠厂/Jiangchang/ClawX/OpenClaw"），
+        实现已抽到 jiangchang_desktop_sdk.window_win32 模块（SDK 子模块），
-        找到后 ShowWindow(SW_MAXIMIZE) + SetForegroundWindow，让录屏看到铺满桌面的匠厂，
+        给 screencast 工具也复用。本方法保留为类内入口（兼容旧调用）。
-        避免任务栏边角和其他应用露出来。
-
-        为什么改用 ctypes（替代上一版 PowerShell AppActivate）：
-        1) AppActivate 只激活不最大化，依然能录到桌面边缘其他窗口；
-        2) ctypes 不 fork powershell，启动更快（~0ms vs ~300-800ms）；
-        3) 直接拿到 hwnd 后续可扩展（如监控窗口尺寸）。
         """
-        if os.name != "nt":
+        from .window_win32 import activate_and_maximize_jiangchang_window
-            return False
+        return activate_and_maximize_jiangchang_window()
-        try:
-            import ctypes
-            from ctypes import wintypes
-        except Exception as exc:
-            _logger.debug("[bring_to_front] ctypes 不可用：%s", exc)
-            return False
-
-        user32 = ctypes.windll.user32
-        SW_MAXIMIZE = 3
-        keywords = ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
-        matching: list = []
-
-        @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
-        def _enum_proc(hwnd, _lparam):
-            if not user32.IsWindowVisible(hwnd):
-                return True
-            length = user32.GetWindowTextLengthW(hwnd)
-            if length == 0:
-                return True
-            buf = ctypes.create_unicode_buffer(length + 1)
-            user32.GetWindowTextW(hwnd, buf, length + 1)
-            if any(kw in buf.value for kw in keywords):
-                matching.append((hwnd, buf.value))
-            return True
-
-        try:
-            user32.EnumWindows(_enum_proc, 0)
-        except Exception as exc:
-            _logger.debug("[bring_to_front] EnumWindows 失败：%s", exc)
-            return False
-
-        if not matching:
-            _logger.debug("[bring_to_front] 未找到匠厂主窗口")
-            return False
-
-        hwnd, title = matching[0]
-        try:
-            user32.ShowWindow(hwnd, SW_MAXIMIZE)
-            user32.SetForegroundWindow(hwnd)
-            _logger.debug("[bring_to_front] 已激活+最大化 hwnd=%s title=%s", hwnd, title)
-            return True
-        except Exception as exc:
-            _logger.debug("[bring_to_front] ShowWindow/SetForegroundWindow 失败：%s", exc)
-            return False
 
     def _launch_via_executable(self, cdp_port: int) -> bool:
         """回退：直接 spawn 可执行文件，带 --remote-debugging-port。"""

src/jiangchang_desktop_sdk/e2e_helpers.py

@@ -0,0 +1,261 @@
+# -*- coding: utf-8 -*-
+"""
+桌面 E2E 通用 helper：所有 skill 的 desktop E2E 测试共用。
+
+为什么独立出来：
+- disburse-payroll-icbc 实战验证：SDK 的 ask() 在 user_idx<0 时有死循环 bug，
+  且 send_file() 是 NotImplementedError。需要绕开它们自写完整测试流程
+- 这些 helper（拖拽附件、等 user 节点、多信号合流等待 agent 完成）是
+  **所有 skill 都通用的**，不应该每个 skill 都内嵌一份
+
+设计原则：
+- 所有 helper 只接受 Playwright Page 作为入口（不依赖 SDK client 实例）
+- 时间常量默认值留出宽容（user message 60s / agent complete 1200s）
+- 失败一律抛 TimeoutError 或 AssertionError，由调用方决定怎么截屏/上报
+
+依赖匠厂 DOM 锚点（docs/JIANGCHANG_CUSTOM_ANCHORS.md 守护）：
+- [data-testid="chat-composer-input"]
+- [data-testid="chat-page"][data-sending]
+- [data-role="user"] / [data-role="assistant"]
+- [data-testid="chat-execution-graph"][data-collapsed]
+- window.__jc_sending__
+"""
+from __future__ import annotations
+
+import base64
+import re
+import time
+from pathlib import Path
+from typing import Optional, Tuple
+
+from playwright.sync_api import Page
+
+
+__all__ = [
+    "drop_file_into_composer",
+    "wait_for_attachment_ready",
+    "wait_for_composer_enabled",
+    "send_prompt_via_composer",
+    "wait_for_user_message",
+    "wait_for_agent_complete",
+]
+
+
+def _normalize_ws(text: str) -> str:
+    return re.sub(r"\s+", "", text)
+
+
+# ─── 附件上传（绕开 SDK send_file 未实现） ──────────────────────────
+
+def drop_file_into_composer(page: Page, file_path: Path) -> None:
+    """通过 DragEvent + DataTransfer 把文件拖入聊天输入框。
+
+    匠厂 ChatInput 的 onDrop 挂在外层 div 上，但 React 17+ 事件代理到 root，
+    所以在 textarea 上派发 bubbles:true 的 drop 事件能正常触发 stageBufferFiles
+    → /api/files/stage-buffer → 真实附件链路。
+    """
+    payload_b64 = base64.b64encode(file_path.read_bytes()).decode("ascii")
+    page.evaluate(
+        """
+        ({ base64, fileName, mimeType }) => {
+          const binary = atob(base64);
+          const bytes = new Uint8Array(binary.length);
+          for (let i = 0; i < binary.length; i++) {
+            bytes[i] = binary.charCodeAt(i);
+          }
+          const file = new File([bytes], fileName, { type: mimeType });
+          const dt = new DataTransfer();
+          dt.items.add(file);
+          const textarea = document.querySelector('[data-testid="chat-composer-input"]');
+          if (!textarea) {
+            throw new Error('chat-composer-input not found');
+          }
+          const opts = { bubbles: true, cancelable: true, dataTransfer: dt };
+          textarea.dispatchEvent(new DragEvent('dragenter', opts));
+          textarea.dispatchEvent(new DragEvent('dragover', opts));
+          textarea.dispatchEvent(new DragEvent('drop', opts));
+        }
+        """,
+        {
+            "base64": payload_b64,
+            "fileName": file_path.name,
+            "mimeType": _guess_mime(file_path),
+        },
+    )
+
+
+def _guess_mime(file_path: Path) -> str:
+    suffix = file_path.suffix.lower()
+    return {
+        ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ".xls": "application/vnd.ms-excel",
+        ".csv": "text/csv",
+        ".pdf": "application/pdf",
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".txt": "text/plain",
+        ".json": "application/json",
+    }.get(suffix, "application/octet-stream")
+
+
+def wait_for_attachment_ready(page: Page, file_name: str, timeout_s: float = 30.0) -> None:
+    """stage-buffer 成功后附件 chip 会展示 fileName，轮询直到可见。"""
+    deadline = time.time() + timeout_s
+    name_loc = page.get_by_text(file_name, exact=False)
+    while time.time() < deadline:
+        try:
+            n = name_loc.count()
+            for i in range(n):
+                if name_loc.nth(i).is_visible():
+                    return
+        except Exception:
+            pass
+        time.sleep(0.5)
+    raise TimeoutError(
+        f"附件 {file_name!r} 在 {timeout_s}s 内未变为 ready"
+    )
+
+
+# ─── 发送 prompt（绕开 SDK ask 的 15s 锁死） ────────────────────────
+
+def wait_for_composer_enabled(page: Page, timeout_s: float = 15.0) -> None:
+    """等输入框可见且非 disabled（避免 canSend=false 静默 no-op）。"""
+    page.wait_for_selector(
+        '[data-testid="chat-composer-input"]',
+        state="visible",
+        timeout=int(timeout_s * 1000),
+    )
+    textarea = page.locator('[data-testid="chat-composer-input"]')
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        if not textarea.evaluate("(el) => el.disabled"):
+            return
+        time.sleep(0.3)
+    raise TimeoutError(
+        f"chat-composer-input 在 {timeout_s}s 内仍为 disabled（gateway 可能未就绪）"
+    )
+
+
+def send_prompt_via_composer(page: Page, prompt: str, *, delay_ms: int = 25) -> None:
+    """拟人输入 + Enter 发送。不做任何等待，返回即调用结束。"""
+    textarea = page.locator('[data-testid="chat-composer-input"]')
+    textarea.click()
+    textarea.fill("")
+    textarea.press_sequentially(prompt, delay=delay_ms)
+    textarea.press("Enter")
+
+
+def wait_for_user_message(page: Page, prompt: str, timeout_s: float = 60.0) -> int:
+    """等我们这条 user 消息进入 DOM，返回它在 [data-role] 列表里的 index。
+
+    绕开 SDK ask() 只给 15s 窗口的脆弱判据：用 prompt 文本反向匹配，
+    宽容到 60s。脏 session（new_task 后 gateway 载回历史）也能正确定位。
+    """
+    prompt_norm = _normalize_ws(prompt)
+    deadline = time.time() + timeout_s
+    items = page.locator('[data-role]')
+    while time.time() < deadline:
+        total = items.count()
+        for i in range(total - 1, -1, -1):
+            node = items.nth(i)
+            role = (node.get_attribute("data-role") or "").lower()
+            if role != "user":
+                continue
+            text = node.inner_text(timeout=2000) or ""
+            text_norm = _normalize_ws(text)
+            if prompt_norm in text_norm or text_norm in prompt_norm:
+                return i
+        time.sleep(0.5)
+    raise TimeoutError(
+        f"在 {timeout_s}s 内未找到匹配 user 消息（prompt 归一化前 80 字="
+        f"{prompt_norm[:80]!r})"
+    )
+
+
+# ─── 等 agent 完成（多信号合流） ────────────────────────────────
+
+def _chat_sending_false(page: Page) -> bool:
+    val = page.locator('[data-testid="chat-page"]').first.get_attribute("data-sending")
+    return val is not None and val.lower() == "false"
+
+
+def _window_sending_not_true(page: Page) -> bool:
+    val = page.evaluate("() => window.__jc_sending__")
+    return val is not True
+
+
+def _no_expanded_execution_graph(page: Page) -> bool:
+    return (
+        page.locator('[data-testid="chat-execution-graph"][data-collapsed="false"]').count()
+        == 0
+    )
+
+
+def _latest_assistant_after(page: Page, user_idx: int) -> Optional[Tuple[int, str, bool]]:
+    items = page.locator('[data-role]')
+    total = items.count()
+    for i in range(total - 1, user_idx, -1):
+        node = items.nth(i)
+        role = (node.get_attribute("data-role") or "").lower()
+        if role != "assistant":
+            continue
+        body = node.inner_text(timeout=2000) or ""
+        streaming = (node.get_attribute("data-streaming") or "false").lower() == "true"
+        return i, body, streaming
+    return None
+
+
+def wait_for_agent_complete(
+    page: Page,
+    user_idx: int,
+    *,
+    overall_timeout_s: float = 1200.0,
+    stable_seconds: float = 5.0,
+    poll_interval: float = 1.0,
+) -> str:
+    """多信号合流等本轮 agent 完成，返回最末 assistant 气泡文本。
+
+    完成条件（全部满足并稳定 stable_seconds 秒）：
+    1) chat-page data-sending="false"
+    2) window.__jc_sending__ 不为 True
+    3) 不存在未折叠的 execution graph
+    4) user_idx 之后存在 assistant 节点，有非空 body，data-streaming != "true"
+    5) 上面 assistant 节点的文本长度连续 stable_seconds 秒无变化
+    """
+    deadline = time.time() + overall_timeout_s
+    last_body_len = -1
+    last_change_at = time.time()
+
+    while time.time() < deadline:
+        chat_ok = _chat_sending_false(page)
+        win_ok = _window_sending_not_true(page)
+        graph_ok = _no_expanded_execution_graph(page)
+        picked = _latest_assistant_after(page, user_idx)
+
+        body_text = ""
+        streaming = True
+        has_body = False
+        if picked is not None:
+            _, body_text, streaming = picked
+            has_body = len(body_text.strip()) > 0
+
+        all_ready = chat_ok and win_ok and graph_ok and has_body and not streaming
+
+        if not all_ready:
+            last_body_len = -1
+            last_change_at = time.time()
+        else:
+            cur_len = len(body_text)
+            if cur_len != last_body_len:
+                last_body_len = cur_len
+                last_change_at = time.time()
+            elif time.time() - last_change_at >= stable_seconds:
+                return body_text
+
+        time.sleep(poll_interval)
+
+    raise TimeoutError(
+        f"agent 在 {overall_timeout_s}s 内未完成（user_idx={user_idx} "
+        f"last_body_len={last_body_len}）"
+    )

src/jiangchang_desktop_sdk/window_win32.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""
+Win32 窗口工具：跨 SDK / screencast 共享的"激活并最大化匠厂主窗口"实现。
+
+为什么单独抽出来：
+- SDK 的 bring_to_front 需要 OS 级窗口前置（page.bring_to_front 只切 tab）
+- screencast 的 run_screencast 录屏前也要把匠厂浮到前面并铺满屏幕
+- 两者用同一份 EnumWindows + ShowWindow(SW_MAXIMIZE) + SetForegroundWindow 实现，
+  避免代码重复维护
+
+跨平台行为：
+- Windows: 真正执行
+- 其他系统: 直接返回 False（调用方按需 fallback 到协议激活）
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Optional, Tuple
+
+_logger = logging.getLogger("jiangchang_desktop_sdk.window_win32")
+
+_DEFAULT_KEYWORDS: Tuple[str, ...] = ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
+SW_MAXIMIZE = 3
+
+
+def activate_and_maximize_jiangchang_window(
+    keywords: Optional[Tuple[str, ...]] = None,
+) -> bool:
+    """Windows: EnumWindows 找匠厂主窗口 → ShowWindow(SW_MAXIMIZE) + SetForegroundWindow。
+
+    Args:
+        keywords: 标题模糊匹配的关键词；不传则用默认
+                  ("匠厂", "Jiangchang", "ClawX", "OpenClaw")
+
+    Returns:
+        True: 找到窗口并执行了 ShowWindow + SetForegroundWindow
+        False: 非 Windows / ctypes 不可用 / 找不到匹配窗口 / 任何异常
+    """
+    if os.name != "nt":
+        return False
+
+    try:
+        import ctypes
+        from ctypes import wintypes
+    except Exception as exc:
+        _logger.debug("ctypes 不可用：%s", exc)
+        return False
+
+    user32 = ctypes.windll.user32
+    kw = keywords or _DEFAULT_KEYWORDS
+    matching: list = []
+
+    @ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)
+    def _enum_proc(hwnd, _lparam):
+        if not user32.IsWindowVisible(hwnd):
+            return True
+        length = user32.GetWindowTextLengthW(hwnd)
+        if length == 0:
+            return True
+        buf = ctypes.create_unicode_buffer(length + 1)
+        user32.GetWindowTextW(hwnd, buf, length + 1)
+        if any(k in buf.value for k in kw):
+            matching.append((hwnd, buf.value))
+        return True
+
+    try:
+        user32.EnumWindows(_enum_proc, 0)
+    except Exception as exc:
+        _logger.debug("EnumWindows 失败：%s", exc)
+        return False
+
+    if not matching:
+        _logger.debug("未找到匠厂主窗口（keywords=%s）", kw)
+        return False
+
+    hwnd, title = matching[0]
+    try:
+        user32.ShowWindow(hwnd, SW_MAXIMIZE)
+        user32.SetForegroundWindow(hwnd)
+        _logger.debug("已激活+最大化 hwnd=%s title=%s", hwnd, title)
+        return True
+    except Exception as exc:
+        _logger.debug("ShowWindow/SetForegroundWindow 失败：%s", exc)
+        return False

tools/screencast/recorder.py

@@ -4,15 +4,27 @@ from __future__ import annotations
 import threading
 import time
 from pathlib import Path
+from typing import Optional
 
 import mss
 import mss.tools
 
 
 class ScreenRecorder:
-    def __init__(self, frames_dir: str, fps: int = 10):
+    def __init__(
+        self,
+        frames_dir: str,
+        fps: int = 10,
+        monitor_index: Optional[int] = None,
+    ):
+        """
+        Args:
+            monitor_index: mss 显示器索引；None=monitors[0]（所有屏合并区域，
+                          旧行为），1=主屏，2+=第 N 屏。多显示器场景推荐显式传 1。
+        """
         self._frames_dir = Path(frames_dir)
         self._fps = fps
+        self._monitor_index = monitor_index
         self._stop_event = threading.Event()
         self._thread: threading.Thread | None = None
         self._frame_count = 0
@@ -36,7 +48,10 @@ class ScreenRecorder:
     def _capture_loop(self) -> None:
         interval = 1.0 / self._fps
         with mss.mss() as sct:
-            monitor = sct.monitors[0]  # 全屏（monitor[0] 是所有显示器合并区域）
+            idx = self._monitor_index if self._monitor_index is not None else 0
+            if idx < 0 or idx >= len(sct.monitors):
+                idx = 0  # 越界兜底
+            monitor = sct.monitors[idx]
             while not self._stop_event.is_set():
                 t0 = time.perf_counter()
                 img = sct.grab(monitor)

tools/screencast/runner.py

@@ -10,6 +10,24 @@ from datetime import datetime
 from pathlib import Path
 from typing import List, Optional, Tuple
 
+
+def _ensure_sdk_on_path() -> None:
+    """record_screencast.py 直接 python 跑时不走 conftest，SDK 可能不在 sys.path。
+    自带兜底：从 tools/screencast/runner.py 上溯到 platform-kit/src。"""
+    import importlib.util
+    if importlib.util.find_spec("jiangchang_desktop_sdk") is not None:
+        return
+    # tools/screencast/runner.py → tools/screencast → tools → platform-kit
+    here = Path(__file__).resolve()
+    src = here.parent.parent.parent / "src"
+    if (src / "jiangchang_desktop_sdk").is_dir():
+        import sys as _sys
+        _sys.path.insert(0, str(src))
+
+
+_ensure_sdk_on_path()
+from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window  # noqa: E402
+
 from .composer import compose_video
 from .recorder import ScreenRecorder
 from .subtitle import SubtitleEngine
@@ -23,6 +41,8 @@ def run_screencast(
     media_assets_root: Optional[str] = None,
     music_subdir: str = "music",
     fps: int = 10,
+    activate_window: bool = True,
+    monitor_index: Optional[int] = None,
 ) -> str:
     """
     录制一次技能桌面 E2E 演示视频。
@@ -37,6 +57,10 @@ def run_screencast(
                            再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
         music_subdir:     music_assets_root 下音乐子目录，默认 "music"
         fps:               截帧帧率，默认 10
+        activate_window:   录屏开始前是否自动激活+最大化匠厂窗口（默认 True；
+                          Windows 走 ctypes，非 Windows 走 jiangchang:// 协议）
+        monitor_index:     指定录哪个显示器；None=所有显示器合并（旧行为），
+                          1=主屏，2+=第 N 屏（推荐多显示器场景显式传 1）
 
     Returns:
         输出 MP4 的绝对路径
@@ -59,7 +83,15 @@ def run_screencast(
     engine = SubtitleEngine(subtitle_script)
 
     with tempfile.TemporaryDirectory(prefix="screencast_frames_") as frames_dir:
-        recorder = ScreenRecorder(frames_dir, fps=fps)
+        recorder = ScreenRecorder(frames_dir, fps=fps, monitor_index=monitor_index)
+
+        if activate_window:
+            ok = activate_and_maximize_jiangchang_window()
+            if ok:
+                print("[screencast] 已激活+最大化匠厂窗口")
+                time.sleep(2)  # 等窗口浮起稳定
+            else:
+                print("[screencast] 未能激活匠厂窗口，继续录制（可能录到桌面边角）")
 
         print(f"[screencast] 启动录制 → {output_mp4.name}")
         recorder.start()

tools/screencast/subtitle.py

@@ -15,9 +15,15 @@ class _Entry:
 
 
 class SubtitleEngine:
-    def __init__(self, script: List[Tuple[str, str]], default_duration: float = 5.0):
+    def __init__(
+        self,
+        script: List[Tuple],
+        default_duration: float = 5.0,
+    ):
         """
-        script: [(关键词, 字幕文案), ...]
+        script: [(关键词, 字幕文案) | (关键词, 字幕文案, 显示时长秒), ...]
+                - 二元组：用 default_duration（向后兼容）
+                - 三元组：本条字幕独立指定显示秒数（用于长等待场景延长字幕停留）
         关键词大小写不敏感，每个关键词只触发一次。
         """
         self._script = script
@@ -31,12 +37,19 @@ class SubtitleEngine:
 
     def process_line(self, line: str) -> None:
         lower = line.lower()
-        for keyword, text in self._script:
+        for entry_tuple in self._script:
+            if len(entry_tuple) == 2:
+                keyword, text = entry_tuple
+                duration = self._default_duration
+            elif len(entry_tuple) == 3:
+                keyword, text, duration = entry_tuple
+            else:
+                continue
             if keyword in self._matched:
                 continue
             if keyword.lower() in lower:
                 elapsed = time.time() - self._t0
-                self._entries.append(_Entry(start=elapsed, text=text, duration=self._default_duration))
+                self._entries.append(_Entry(start=elapsed, text=text, duration=float(duration)))
                 self._matched.add(keyword)
 
     def generate_srt(self, output_path: str) -> None: