chore: sync ffmpeg RpaVideoSession, remove Playwright recording

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

.env.example

@@ -0,0 +1,21 @@
+# ── 运行模式（对应 adapter 四档，见 references/ADAPTER.md）──
+OPENCLAW_TEST_TARGET=simulator_rpa     # mock | simulator_rpa | real_api | real_rpa
+
+# ── 目标系统 ──
+TARGET_BASE_URL=https://sandbox.jc2009.com   # 仿真/生产地址，可被进程环境变量覆盖
+
+# ── 默认账号（仅非敏感标识；密码走 account-manager）──
+DEFAULT_LOGIN_ID=04110001
+
+# ── 浏览器 / RPA ──
+OPENCLAW_BROWSER_HEADLESS=0            # 0=有头(默认) 1=无头(CI)
+OPENCLAW_PLAYWRIGHT_STEALTH=1          # 1=开启反检测指纹(默认)
+
+# ── 存证 ──
+OPENCLAW_RECORD_VIDEO=1
+OPENCLAW_ARTIFACTS_ON_FAILURE=1        # 1=失败截图(默认)
+
+# ── 节流 / 超时 ──
+STEP_DELAY_MIN=1.0                     # 步骤间随机等待下限(秒)
+STEP_DELAY_MAX=5.0                     # 步骤间随机等待上限(秒)
+HUMAN_WAIT_TIMEOUT=180                 # 滑块/验证码/2FA 等人工超时(秒)

.gitignore

@@ -1,3 +1,4 @@
 __pycache__/
 *.py[cod]
 .env
+*.env.local

README.md

@@ -36,6 +36,7 @@
 | `scripts/` | 规范分层后的代码骨架 |
 | `tests/` | 单元测试与分层测试范式（默认根目录 unittest） |
 | `evals/` | 人工/半自动评估材料 |
+| `requirements.txt` | 本技能 Python 三方依赖声明（由匠厂宿主安装到共享 runtime） |
 | `.github/workflows/release_skill.yaml` | 标准发布工作流 |
 | `release.ps1` | 对齐现有 skill 的发布脚本入口 |
 
@@ -52,3 +53,12 @@ python scripts/main.py version
 - 新技能若不需要某些目录，也建议先保留结构，再按实际业务填充内容。
 - `metadata.openclaw.developer_ids` 是发布元数据，不是用户展示文案；用于非公开技能的开发者默认可见授权。
 - **发起 `release.ps1` 之前**，务必确认 `python tests/run_tests.py -v` 已全部通过。
+
+## Python 依赖（requirements.txt）
+
+- 在技能**根目录**维护 `requirements.txt`，声明本技能特有的 Python 三方包。
+- 匠厂宿主在**安装或更新**技能后，会将该文件中的依赖安装到共享 Python runtime：
+  `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`（Windows 示例：`D:\jiangchang-data\python-runtime\.venv`）。
+- 平台通用依赖由宿主 `resources/jiangchang-python-runtime/pyproject.toml` 维护；**不要**把业务依赖写进宿主 pyproject。
+- **Skill 业务代码不得自动 `pip install`**；缺依赖时应在 `health` / preflight 中给出清晰错误。
+- **系统级组件**（如 Microsoft Visual C++ Redistributable）**不要**写进 `requirements.txt`；仅在 `health` / preflight 中提示用户自行安装。

SKILL.md

@@ -38,6 +38,12 @@ python {baseDir}/scripts/main.py health
 python {baseDir}/scripts/main.py version
 ```
 
+## 运行依赖
+
+- Python 运行环境由匠厂宿主注入共享 runtime：`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。
+- 本技能**特有**的 Python 三方依赖请声明在根目录 `requirements.txt`；宿主安装/更新技能后会自动安装到共享 venv。
+- Skill 代码**不要**自行 `pip install`；系统级依赖（如 VC++ Runtime）仅在 `health` / preflight 中提示用户安装。
+
 ## 重要说明
 
 - 复制后请同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION`。

references/ADAPTER.md

@@ -0,0 +1,115 @@
+# 适配器标准：真实/仿真 × API/RPA 四档模式
+
+> 凡是"连接三方系统"（ERP、CRM、SaaS、银行等）的 skill，都应采用本文的 **adapter 四档模式**。这样同一套业务逻辑可以在不同档位间切换：开发用仿真、CI 用 mock、上线用真实，互不影响。
+
+## 为什么要分档
+
+对接一个外部系统，工程上无非四种方式的组合：
+
+| | API（有接口） | RPA（操作界面） |
+|---|---|---|
+| **真实系统** | `real_api` 真实 API | `real_rpa` 真实界面 |
+| **仿真系统** | `mock` 离线仿真 | `sim_rpa` 仿真平台界面 |
+
+- **mock**：纯离线、不联网，给单测 / CI / 开发自测，**保证可重复**。
+- **sim_rpa**：操作仿真平台（如 `sandbox.jc2009.com`），跑真实端到端流程但不碰生产，**默认开发档位**。
+- **real_api**：有官方接口时**首选**（最稳、最快、最易维护）。
+- **real_rpa**：没有 API 只能操作生产界面，**风险最高、放最后**。
+
+> 推荐优先级：**real_api > sim_rpa > real_rpa**，mock 永远保留做 CI。
+
+## 目录骨架
+
+业务逻辑只依赖 `base` 里定义的接口与数据契约，具体档位实现各自一个文件：
+
+```
+scripts/service/<domain>_adapter/
+  __init__.py        # dispatch：按档位返回对应 adapter 实例
+  base.py            # 数据契约（dataclass）+ AdapterBase 接口（NotImplementedError）
+  mock.py            # 离线仿真，给 CI/单测
+  real_api.py        # 真实系统 API
+  sim_rpa.py         # 仿真平台 RPA（Playwright/pywinauto/uiautomator2）
+  real_rpa.py        # 真实系统 RPA（占位，谨慎实现）
+```
+
+> 参考实现：`disburse-payroll-icbc/scripts/service/disburse_adapter/`（已有 `base.py` / `mock.py` / `icbc_simulator_rpa.py`）。
+> 模板示例：`scripts/service/example_adapter/`（复制改名即用）。
+
+### 引用方式
+
+```python
+from service.example_adapter import get_adapter
+from service.example_adapter.base import ExampleItem
+
+adapter = get_adapter()  # 由 OPENCLAW_TEST_TARGET 决定档位
+result = adapter.do_batch(
+    target="https://sandbox.jc2009.com",
+    items=[ExampleItem(id="1", label="demo")],
+)
+```
+
+mock 档位纯离线验证：
+
+```bash
+set OPENCLAW_TEST_TARGET=mock
+python -c "import sys; sys.path.insert(0,'scripts'); from service.example_adapter import get_adapter; from service.example_adapter.base import ExampleItem; r=get_adapter().do_batch('x',[ExampleItem('1','a')]); print(r.ok, r.serial_no)"
+```
+
+## base.py 写法（要点）
+
+1. **数据契约用 `@dataclass`**：输入项（如 `PayrollItem`）、批量结果（如 `BatchResult{ok, serial_no, error_msg, artifacts}`）。
+2. **接口方法抛 `NotImplementedError`**：所有档位实现同一签名。
+3. **以"批"为单位返回**：真实 RPA 往往是整批提交，结果按批次返回，不要假设逐条。
+4. **结果里带 `artifacts`**：截图/录屏/日志路径，便于排查。
+
+```python
+@dataclass
+class XxxResult:
+    ok: bool
+    serial_no: Optional[str]
+    error_msg: Optional[str]
+    artifacts: Dict[str, Any] = field(default_factory=dict)
+
+class AdapterBase:
+    name = "base"
+    def do_batch(self, target: str, items: list) -> XxxResult:
+        raise NotImplementedError
+```
+
+## 档位 dispatch
+
+由 `OPENCLAW_TEST_TARGET` 统一决定用哪个 adapter（与现有测试体系一致）：
+
+| `OPENCLAW_TEST_TARGET` | adapter | 用途 |
+|---|---|---|
+| `unit` / `mock` | `MockAdapter` | 单测 / CI，离线 |
+| `simulator_rpa`（默认） | `SimRpaAdapter` | 开发/演示，操作仿真平台 |
+| `real_api` | `RealApiAdapter` | 生产，走官方接口 |
+| `real_rpa` | `RealRpaAdapter` | 生产，操作真实界面（最后手段） |
+
+```python
+# __init__.py
+def get_adapter():
+    target = (os.environ.get("OPENCLAW_TEST_TARGET") or "simulator_rpa").lower()
+    if target in ("unit", "mock"):
+        return MockAdapter()
+    if target == "real_api":
+        return RealApiAdapter()
+    if target == "real_rpa":
+        return RealRpaAdapter()
+    return SimRpaAdapter()   # 默认
+```
+
+> 实际可改为读 `.env` 里的运行模式（见 `CONFIG.md`），`OPENCLAW_TEST_TARGET` 作为进程级覆盖优先。
+
+## 测试要求
+
+- **mock 档必须能纯离线跑通**，进 CI。
+- **每个 adapter 单测同一组用例**（契约测试），保证档位间行为一致。
+- RPA 档的真实联网测试单独标记，不进默认 CI（参考 `tests/desktop/`）。
+
+## 相关文档
+
+- `RPA.md` — 三端 RPA 技术选型与拟人/反反爬范式
+- `CONFIG.md` — `.env` 里如何配置运行模式与目标地址
+- `TESTING.md` — 测试 target 与隔离体系

references/CONFIG.md

@@ -0,0 +1,99 @@
+# 配置标准：`.env` 规范与首次落盘机制
+
+> 解决"每个 skill 每次都要手动搞一堆环境变量、账号、地址"的痛点。统一约定：**模板内置 `.env.example`，skill 首次运行自动落盘到用户数据目录，之后从那里读。**
+
+## 核心原则
+
+1. **`.env.example` 进仓库**：带注释的全量默认配置，是配置项的"单一事实来源"。
+2. **首次运行自动 copy**：`setup` / `doctor` / 首次 `run` 时，把 `.env.example` 复制到
+   `{CLAW_DATA_ROOT}/{user}/{slug}/.env`（**已存在则不覆盖**，保护用户改过的值）。
+3. **三级优先级读取**：`进程环境变量` > `数据目录/.env` > `.env.example 默认值`。
+4. **敏感信息绝不进 `.env`**：密码/密钥/口令走 account-manager 的 `secret-ref`（见下"红线"）。
+
+## 标准配置项
+
+每个 skill 的 `.env.example` 至少包含这些（按需增减）：
+
+```ini
+# ── 运行模式（对应 adapter 四档，见 ADAPTER.md）──
+OPENCLAW_TEST_TARGET=simulator_rpa     # mock | simulator_rpa | real_api | real_rpa
+
+# ── 目标系统 ──
+TARGET_BASE_URL=https://sandbox.jc2009.com   # 仿真/生产地址，可被进程环境变量覆盖
+
+# ── 默认账号（仅非敏感标识；密码走 account-manager）──
+DEFAULT_LOGIN_ID=04110001
+
+# ── 浏览器 / RPA ──
+OPENCLAW_BROWSER_HEADLESS=0            # 0=有头(默认) 1=无头(CI)
+OPENCLAW_PLAYWRIGHT_STEALTH=1          # 1=开启反检测指纹(默认)
+
+# ── 存证 ──
+OPENCLAW_RECORD_VIDEO=1                # 1=录屏+字幕+背景音+MP4成片（RPA技能默认开，见 RPA.md）
+OPENCLAW_ARTIFACTS_ON_FAILURE=1        # 1=失败截图(默认)
+
+# ── 节流 / 超时 ──
+STEP_DELAY_MIN=1.0                     # 步骤间随机等待下限(秒)
+STEP_DELAY_MAX=5.0                     # 步骤间随机等待上限(秒)
+HUMAN_WAIT_TIMEOUT=180                 # 滑块/验证码/2FA 等人工超时(秒)
+```
+
+## 🚫 红线：敏感信息不进 `.env`
+
+`.env` 只放**非敏感运行参数**（模式、地址、开关、超时）。密码、密钥、动态口令、token——
+
+- 走 **account-manager** 注册凭据，用 `--secret-storage env --secret-ref XXX` 引用；
+- 真实值由宿主/用户通过环境变量注入，**不落任何文件、不进 git**。
+- 参考 `disburse-payroll-icbc/scripts/cli/cmd_setup.py`：账号写入 account-manager，密码只登记 `secret-ref`（如 `ICBC_SIM_PASSWORD`）。
+
+`.gitignore` 必须包含：
+
+```gitignore
+.env
+*.env.local
+```
+
+## 配置读取层（模板提供）
+
+模板会提供一个统一 `config.py`，约定行为：
+
+```python
+config.get("TARGET_BASE_URL")              # 三级优先级解析
+config.get_bool("OPENCLAW_BROWSER_HEADLESS")
+config.get_float("STEP_DELAY_MIN", 1.0)
+config.ensure_env_file()                   # 首次把 .env.example copy 到数据目录
+```
+
+- 启动时调一次 `ensure_env_file()` 完成落盘。
+- 所有业务代码**只通过 `config.get*` 读配置**，不直接 `os.environ`，便于统一管理与测试覆盖。
+
+### 引用方式
+
+实现位于 `jiangchang-platform-kit` 的 `jiangchang_skill_core.config`；模板 vendor 拷贝在 `scripts/jiangchang_skill_core/config.py`：
+
+```python
+from jiangchang_skill_core import config
+from util.constants import SKILL_SLUG
+from util.runtime_paths import get_skill_root
+import os
+
+example = os.path.join(get_skill_root(), ".env.example")
+config.ensure_env_file(SKILL_SLUG, example)
+
+headless = config.get_bool("OPENCLAW_BROWSER_HEADLESS")
+target = config.get("TARGET_BASE_URL")
+timeout = config.get_int("HUMAN_WAIT_TIMEOUT", 180)
+```
+
+## doctor / setup 命令
+
+每个 skill 应提供自检命令（参考 `disburse` 的 `test_doctor`）：
+
+- `python scripts/main.py doctor`：检查浏览器是否安装、Profile/账号是否就绪、`.env` 是否已落盘、设备(手机)是否连接。
+- `python scripts/main.py setup`：初始化演示账号、首次落盘 `.env`。
+
+## 相关文档
+
+- `RPA.md` — 三端 RPA 标准与各开关含义
+- `ADAPTER.md` — `OPENCLAW_TEST_TARGET` 四档模式
+- `RUNTIME.md` — `CLAW_*` 环境变量与数据目录约定

references/DEVELOPMENT.md

@@ -103,6 +103,7 @@ your-skill/
 ├─ tests/
 ├─ .gitignore
 ├─ README.md
+├─ requirements.txt
 ├─ release.ps1
 └─ SKILL.md
 ```
@@ -151,9 +152,42 @@ scripts/
   作用：常量、日志、路径、时间工具、通用帮助函数
 
 - `jiangchang_skill_core/`
-  作用：运行时环境与统一日志副本
+  作用：运行时环境、统一日志、配置读取、RPA 共享库（vendor 自 platform-kit）
   一般按现有规范技能复制，不要自己乱改结构
 
+## 3.2 开发 RPA 类 skill
+
+若 skill 需要浏览器/桌面/手机自动化，按以下顺序落地：
+
+1. **先读三份标准**：`references/RPA.md`（三端范式与反反爬）、`references/CONFIG.md`（`.env` 落盘与读取）、`references/ADAPTER.md`（四档 adapter）。
+2. **从模板复制骨架**：
+   - `scripts/service/example_adapter/` → 改名为 `scripts/service/<domain>_adapter/`
+   - 参考 `scripts/service/example_browser_rpa.py` 写浏览器会话逻辑
+3. **只用共享库，不在 skill 里重写反反爬**：
+   ```python
+   from jiangchang_skill_core import config
+   from jiangchang_skill_core.rpa import launch_persistent_browser, anti_detect, wait_for_captcha_pass
+   ```
+   vendor 源码在 `scripts/jiangchang_skill_core/rpa/`（与 `jiangchang-platform-kit` 保持同步拷贝）。
+4. **mock 档必须离线可跑**（`OPENCLAW_TEST_TARGET=mock`）；sim_rpa / real_* 按需单独测。
+5. **桌面/手机**：本期标准见 RPA.md 第 2/3 节，复用 `jiangchang_desktop_sdk` / `screencast`，**不要在新 skill 里重复造包**（尚待实战验证）。
+
+---
+
+### 3.1 `requirements.txt` 依赖规范
+
+技能根目录的 `requirements.txt` 是**标准文件**，用于声明本技能特有的 Python 三方依赖。
+
+- 匠厂宿主安装/更新技能后，会将其安装到共享 venv：`{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。
+- **不要**在业务代码中 `subprocess` / `pip install`；缺依赖由 `health` 报错，由宿主负责安装。
+- **版本约束尽量收窄**，降低多技能共享 venv 时的冲突风险。推荐范围写法：
+  - `pkg>=1.2.0,<2`
+- 对**原生扩展 / 高风险依赖**（如 `chromadb`、`onnxruntime`）建议 pin 或窄范围，避免无上限：
+  - 不推荐：`chromadb>=0.5.0`
+  - 推荐：`chromadb>=0.5.23,<0.6`
+- **不要**把系统组件（VC++ Runtime、浏览器安装包等）写进 `requirements.txt`；这类前置条件写在 `health` / preflight 文档与错误提示中。
+- 若技能无额外 Python 依赖，可保留仅含注释的 `requirements.txt`（宿主会识别为 empty 并跳过安装）。
+
 ## 4. 开发一个新 skill 的标准步骤
 
 下面这套顺序建议严格按步骤做，不要一上来就直接写 `service`。
@@ -461,7 +495,7 @@ python scripts/main.py <your-command>
 
 新 skill 不应该等所有业务做完才补测试。建议在每个开发阶段都先把对应测试跑一遍：
 
-1. **目录搭建后**：跑 `python tests/run_tests.py -v`，确认 6 个默认测试套件全部通过。
+1. **目录搭建后**：跑 `python tests/run_tests.py -v`，确认全部默认测试套件通过。
    - 此时 `test_skill_metadata.py` 会校验 `SKILL.md` slug 与 `constants.SKILL_SLUG` 一致；如果你只改了一边，会立刻发现。
 2. **改完 constants 后**：再跑一次必跑套件，确认未引入回归。
 3. **写完 service 业务后**：从 `tests/samples/test_service_contract.py.sample` 复制一份做契约测试。

references/README.md

@@ -46,3 +46,19 @@ description: "规范化的新技能模板说明；复制后替换名称、CLI
 - 旧入口 `scripts/skill_main.py`
 
 新模板统一使用 `scripts/main.py` 作为入口。
+
+## 参考文档索引
+
+| 文档 | 内容 |
+|------|------|
+| `DEVELOPMENT.md` | 从复制模板到开发出新 skill 的完整教程 |
+| `RUNTIME.md` | 运行时目录结构与 `CLAW_*` 环境变量约定 |
+| `CLI.md` | CLI 命令与参数示例 |
+| `SCHEMA.md` | task_logs 与字段映射说明 |
+| `TESTING.md` | 测试 target 与隔离体系 |
+| **`RPA.md`** | **三端 RPA 标准（浏览器/桌面/手机）：选型、拟人操作、反反爬、人工兜底、存证** |
+| **`ADAPTER.md`** | **真实/仿真 × API/RPA 四档适配器模式** |
+| **`CONFIG.md`** | **`.env` 配置规范与首次自动落盘机制** |
+| `REQUIREMENTS.md` | 依赖与运行要求 |
+
+> 开发"操作浏览器/桌面/手机"或"对接 ERP/CRM/SaaS/银行"类 skill 前，**务必先读 `RPA.md` + `ADAPTER.md` + `CONFIG.md`** 这三份标准，避免重复踩坑。

references/REQUIREMENTS.md

@@ -164,6 +164,18 @@
 - `【平台名】`：作用是 `【用途】`
 - `【浏览器 / API / 第三方服务】`：作用是 `【用途】`
 
+### Python 包依赖（requirements.txt）
+
+- 本技能若需额外 Python 三方包，写入根目录 `requirements.txt`。
+- 匠厂宿主安装/更新技能后，会安装到 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`。
+- 版本须尽量收窄（如 `requests>=2.31.0,<3`），避免共享 venv 冲突。
+- **不要**把系统运行时组件（如 Microsoft Visual C++ Redistributable）写进 requirements.txt。
+
+### 系统运行时前置条件（preflight / health）
+
+- 若依赖 Windows 原生 DLL、本机浏览器 channel 等**系统级**能力，在 `health` / preflight 中检测并提示用户自行安装。
+- 此类依赖**不**通过 pip / requirements.txt 由宿主自动安装。
+
 示例：
 
 ### 兄弟技能依赖
@@ -217,6 +229,8 @@
 - skill 可以正常安装
 - 安装后可在“新建任务”中调用
 
+**测试与日志相关验收（所有新 skill 必须满足）**：
+
 - `python tests/run_tests.py -v` 必跑测试全部通过
 - 任务日志（task_logs）写入和查询符合预期
 - 真实联调（如有）已在 tests/integration/ 中验证，且默认套件不包含真实外联

references/RPA.md

@@ -0,0 +1,175 @@
+# RPA 操作标准（浏览器 / 桌面 / 手机）
+
+> 本文是团队 RPA 开发的**统一标准**。任何需要"自动操作软件界面"的 skill，都应先读这份文档，按这里的选型和范式落地，不要每个项目重新踩坑。
+
+我们开发的各类 skill，本质上都是在替人操作三类界面：**浏览器、桌面软件、手机软件**。三类的底层技术不同，但**工程范式相同**：保持登录态、有头运行、拟人操作、失败存证、人工兜底。
+
+---
+
+## 0. 三端通用范式（先看这个）
+
+无论操作哪类界面，都遵循同一套约定：
+
+| 约定 | 说明 |
+|------|------|
+| **保持登录态** | 复用持久化 Profile / session，避免每次重新登录触发风控；账号由 account-manager 下发，不硬编码 |
+| **有头运行** | 默认有头（headless 易被识别 / 难人工介入）；`OPENCLAW_BROWSER_HEADLESS=1` 仅给 CI |
+| **拟人操作** | 真实事件（isTrusted=true），逐字输入、随机延迟、贝塞尔鼠标轨迹；严禁 JS 直接设值/JS 点击/JS 跳转 |
+| **步骤间随机等待** | 每两步操作之间 `random_delay(min,max)`，区间由 `.env` 配置（默认 1~5s） |
+| **人工兜底（HITL）** | 滑块 / 短信验证码 / 人脸 / U盾 / 动态口令 → **停下来轮询等人工**，超时报 `ERROR:XXX_NEED_HUMAN`，绝不自动硬闯 |
+| **失败存证** | 失败必截图，合规场景全程录屏，统一存 `{数据目录}/rpa-artifacts/{batch_id}/{tag}_{ts}.png` |
+| **选择器纪律** | 语义选择器优先（id/name/text/aria）；**F12 确认后再写，严禁凭记忆猜 DOM** |
+| **统一错误码** | `ERROR:REQUIRE_LOGIN` / `ERROR:CAPTCHA_NEED_HUMAN` / `ERROR:RATE_LIMITED` / `ERROR:LOGIN_TIMEOUT` 等，见下方错误码表 |
+| **幂等 / 断点续跑** | 批量操作记录"已处理到第几条"，崩溃后能续跑、不重复提交 |
+
+> 三端各自实现一个会话抽象 `RpaSession`（launch / login / act / screenshot / close），上层 skill 不感知是浏览器还是手机。
+
+---
+
+## 1. 浏览器（标准已成熟）
+
+**选型：Playwright + 系统 Chrome/Edge。** 这是团队验证最充分的一条线（参考 `1688-scrape-contacts`、`disburse-payroll-icbc`）。
+
+| 项 | 标准 |
+|----|------|
+| 驱动 | Playwright（Python） |
+| 浏览器 | 系统已装 Chrome/Edge，`channel="chrome"`；**不用内置 Chromium，无需 `playwright install`** |
+| 登录态 | `launch_persistent_context` + 用户 Profile 目录（由 account-manager 下发） |
+| 模式 | 默认有头；`OPENCLAW_BROWSER_HEADLESS=1` 转无头 |
+| 反反爬 | stealth 指纹注入 + 拟人输入/鼠标/延迟 + 滑块停下等人工（见下） |
+
+### 1.1 防反爬三件套（必做）
+
+1. **指纹淡化（stealth）**：启动时 `add_init_script` 注入，抹平自动化特征——
+   `navigator.webdriver=undefined`、`chrome.runtime`、`permissions.query`、`plugins`、`languages`、`hardwareConcurrency`、`deviceMemory`、WebGL vendor。
+   启动参数 `--disable-blink-features=AutomationControlled` + `ignore_default_args=["--enable-automation"]`。
+   开关：`OPENCLAW_PLAYWRIGHT_STEALTH`（默认开）。
+2. **拟人操作**：
+   - 输入：逐字符 `keyboard.type(delay=90~240ms)`，**先真实点击聚焦再输入，绝不 `el.value=`**。
+   - 鼠标：贝塞尔曲线轨迹移动 + 微抖动；进场随机晃动 2~4 次。
+   - 导航：用真实点击触发，**不要 `window.location.href=` / `el.click()` 这类 JS 跳转/点击**。
+   - 翻页：真实点击翻页箭头，注意排除禁用态选择器。
+   - 延迟：每步之间 `random_delay`。
+3. **风控页 = 停下等人工**：检测到滑块/拦截（URL 含 `punish/x5sec/baxia/tmd`，或 DOM 命中 `#nocaptcha/.nc-container/punish-component` 等）→ 抛 `ERROR:CAPTCHA_NEED_HUMAN`，轮询等人工通过或超时，**不自动操作滑块**。
+
+> 参考实现：`1688-scrape-contacts/scripts/service/anti_detect.py`、`util/playwright_stealth.py`、`service/login_service.py`（验证码等待）。已提升为共享库 `jiangchang_skill_core.rpa`（见 platform-kit）。
+
+### 1.2 安装
+
+```bash
+pip install playwright   # 仅装包；用系统 Chrome，无需 playwright install chromium
+```
+
+### 引用方式
+
+共享实现位于 `jiangchang-platform-kit` 的 `jiangchang_skill_core.rpa`；模板通过 vendor 拷贝引用：
+
+```python
+# scripts/jiangchang_skill_core/rpa/ — vendored from platform-kit
+from jiangchang_skill_core.rpa import (
+    launch_persistent_browser,
+    anti_detect,
+    wait_for_captcha_pass,
+    capture_failure,
+    errors,
+)
+from jiangchang_skill_core.rpa.stealth import stealth_enabled, STEALTH_INIT_SCRIPT
+```
+
+完整示例见 `scripts/service/example_browser_rpa.py` 与 `scripts/service/example_adapter/sim_rpa.py`。
+
+---
+
+## 2. 桌面软件（Windows 原生程序）
+
+**选型：pywinauto（UIA backend）为主 + 图像识别兜底。**
+
+桌面端常见于 ERP 客户端、网银控件、银企直连等本地程序。优先走可访问性树（控件 ID/名字），坐标点击只做最后兜底。
+
+| 技术 | 优先级 | 适用 / 说明 |
+|------|--------|------|
+| **pywinauto（`backend="uia"`）** | ✅ 首选 | 基于微软 UI Automation 树，拿控件 AutomationId/Name/ControlType，**稳定、不依赖屏幕坐标**，纯 Python |
+| **FlaUI**（经 pythonnet 调 .NET） | 备选 | UIA 拿不到的复杂/自绘控件时更完整，但需引入 .NET 运行时 |
+| **Playwright** | 特例 | 目标是 **Electron 套壳应用**（很多新 SaaS 客户端）时，可当浏览器驱动 |
+| **PyAutoGUI / SikuliX（图像识别）** | ⚠️ 兜底 | 控件树完全拿不到时（Flash/远程桌面/纯自绘 UI）；**靠截图找图+坐标，分辨率/缩放一变就崩**，仅最后手段 |
+
+### 桌面端注意事项
+
+- **窗口聚焦/置顶**：操作前确保目标窗口前置，避免误操作其它窗口。
+- **DPI/缩放**：图像识别方案必须固定显示缩放比例；UIA 方案不受影响（优先用 UIA 即是为此）。
+- **存证**：同样要失败截图（截目标窗口/全屏），存到 `rpa-artifacts`。
+- **人工兜底**：U盾插拔、动态口令、人脸 → 停下等人工，超时 `ERROR:XXX_NEED_HUMAN`。
+
+```bash
+pip install pywinauto        # UIA 自动化
+# 图像兜底：pip install pyautogui opencv-python
+```
+
+> 状态：桌面端选型为推荐标准，**尚待真实项目实战验证**，落地时回补踩坑记录到本文。
+
+---
+
+## 3. 手机软件（USB 连接电脑）
+
+**选型：Android 用 uiautomator2（或 Appium）；iOS 用 Appium + WebDriverAgent（需 Mac）。** 底层都是经 USB 的 ADB / WDA。
+
+| 平台 | 技术 | 优先级 | 说明 |
+|------|------|--------|------|
+| **Android** | **uiautomator2**（python 原生） | ✅ 首选 | ADB over USB，直接拿控件树点击/输入，比 Appium 轻快，纯 Python |
+| Android | **Appium**（uiautomator2 driver） | 备选 | 需要跨平台统一接口、或团队已有 Appium 资产时用 |
+| Android | **Airtest + Poco**（网易开源） | 兜底 | 图像+控件混合，自带 IDE 可录制；控件树拿不到时用 |
+| **iOS** | **Appium + WebDriverAgent（XCUITest）** | 唯一可行 | **必须有一台 Mac 做中转**，Windows host 无法直接驱动 iOS |
+| 投屏/人工介入 | **scrcpy** | 辅助 | USB 投屏到电脑，配合人工过验证码/人脸 |
+
+### 手机端注意事项
+
+- **设备就绪检查**：`adb devices` 确认已授权连接；放进 `doctor` 自检。
+- **登录态**：靠 App 自身保持登录，必要时引导人工首登一次。
+- **人工兜底**：短信验证码、人脸、指纹 → scrcpy 投屏让人工完成，程序轮询等待。
+- **存证**：失败时 `adb screencap` / Appium 截图存 `rpa-artifacts`。
+
+```bash
+# Android（首选）
+pip install uiautomator2
+python -m uiautomator2 init   # 初始化设备端 agent
+# 或统一走 Appium：pip install Appium-Python-Client（另需 Appium Server）
+```
+
+> 状态：手机端选型为推荐标准，**尚待真实项目实战验证**，落地时回补踩坑记录到本文。
+
+---
+
+## 4. 统一错误码（RPA 场景）
+
+skill 退出/抛错统一用 `ERROR:` 前缀 + 稳定码，方便宿主与上层判断与重试：
+
+| 错误码 | 含义 | 上层处理建议 |
+|--------|------|------|
+| `ERROR:REQUIRE_LOGIN` | 未登录 / 登录态失效 | 触发登录流程 |
+| `ERROR:LOGIN_TIMEOUT` | 等待人工登录超时 | 提示用户重跑并及时操作 |
+| `ERROR:CAPTCHA_NEED_HUMAN` | 命中滑块/验证码拦截 | 暂停等人工，或转人工队列 |
+| `ERROR:RATE_LIMITED` | 触发频控 | 退避后重试 |
+| `ERROR:MISSING_BROWSER` | 未检测到 Chrome/Edge | 提示安装 |
+| `ERROR:DEVICE_NOT_READY` | 手机未连接/未授权 | 检查 USB/ADB |
+| `ERROR:WINDOW_NOT_FOUND` | 桌面目标窗口未找到 | 检查程序是否启动 |
+
+---
+
+## 5. 存证规范
+
+- **路径**：`{CLAW_DATA_ROOT}/{user}/{slug}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.png`
+- **失败必截图**：受 `OPENCLAW_ARTIFACTS_ON_FAILURE`（默认开）控制。
+- **运行录屏成片**：RPA 型技能默认 `OPENCLAW_RECORD_VIDEO=1`，表示 **ffmpeg 录制运行过程** + 字幕 + 背景音乐 + 最终 MP4 **一体开启**。由 `RpaVideoSession` 统一触发，不在 CLI/E2E 单独实现。
+- **Playwright 不负责录屏**，仅浏览器自动化。
+- **ffmpeg 是唯一录屏器**（Windows：`gdigrab` + `desktop`）。
+- **最终视频**：`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
+- **中间产物**：`rpa-artifacts/{batch_id}/capture.mp4`、`subtitles/`、`logs/` 等，**不放最终 MP4**。
+- **常见 tag**：`before_submit` / `after_submit` / `captcha` / `login_fail` / `error`。
+
+---
+
+## 相关文档
+
+- `ADAPTER.md` — 真实/仿真 × API/RPA 的四档适配器模式
+- `CONFIG.md` — `.env` 配置规范与首次落盘机制
+- `RUNTIME.md` — 运行时目录与环境变量约定

references/TESTING.md

@@ -101,7 +101,7 @@ def test_whatever():
 
 **何时用它**：service 层出现『调用第三方 HTTP / RPA stub』但又不能把真实系统纳入 CI。**契约测试**（复制 [`tests/samples/test_service_contract.py.sample`](../tests/samples/test_service_contract.py.sample)）应优先组合 FakeAdapter，而不是直接把 CLI when-json 断言堆上天。
 
-把它看成：**你把不确定性折叠进可控表格**，而不是在生产日志里才第一次看到错位字段。
+把它看成：**你把不确定性收敛到可控的测试矩阵里**，而不是在生产日志里才第一次看到错位字段。
 
 ---
 
@@ -124,12 +124,12 @@ Golden fixture 流程同理（[`tests/samples/test_golden_cases.py.sample`](../t
 任何 touching **真实租户数据** 的路径：
 
 1. **禁止**硬编码 token / cookie / 内部域名落入仓库；
-2. **禁止**默认套件 silent import integration；
+2. **禁止**默认套件隐式导入 integration；
 3. **真实 RPA** 只能标记为手动触发（双人复核 / 本地 `.env` 不入库）。
 
 范式阅读 [`tests/integration/README.md`](../tests/integration/README.md)：那里有针对凭证来源、目录 artifact 忽略策略的补充。
 
-记住：**测试代码也是一种部署面**，别把 staging 凭证烘焙进来。
+记住：**测试代码也是一种部署面**，别把 staging 凭证写死进仓库。
 
 ---
 
@@ -156,9 +156,9 @@ Golden fixture 流程同理（[`tests/samples/test_golden_cases.py.sample`](../t
 
 | 红线 | 解释 |
 |------|------|
-| 不改业务凑测试 | 除非需求变更已确认，否则别让 CI 绿通过阉割业务分支达成 |
+| 不改业务凑测试 | 除非需求变更已确认，否则**不要为了让 CI 变绿而砍掉业务分支** |
 | **默认套件零外联** | 不把真实 HTTP / 浏览器写进 `tests/test_*.py` |
-| `.sample` 尊重 | 集成范式改名前先读完 README；别让 `.sample` silent 变成根测试 |
+| `.sample` 尊重 | 集成范式改名前先读完 README；别让 `.sample` 悄悄变成根测试 |
 | **零硬编码凭证** | token / cookie / 生产 URL → 用虚构域名或 vault ref |
 | mock 优先 | 逻辑应在 service + FakeAdapter，而不是巨胖 CLI 断言 |
 | 结构化错误 | 断言错误码字段，而不是 substring of stderr 漂移集合 |

requirements.txt

@@ -0,0 +1,7 @@
+# Skill-specific Python dependencies.
+# Keep this file minimal.
+# Platform-provided shared dependencies are installed by Jiangchang runtime.
+# Add only dependencies required by this skill.
+#
+# Browser RPA: uses system Chrome/Edge; no `playwright install chromium` needed.
+playwright>=1.42.0,<2

scripts/jiangchang_skill_core/__init__.py

@@ -1 +1,16 @@
-# Vendored from jiangchang-platform-kit/sdk/jiangchang_skill_core/ — keep runtime_env + unified_logging in sync.
+# Vendored from jiangchang-platform-kit/sdk/jiangchang_skill_core/ — keep runtime_env + unified_logging + config + rpa in sync.
+from .config import ensure_env_file, get, get_bool, get_float, get_int
+
+try:
+    from . import rpa
+except ImportError:
+    rpa = None  # type: ignore[assignment,misc]
+
+__all__ = [
+    "ensure_env_file",
+    "get",
+    "get_bool",
+    "get_float",
+    "get_int",
+    "rpa",
+]

scripts/jiangchang_skill_core/config.py

@@ -0,0 +1,126 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""三级优先级配置读取 + 首次 .env 落盘（进程 env > 数据目录 .env > .env.example）。"""
+
+from __future__ import annotations
+
+import os
+import shutil
+from typing import Any
+
+from .runtime_env import get_data_root, get_user_id
+
+_skill_slug: str | None = None
+_example_path: str | None = None
+_env_file_path: str | None = None
+_user_env_cache: dict[str, str] | None = None
+_example_defaults_cache: dict[str, str] | None = None
+
+
+def _parse_env_file(path: str) -> dict[str, str]:
+    """标准库手写 .env 解析（不引 python-dotenv）。"""
+    result: dict[str, str] = {}
+    if not path or not os.path.isfile(path):
+        return result
+    with open(path, encoding="utf-8") as f:
+        for raw_line in f:
+            line = raw_line.strip()
+            if not line or line.startswith("#"):
+                continue
+            if "=" not in line:
+                continue
+            key, _, value = line.partition("=")
+            key = key.strip()
+            value = value.strip()
+            if not key:
+                continue
+            if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
+                value = value[1:-1]
+            elif " #" in value:
+                value = value.split(" #", 1)[0].strip()
+            result[key] = value
+    return result
+
+
+def _get_user_env() -> dict[str, str]:
+    global _user_env_cache
+    if _user_env_cache is not None:
+        return _user_env_cache
+    if _env_file_path and os.path.isfile(_env_file_path):
+        _user_env_cache = _parse_env_file(_env_file_path)
+    else:
+        _user_env_cache = {}
+    return _user_env_cache
+
+
+def _get_example_defaults() -> dict[str, str]:
+    global _example_defaults_cache
+    if _example_defaults_cache is not None:
+        return _example_defaults_cache
+    if _example_path and os.path.isfile(_example_path):
+        _example_defaults_cache = _parse_env_file(_example_path)
+    else:
+        _example_defaults_cache = {}
+    return _example_defaults_cache
+
+
+def reset_cache() -> None:
+    """测试用：清空解析缓存。"""
+    global _user_env_cache, _example_defaults_cache
+    _user_env_cache = None
+    _example_defaults_cache = None
+
+
+def ensure_env_file(skill_slug: str, example_path: str) -> str:
+    """首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env（已存在不覆盖），返回路径。"""
+    global _skill_slug, _example_path, _env_file_path
+    _skill_slug = skill_slug
+    _example_path = os.path.abspath(example_path)
+    dest_dir = os.path.join(get_data_root(), get_user_id(), skill_slug)
+    os.makedirs(dest_dir, exist_ok=True)
+    dest = os.path.join(dest_dir, ".env")
+    _env_file_path = dest
+    if not os.path.isfile(dest) and os.path.isfile(_example_path):
+        shutil.copy2(_example_path, dest)
+    reset_cache()
+    return dest
+
+
+def get(key: str, default: Any = None) -> str | None:
+    """进程环境变量 > 数据目录 .env > .env.example 默认值。"""
+    env_val = os.environ.get(key)
+    if env_val is not None and env_val != "":
+        return env_val
+    user_val = _get_user_env().get(key)
+    if user_val is not None and user_val != "":
+        return user_val
+    example_val = _get_example_defaults().get(key)
+    if example_val is not None and example_val != "":
+        return example_val
+    return default
+
+
+def get_bool(key: str, default: bool = False) -> bool:
+    val = get(key)
+    if val is None:
+        return default
+    return str(val).strip().lower() in ("1", "true", "yes", "on")
+
+
+def get_float(key: str, default: float) -> float:
+    val = get(key)
+    if val is None:
+        return default
+    try:
+        return float(val)
+    except (TypeError, ValueError):
+        return default
+
+
+def get_int(key: str, default: int) -> int:
+    val = get(key)
+    if val is None:
+        return default
+    try:
+        return int(val)
+    except (TypeError, ValueError):
+        return default

scripts/jiangchang_skill_core/rpa/__init__.py

@@ -0,0 +1,66 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""浏览器 RPA 共享库：stealth、拟人操作、启动封装、HITL 等待、失败存证。"""
+
+from __future__ import annotations
+
+from . import anti_detect, artifacts, errors, human_login, stealth
+
+try:
+    from .browser import launch_persistent_browser
+except ImportError:
+    launch_persistent_browser = None  # type: ignore[misc, assignment]
+
+try:
+    from .video_session import RpaVideoSession
+except ImportError:
+    RpaVideoSession = None  # type: ignore[misc, assignment]
+
+__all__ = [
+    "anti_detect",
+    "artifacts",
+    "errors",
+    "human_login",
+    "stealth",
+    "launch_persistent_browser",
+    "RpaVideoSession",
+    # re-export common anti_detect helpers
+    "random_delay",
+    "human_delay_short",
+    "human_delay_page",
+    "human_delay_batch",
+    "human_mouse_move",
+    "human_mouse_wiggle",
+    "human_click",
+    "human_type",
+    "human_scroll",
+    # human_login
+    "wait_for_captcha_pass",
+    "wait_for_login",
+    "DEFAULT_CAPTCHA_MARKERS",
+    # artifacts
+    "artifact_path",
+    "capture_failure",
+    # stealth
+    "STEALTH_INIT_SCRIPT",
+    "stealth_enabled",
+    "persistent_context_launch_parts",
+]
+
+from .anti_detect import (
+    human_click,
+    human_delay_batch,
+    human_delay_page,
+    human_delay_short,
+    human_mouse_move,
+    human_mouse_wiggle,
+    human_scroll,
+    human_type,
+    random_delay,
+)
+from .artifacts import artifact_path, capture_failure
+from .human_login import DEFAULT_CAPTCHA_MARKERS, wait_for_captcha_pass, wait_for_login
+from .stealth import (
+    STEALTH_INIT_SCRIPT,
+    persistent_context_launch_parts,
+    stealth_enabled,
+)

scripts/jiangchang_skill_core/rpa/anti_detect.py

@@ -0,0 +1,216 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""防反爬：随机延迟、人工鼠标轨迹模拟、拟人滚动。
+
+所有延迟使用 random.uniform() + asyncio.sleep()。
+鼠标移动使用三次贝塞尔曲线 + 随机抖动，避免机械直线轨迹。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import math
+import random
+from typing import Optional
+
+
+# ── 随机延迟 ────────────────────────────────────────────────
+
+
+async def random_delay(min_s: float, max_s: float) -> float:
+    """随机延迟 min_s ~ max_s 秒，返回实际延迟秒数。"""
+    delay = random.uniform(min_s, max_s)
+    await asyncio.sleep(delay)
+    return delay
+
+
+async def human_delay_short() -> float:
+    """页面内操作间短延迟 0.5~2 秒（点击/滚动）。"""
+    return await random_delay(0.5, 2.0)
+
+
+async def human_delay_page(min_s: float = 2, max_s: float = 8) -> float:
+    """页面间/店铺间切换延迟（可配置范围）。"""
+    return await random_delay(min_s, max_s)
+
+
+async def human_delay_batch(min_s: float = 5, max_s: float = 15) -> float:
+    """批次间/翻页间长延迟。"""
+    return await random_delay(min_s, max_s)
+
+
+# ── 鼠标轨迹 ────────────────────────────────────────────────
+
+
+async def human_mouse_move(
+    page,
+    target_x: float,
+    target_y: float,
+    steps: Optional[int] = None,
+) -> None:
+    """模拟人工鼠标移动到目标坐标。
+
+    使用三次贝塞尔曲线生成 S/C 形路径，中间控制点随机偏移，
+    每步添加微小随机抖动，避免完美直线移动。
+
+    Args:
+        page: Playwright Page 对象。
+        target_x: 目标 X 坐标。
+        target_y: 目标 Y 坐标。
+        steps: 移动步数（None 时按距离自动计算）。
+    """
+    if steps is None:
+        dist = math.hypot(target_x, target_y)
+        steps = max(8, int(dist / 15))
+
+    # 随机起始点（模拟鼠标从页面其他位置移动过来）
+    start_x = target_x + random.uniform(-300, 300)
+    start_y = target_y + random.uniform(-200, 200)
+
+    # 两个贝塞尔控制点（制造弯曲路径）
+    dx = target_x - start_x
+    dy = target_y - start_y
+
+    cp1_x = start_x + dx * random.uniform(0.2, 0.5) + random.uniform(-120, 120)
+    cp1_y = start_y + dy * random.uniform(0.1, 0.4) + random.uniform(-100, 100)
+    cp2_x = start_x + dx * random.uniform(0.5, 0.8) + random.uniform(-100, 100)
+    cp2_y = start_y + dy * random.uniform(0.4, 0.7) + random.uniform(-80, 80)
+
+    for i in range(steps + 1):
+        t = i / steps
+        # 三次贝塞尔: B(t) = (1-t)³·P0 + 3(1-t)²·t·P1 + 3(1-t)·t²·P2 + t³·P3
+        u = 1 - t
+        x = (
+            (u ** 3) * start_x
+            + 3 * (u ** 2) * t * cp1_x
+            + 3 * u * (t ** 2) * cp2_x
+            + (t ** 3) * target_x
+        )
+        y = (
+            (u ** 3) * start_y
+            + 3 * (u ** 2) * t * cp1_y
+            + 3 * u * (t ** 2) * cp2_y
+            + (t ** 3) * target_y
+        )
+
+        # 微小随机抖动
+        x += random.uniform(-1.5, 1.5)
+        y += random.uniform(-1.5, 1.5)
+
+        await page.mouse.move(x, y)
+        await asyncio.sleep(random.uniform(0.003, 0.015))
+
+    # 最终精确到位
+    await page.mouse.move(target_x, target_y)
+
+
+async def human_click(
+    page,
+    selector: Optional[str] = None,
+    *,
+    x: Optional[float] = None,
+    y: Optional[float] = None,
+) -> None:
+    """拟人点击：贝塞尔轨迹移动 → 微延迟 → 点击。
+
+    支持两种定位方式：
+    - selector: CSS 选择器，自动取元素中心附近随机偏移
+    - x, y: 直接指定坐标（自动加微小随机偏移）
+
+    Args:
+        page: Playwright Page 对象。
+        selector: CSS 选择器（与 x/y 二选一）。
+        x: 目标 X 坐标。
+        y: 目标 Y 坐标。
+    """
+    if selector:
+        locator = page.locator(selector).first
+        box = await locator.bounding_box()
+        if box is None:
+            raise ValueError(f"无法获取元素边界: {selector}")
+        target_x = box["x"] + box["width"] * random.uniform(0.3, 0.7)
+        target_y = box["y"] + box["height"] * random.uniform(0.3, 0.7)
+    elif x is not None and y is not None:
+        target_x = x + random.uniform(-3, 3)
+        target_y = y + random.uniform(-3, 3)
+    else:
+        raise ValueError("必须提供 selector 或 (x, y) 坐标")
+
+    await human_mouse_move(page, target_x, target_y)
+    await human_delay_short()
+    await page.mouse.click(target_x, target_y)
+
+
+# ── 滚动模拟 ────────────────────────────────────────────────
+
+
+async def human_scroll(
+    page, direction: str = "down", amount: Optional[int] = None
+) -> None:
+    """模拟人工滚动（分段 + 变速 + 微延迟）。
+
+    Args:
+        page: Playwright Page 对象。
+        direction: "up" 或 "down"。
+        amount: 滚动像素总量，默认随机 300~800。
+    """
+    if amount is None:
+        amount = random.randint(300, 800)
+
+    sign = -1 if direction == "up" else 1
+    remaining = amount
+    step_count = random.randint(3, 7)
+
+    for i in range(step_count):
+        step = remaining // (step_count - i) + random.randint(-30, 30)
+        step = max(10, min(step, remaining))
+        remaining -= step
+
+        await page.mouse.wheel(0, sign * step)
+        await asyncio.sleep(random.uniform(0.05, 0.2))
+
+    if remaining > 0:
+        await page.mouse.wheel(0, sign * remaining)
+
+    await human_delay_short()
+
+
+# ── 进场鼠标随机晃动 ────────────────────────────────────────
+
+
+async def human_mouse_wiggle(page) -> None:
+    """页面进入后随机移动鼠标 2~4 次（贝塞尔轨迹），模拟真人到场环顾。"""
+    try:
+        vp = page.viewport_size or {"width": 1280, "height": 800}
+    except Exception:
+        vp = {"width": 1280, "height": 800}
+    w, h = vp.get("width", 1280), vp.get("height", 800)
+    for _ in range(random.randint(2, 4)):
+        tx = random.uniform(w * 0.2, w * 0.8)
+        ty = random.uniform(h * 0.2, h * 0.7)
+        await human_mouse_move(page, tx, ty)
+        await asyncio.sleep(random.uniform(0.2, 0.8))
+
+
+# ── 拟人逐字输入 ────────────────────────────────────────────
+
+
+async def human_type(page, locator, text: str) -> None:
+    """真人式输入：贝塞尔移动到输入框 → 真实点击聚焦 → 逐字符 keyboard.type。
+
+    全程使用 isTrusted=true 的可信事件，绝不用 JS 设置 value。
+    """
+    box = await locator.bounding_box()
+    if box is None:
+        raise ValueError("human_type: 无法获取输入框边界")
+    tx = box["x"] + box["width"] * random.uniform(0.3, 0.7)
+    ty = box["y"] + box["height"] * random.uniform(0.3, 0.7)
+    await human_mouse_move(page, tx, ty)
+    await asyncio.sleep(random.uniform(0.15, 0.5))
+    await page.mouse.click(tx, ty)
+    await asyncio.sleep(random.uniform(0.2, 0.6))
+    # 清空已有内容（用键盘，不用 JS）
+    await page.keyboard.press("Control+A")
+    await page.keyboard.press("Delete")
+    await asyncio.sleep(random.uniform(0.2, 0.5))
+    for ch in text:
+        await page.keyboard.type(ch, delay=random.uniform(90, 240))

scripts/jiangchang_skill_core/rpa/artifacts.py

@@ -0,0 +1,29 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""RPA 失败存证路径与截图。"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+
+
+def _artifacts_enabled() -> bool:
+    v = (os.environ.get("OPENCLAW_ARTIFACTS_ON_FAILURE") or "1").strip().lower()
+    return v not in ("0", "false", "no", "off")
+
+
+def artifact_path(data_dir: str, batch_id: str, tag: str, ext: str = "png") -> str:
+    """返回 {data_dir}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.{ext}"""
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    directory = os.path.join(data_dir, "rpa-artifacts", batch_id)
+    os.makedirs(directory, exist_ok=True)
+    return os.path.join(directory, f"{tag}_{ts}.{ext}")
+
+
+async def capture_failure(page, data_dir: str, batch_id: str, tag: str) -> str:
+    """受 OPENCLAW_ARTIFACTS_ON_FAILURE 控制，截图并返回路径；禁用时返回空字符串。"""
+    if not _artifacts_enabled():
+        return ""
+    path = artifact_path(data_dir, batch_id, tag, ext="png")
+    await page.screenshot(path=path, full_page=True)
+    return path

scripts/jiangchang_skill_core/rpa/browser.py

@@ -0,0 +1,58 @@
+"""统一 persistent context 启动封装（仅浏览器自动化，不负责录屏）。"""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+
+from ..runtime_env import find_chrome_executable
+from .stealth import (
+    STEALTH_INIT_SCRIPT,
+    persistent_context_launch_parts,
+    stealth_enabled,
+)
+
+if TYPE_CHECKING:
+    from playwright.async_api import BrowserContext
+
+
+async def launch_persistent_browser(
+    playwright,
+    *,
+    profile_dir: str,
+    channel: str = "chrome",
+    executable_path: str | None = None,
+    headless: bool | None = None,
+    extra_args: list[str] | None = None,
+) -> "BrowserContext":
+    """用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。
+
+    headless=None 时读 OPENCLAW_BROWSER_HEADLESS。
+    录屏由 RpaVideoSession（ffmpeg）负责，本函数不向 Playwright 传递任何录屏参数。
+    """
+    if headless is None:
+        v = (os.environ.get("OPENCLAW_BROWSER_HEADLESS") or "0").strip().lower()
+        headless = v in ("1", "true", "yes", "on")
+
+    chrome = executable_path or find_chrome_executable()
+    if not chrome:
+        raise RuntimeError(
+            "ERROR:MISSING_BROWSER 未检测到 Chrome 或 Edge 浏览器，请安装后重试。"
+        )
+
+    args, ignore = persistent_context_launch_parts(extra_args=extra_args)
+    launch_kwargs: dict = dict(
+        user_data_dir=profile_dir,
+        headless=headless,
+        executable_path=chrome,
+        locale="zh-CN",
+        no_viewport=True,
+        args=args,
+    )
+    if ignore is not None:
+        launch_kwargs["ignore_default_args"] = ignore
+
+    context = await playwright.chromium.launch_persistent_context(**launch_kwargs)
+    if stealth_enabled():
+        await context.add_init_script(STEALTH_INIT_SCRIPT)
+    return context

scripts/jiangchang_skill_core/rpa/errors.py

@@ -0,0 +1,20 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""RPA 场景统一错误码（stdout / 异常消息前缀）。"""
+
+ERR_REQUIRE_LOGIN = "ERROR:REQUIRE_LOGIN"
+ERR_LOGIN_TIMEOUT = "ERROR:LOGIN_TIMEOUT"
+ERR_CAPTCHA_NEED_HUMAN = "ERROR:CAPTCHA_NEED_HUMAN"
+ERR_RATE_LIMITED = "ERROR:RATE_LIMITED"
+ERR_MISSING_BROWSER = "ERROR:MISSING_BROWSER"
+ERR_DEVICE_NOT_READY = "ERROR:DEVICE_NOT_READY"
+ERR_WINDOW_NOT_FOUND = "ERROR:WINDOW_NOT_FOUND"
+
+__all__ = [
+    "ERR_REQUIRE_LOGIN",
+    "ERR_LOGIN_TIMEOUT",
+    "ERR_CAPTCHA_NEED_HUMAN",
+    "ERR_RATE_LIMITED",
+    "ERR_MISSING_BROWSER",
+    "ERR_DEVICE_NOT_READY",
+    "ERR_WINDOW_NOT_FOUND",
+]

scripts/jiangchang_skill_core/rpa/human_login.py

@@ -0,0 +1,97 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""验证码/登录 HITL 等待（站点无关通用部分）。"""
+
+from __future__ import annotations
+
+import asyncio
+
+DEFAULT_CAPTCHA_MARKERS = (
+    "punish",
+    "baxia",
+    "nc_login",
+    "tmd",
+    "x5sec",
+    "安全验证",
+    "请按住滑块",
+    "拖动到最右边",
+)
+
+
+async def wait_for_captcha_pass(
+    page,
+    captcha_markers: tuple[str, ...] = DEFAULT_CAPTCHA_MARKERS,
+    timeout_sec: int = 120,
+) -> bool:
+    """
+    检测到验证码/拦截页时，暂停等待用户手工完成验证。
+    每 3 秒轮询一次页面状态，直到页面不再是拦截状态或超时。
+
+    Returns:
+        True  — 验证码已通过，可以继续
+        False — 超时，用户未在规定时间内完成验证
+    """
+    print(
+        f"[验证码] 检测到安全拦截，请在浏览器中手工完成验证。"
+        f"等待最多 {timeout_sec} 秒..."
+    )
+
+    elapsed = 0
+    while elapsed < timeout_sec:
+        await asyncio.sleep(3)
+        elapsed += 3
+        try:
+            url = page.url or ""
+            text = ""
+            try:
+                text = await page.locator("body").inner_text(timeout=2_000)
+            except Exception:
+                pass
+            haystack = (url + " " + text[:1000]).lower()
+            still_blocked = any(m.lower() in haystack for m in captcha_markers)
+            if not still_blocked:
+                print(f"[验证码] 已通过！({elapsed}s 后检测到正常页面)")
+                return True
+        except Exception:
+            pass
+
+    print(f"[验证码] 超时 {timeout_sec}s，用户未完成验证")
+    return False
+
+
+async def wait_for_login(
+    page,
+    success_selectors: list[str],
+    timeout_sec: int = 180,
+) -> bool:
+    """
+    轮询等待登录成功标志出现（站点专属选择器由调用方传入）。
+
+    Args:
+        page: Playwright Page 对象。
+        success_selectors: 登录成功后应出现的 CSS 选择器列表。
+        timeout_sec: 最长等待秒数。
+
+    Returns:
+        True  — 检测到登录成功标志
+        False — 超时
+    """
+    print(
+        f"[登录] 请在浏览器中完成登录。等待最多 {timeout_sec} 秒，"
+        "期间程序不会操作页面..."
+    )
+
+    elapsed = 0
+    while elapsed < timeout_sec:
+        await asyncio.sleep(2)
+        elapsed += 2
+        for selector in success_selectors:
+            try:
+                el = await page.query_selector(selector)
+                if el is not None:
+                    print("[登录] 检测到登录成功！")
+                    return True
+            except Exception:
+                pass
+
+    print(f"[登录] 超时 {timeout_sec}s，未完成登录")
+    return False

scripts/jiangchang_skill_core/rpa/stealth.py

@@ -0,0 +1,67 @@
+# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
+"""淡化 Playwright 自动化指纹，降低 baxia/x5sec 风控命中率。
+
+关闭：环境变量 OPENCLAW_PLAYWRIGHT_STEALTH=0（或 false/off/no）。
+"""
+
+from __future__ import annotations
+
+import os
+
+STEALTH_INIT_SCRIPT = """
+(() => {
+  try { Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); } catch (e) {}
+  try {
+    window.chrome = window.chrome || {};
+    window.chrome.runtime = window.chrome.runtime || {};
+  } catch (e) {}
+  try {
+    const orig = navigator.permissions && navigator.permissions.query;
+    if (orig) {
+      navigator.permissions.query = (p) =>
+        p && p.name === 'notifications'
+          ? Promise.resolve({ state: Notification.permission })
+          : orig(p);
+    }
+  } catch (e) {}
+  try {
+    Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
+  } catch (e) {}
+  try {
+    Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] });
+  } catch (e) {}
+  try {
+    Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 });
+  } catch (e) {}
+  try {
+    Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 });
+  } catch (e) {}
+  try {
+    const getParameter = WebGLRenderingContext.prototype.getParameter;
+    WebGLRenderingContext.prototype.getParameter = function (p) {
+      if (p === 37445) return 'Intel Inc.';
+      if (p === 37446) return 'Intel Iris OpenGL Engine';
+      return getParameter.call(this, p);
+    };
+  } catch (e) {}
+})();
+"""
+
+
+def stealth_enabled() -> bool:
+    v = (os.environ.get("OPENCLAW_PLAYWRIGHT_STEALTH") or "1").strip().lower()
+    return v not in ("0", "false", "no", "off")
+
+
+def persistent_context_launch_parts(
+    *, extra_args: list[str] | None = None
+) -> tuple[list[str], list[str] | None]:
+    """返回 (args, ignore_default_args)。ignore_default_args 为 None 时不要传给 launch。"""
+    args = ["--start-maximized"]
+    if extra_args:
+        args = args + list(extra_args)
+    if not stealth_enabled():
+        return args, None
+    if "--disable-blink-features=AutomationControlled" not in args:
+        args.append("--disable-blink-features=AutomationControlled")
+    return args, ["--enable-automation"]

scripts/jiangchang_skill_core/rpa/video_session.py

@@ -0,0 +1,420 @@
+"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。
+
+Playwright 仅用于浏览器自动化，不负责录屏。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import random
+import shutil
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from .. import config
+
+_JIANGCHANG_SUBTITLE_STYLE = (
+    "FontName=Arial,FontSize=14,"
+    "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
+    "Outline=2,Shadow=1,Alignment=2"
+)
+
+_DEFAULT_MUSIC_ROOT = os.environ.get(
+    "MEDIA_ASSETS_ROOT",
+    r"D:\OpenClaw\client-commons\media-assets",
+)
+
+_CAPTURE_FRAMERATE = 15
+
+
+def _record_video_enabled() -> bool:
+    return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
+
+
+def _resolve_music_file() -> Optional[Path]:
+    music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music"
+    if not music_dir.is_dir():
+        return None
+    files = list(music_dir.rglob("*.mp3"))
+    if not files:
+        return None
+    return random.choice(files)
+
+
+@dataclass
+class _StepEntry:
+    start: float
+    text: str
+    duration: float = 4.0
+
+
+def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
+    MIN_DURATION = 1.0
+    GAP = 0.1
+    steps = sorted(entries, key=lambda e: e.start)
+    if not steps:
+        steps = [_StepEntry(0.0, "任务执行中", 4.0)]
+
+    for i in range(len(steps) - 1):
+        cur, nxt = steps[i], steps[i + 1]
+        max_end = nxt.start - GAP
+        new_dur = max_end - cur.start
+        if new_dur < MIN_DURATION:
+            cur.duration = MIN_DURATION
+            nxt.start = cur.start + MIN_DURATION + GAP
+        else:
+            cur.duration = min(cur.duration, new_dur)
+
+    def _fmt(s: float) -> str:
+        h = int(s // 3600)
+        m = int((s % 3600) // 60)
+        sec = int(s % 60)
+        ms = int((s % 1) * 1000)
+        return f"{h:02d}:{m:02d}:{sec:02d},{ms:03d}"
+
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w", encoding="utf-8") as f:
+        for i, e in enumerate(steps, 1):
+            f.write(f"{i}\n")
+            f.write(f"{_fmt(e.start)} --> {_fmt(e.start + e.duration)}\n")
+            f.write(f"{e.text}\n\n")
+
+
+def _escape_srt_path(path: str) -> str:
+    return path.replace("\\", "/").replace(":", "\\:")
+
+
+def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+        )
+    except FileNotFoundError:
+        return False, "ffmpeg not found"
+    if result.returncode != 0:
+        err = (result.stderr or result.stdout or "").strip()
+        return False, err[:500] if err else f"exit {result.returncode}"
+    return True, ""
+
+
+def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]:
+    """Windows gdigrab 全桌面录制；非 Windows 返回 None。"""
+    if sys.platform != "win32":
+        return None
+    return [
+        "ffmpeg",
+        "-y",
+        "-f",
+        "gdigrab",
+        "-framerate",
+        str(_CAPTURE_FRAMERATE),
+        "-i",
+        "desktop",
+        "-c:v",
+        "libx264",
+        "-preset",
+        "ultrafast",
+        "-pix_fmt",
+        "yuv420p",
+        capture_path,
+    ]
+
+
+def _compose_capture_mp4(
+    capture_path: Path,
+    srt_path: Path,
+    output_path: Path,
+    *,
+    music_file: Optional[Path] = None,
+    music_volume: float = 0.15,
+) -> tuple[bool, str]:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    srt_esc = _escape_srt_path(str(srt_path.resolve()))
+    subtitle_filter = (
+        f"subtitles='{srt_esc}'"
+        f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
+    )
+
+    if music_file and music_file.is_file():
+        filter_complex = (
+            f"[0:v]{subtitle_filter}[vout];"
+            f"[1:a]volume={music_volume},apad[aout]"
+        )
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(capture_path),
+            "-i",
+            str(music_file),
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            "[vout]",
+            "-map",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-shortest",
+            "-pix_fmt",
+            "yuv420p",
+            str(output_path),
+        ]
+    else:
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(capture_path),
+            "-vf",
+            subtitle_filter,
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-pix_fmt",
+            "yuv420p",
+            str(output_path),
+        ]
+    return _run_ffmpeg(cmd)
+
+
+class RpaVideoSession:
+    """Service/RPA 层统一 ffmpeg 桌面录屏；OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
+
+    def __init__(
+        self,
+        *,
+        skill_slug: str,
+        skill_data_dir: str,
+        batch_id: str,
+        title: str = "",
+    ) -> None:
+        self.skill_slug = skill_slug
+        self.skill_data_dir = os.path.abspath(skill_data_dir)
+        self.batch_id = batch_id
+        self.title = title
+        self.enabled = _record_video_enabled()
+        self.warnings: List[str] = []
+        self.output_video_path: Optional[str] = None
+        self.capture_path: Optional[str] = None
+        # 已废弃：勿用于 Playwright；始终为 None。
+        self.record_video_dir: Optional[str] = None
+        self.artifacts: Dict[str, Any] = {}
+
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        safe_batch = "".join(c if c.isalnum() or c in "-_" else "_" for c in batch_id)
+        self._video_stem = f"{skill_slug}_{ts}_{safe_batch}"
+
+        self._videos_dir = os.path.join(self.skill_data_dir, "videos")
+        self._artifact_root = os.path.join(
+            self.skill_data_dir, "rpa-artifacts", batch_id
+        )
+        self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
+        self._logs_dir = os.path.join(self._artifact_root, "logs")
+        self._capture_path = os.path.join(self._artifact_root, "capture.mp4")
+        self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt")
+        self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
+
+        self._steps: List[_StepEntry] = []
+        self._t0: Optional[float] = None
+        self._ffmpeg_proc: Optional[subprocess.Popen] = None
+
+        if self.enabled:
+            os.makedirs(self._videos_dir, exist_ok=True)
+            os.makedirs(self._subtitles_dir, exist_ok=True)
+            os.makedirs(self._logs_dir, exist_ok=True)
+            os.makedirs(self._artifact_root, exist_ok=True)
+
+    def step(self, text: str, *, duration: float = 4.0) -> None:
+        self.add_step(text, duration=duration)
+
+    def add_step(self, text: str, *, duration: float = 4.0) -> None:
+        """记录一步骤字幕（相对录屏开始时间）。"""
+        if not self.enabled:
+            return
+        if self._t0 is None:
+            self._t0 = time.monotonic()
+        elapsed = time.monotonic() - self._t0
+        self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration))
+
+    def _start_ffmpeg_capture(self) -> None:
+        if not self.enabled:
+            return
+        if not shutil.which("ffmpeg"):
+            self.warnings.append("ffmpeg_not_found")
+            return
+
+        cmd = _build_desktop_record_cmd(self._capture_path)
+        if cmd is None:
+            self.warnings.append(
+                "ffmpeg_record_start_failed: desktop capture only supported on Windows"
+            )
+            return
+
+        try:
+            self._ffmpeg_proc = subprocess.Popen(
+                cmd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+            self.capture_path = self._capture_path
+        except Exception as exc:
+            self.warnings.append(f"ffmpeg_record_start_failed: {exc}")
+            self._ffmpeg_proc = None
+
+    def _stop_ffmpeg_capture(self) -> None:
+        proc = self._ffmpeg_proc
+        if proc is None:
+            return
+        if proc.poll() is None:
+            try:
+                if proc.stdin:
+                    proc.stdin.write(b"q")
+                    proc.stdin.flush()
+            except Exception:
+                try:
+                    proc.terminate()
+                except Exception:
+                    pass
+            try:
+                proc.wait(timeout=20)
+            except subprocess.TimeoutExpired:
+                try:
+                    proc.kill()
+                except Exception:
+                    pass
+                try:
+                    proc.wait(timeout=5)
+                except Exception:
+                    pass
+        self._ffmpeg_proc = None
+        await_sleep = 0.3
+        time.sleep(await_sleep)
+
+    async def __aenter__(self) -> "RpaVideoSession":
+        if self.enabled:
+            self._t0 = time.monotonic()
+            if self.title:
+                self.add_step(self.title, duration=3.0)
+            self._start_ffmpeg_capture()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        if self.enabled:
+            self._stop_ffmpeg_capture()
+            await self.finalize()
+
+    async def finalize(self) -> None:
+        """停止录制后合成最终 MP4；失败只记 warning，不抛异常。"""
+        if not self.enabled:
+            self.artifacts = self.summary()
+            return
+
+        if not self._steps and self.title:
+            self.add_step(self.title)
+
+        try:
+            capture = Path(self._capture_path)
+            if not capture.is_file() or capture.stat().st_size <= 0:
+                self.warnings.append("ffmpeg_capture_missing")
+                self.artifacts = self.summary()
+                return
+
+            self.capture_path = str(capture.resolve())
+
+            entries = list(self._steps)
+            if entries:
+                entries[-1].duration = max(entries[-1].duration, 5.0)
+            try:
+                _generate_srt(entries, self._srt_path)
+            except Exception as exc:
+                self.warnings.append(f"subtitle_write_failed: {exc}")
+
+            music = _resolve_music_file()
+            if music is None:
+                self.warnings.append("background_music_missing")
+
+            srt_file = Path(self._srt_path)
+            use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
+
+            ok, err = _compose_capture_mp4(
+                capture,
+                srt_file if use_srt else Path(self._srt_path),
+                Path(self._planned_output),
+                music_file=music,
+            )
+            if not ok and music is not None:
+                self.warnings.append(f"ffmpeg_with_music_failed: {err}")
+                ok, err = _compose_capture_mp4(
+                    capture,
+                    srt_file if use_srt else Path(self._srt_path),
+                    Path(self._planned_output),
+                    music_file=None,
+                )
+            if not ok and not use_srt:
+                ok, err = _run_ffmpeg(
+                    [
+                        "ffmpeg",
+                        "-y",
+                        "-i",
+                        str(capture),
+                        "-c:v",
+                        "libx264",
+                        "-preset",
+                        "fast",
+                        "-crf",
+                        "23",
+                        "-pix_fmt",
+                        "yuv420p",
+                        str(self._planned_output),
+                    ]
+                )
+            if not ok:
+                self.warnings.append(f"ffmpeg_compose_failed: {err}")
+                self.artifacts = self.summary()
+                return
+
+            self.output_video_path = str(Path(self._planned_output).resolve())
+        except Exception as exc:
+            self.warnings.append(f"video_finalize_error: {exc}")
+
+        self.artifacts = self.summary()
+
+    def summary(self) -> Dict[str, Any]:
+        return {
+            "enabled": self.enabled,
+            "path": self.output_video_path,
+            "capture_path": self.capture_path,
+            "warnings": list(self.warnings),
+        }
+
+    def to_artifact_dict(self) -> Dict[str, Any]:
+        data = self.summary()
+        data["batch_id"] = self.batch_id
+        return data

scripts/service/example_adapter/__init__.py

@@ -0,0 +1,26 @@
+"""示例 adapter 四档 dispatch。
+
+由 ``OPENCLAW_TEST_TARGET``（或 ``config.get``）决定档位。
+复制本目录后把 ``example_adapter`` 改名为 ``<domain>_adapter``。
+"""
+
+from __future__ import annotations
+
+import os
+
+from .mock import MockAdapter
+from .real_api import RealApiAdapter
+from .real_rpa import RealRpaAdapter
+from .sim_rpa import SimRpaAdapter
+
+
+def get_adapter():
+    """返回当前档位 adapter 实例。"""
+    target = (os.environ.get("OPENCLAW_TEST_TARGET") or "simulator_rpa").strip().lower()
+    if target in ("unit", "mock"):
+        return MockAdapter()
+    if target == "real_api":
+        return RealApiAdapter()
+    if target == "real_rpa":
+        return RealRpaAdapter()
+    return SimRpaAdapter()

scripts/service/example_adapter/base.py

@@ -0,0 +1,37 @@
+"""示例适配器数据契约与基类。
+
+复制本目录为 ``<domain>_adapter/``，按 ADAPTER.md 实现各档位。
+业务逻辑只依赖 ``base`` 里的接口，不感知 mock / sim_rpa / real_* 差异。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class ExampleItem:
+    """单条业务输入（示例）。"""
+
+    id: str
+    label: str
+
+
+@dataclass
+class ExampleResult:
+    """批量操作结果（以批为单位返回）。"""
+
+    ok: bool
+    serial_no: Optional[str]
+    error_msg: Optional[str]
+    artifacts: Dict[str, Any] = field(default_factory=dict)
+
+
+class AdapterBase:
+    """四档 adapter 统一接口；子类实现 ``do_batch``。"""
+
+    name = "base"
+
+    def do_batch(self, target: str, items: List[ExampleItem]) -> ExampleResult:
+        raise NotImplementedError

scripts/service/example_adapter/mock.py

@@ -0,0 +1,24 @@
+"""离线 mock 档位：纯内存仿真，CI / 单测必跑。"""
+
+from __future__ import annotations
+
+from .base import AdapterBase, ExampleItem, ExampleResult
+
+
+class MockAdapter(AdapterBase):
+    name = "mock"
+
+    def do_batch(self, target: str, items: list[ExampleItem]) -> ExampleResult:
+        if not items:
+            return ExampleResult(
+                ok=False,
+                serial_no=None,
+                error_msg="empty batch",
+                artifacts={},
+            )
+        return ExampleResult(
+            ok=True,
+            serial_no=f"MOCK-{len(items)}",
+            error_msg=None,
+            artifacts={"mode": "mock", "target": target, "count": len(items)},
+        )

scripts/service/example_adapter/real_api.py

@@ -0,0 +1,14 @@
+"""真实 API 档位占位：有官方接口时在此实现。"""
+
+from __future__ import annotations
+
+from .base import AdapterBase, ExampleItem, ExampleResult
+
+
+class RealApiAdapter(AdapterBase):
+    name = "real_api"
+
+    def do_batch(self, target: str, items: list[ExampleItem]) -> ExampleResult:
+        raise NotImplementedError(
+            "RealApiAdapter: 复制 example_adapter 后在此对接真实 API"
+        )

scripts/service/example_adapter/real_rpa.py

@@ -0,0 +1,14 @@
+"""真实 RPA 档位占位：无 API 时最后手段，谨慎实现。"""
+
+from __future__ import annotations
+
+from .base import AdapterBase, ExampleItem, ExampleResult
+
+
+class RealRpaAdapter(AdapterBase):
+    name = "real_rpa"
+
+    def do_batch(self, target: str, items: list[ExampleItem]) -> ExampleResult:
+        raise NotImplementedError(
+            "RealRpaAdapter: 复制 example_adapter 后在此实现生产界面 RPA"
+        )

scripts/service/example_adapter/sim_rpa.py

@@ -0,0 +1,108 @@
+"""仿真 RPA 档位：演示 RpaVideoSession + launch_persistent_browser（需浏览器，默认开发档）。"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from datetime import datetime
+
+from jiangchang_skill_core import config
+from jiangchang_skill_core.rpa import anti_detect, artifacts, errors, launch_persistent_browser
+from jiangchang_skill_core.rpa.human_login import wait_for_captcha_pass
+from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+
+from util.constants import SKILL_SLUG
+from util.runtime_paths import get_skill_data_dir, get_skill_root
+
+from .base import AdapterBase, ExampleItem, ExampleResult
+
+
+class SimRpaAdapter(AdapterBase):
+    name = "sim_rpa"
+
+    def do_batch(self, target: str, items: list[ExampleItem]) -> ExampleResult:
+        return asyncio.run(self._do_batch_async(target, items))
+
+    async def _do_batch_async(self, target: str, items: list[ExampleItem]) -> ExampleResult:
+        example_path = os.path.join(get_skill_root(), ".env.example")
+        config.ensure_env_file(SKILL_SLUG, example_path)
+
+        url = config.get("TARGET_BASE_URL") or target or "https://example.com"
+        batch_id = f"sim_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+        data_dir = get_skill_data_dir()
+        profile_dir = os.path.join(data_dir, "browser-profile")
+        os.makedirs(profile_dir, exist_ok=True)
+
+        try:
+            from playwright.async_api import async_playwright
+        except ImportError:
+            return ExampleResult(
+                ok=False,
+                serial_no=None,
+                error_msg="playwright not installed",
+                artifacts={},
+            )
+
+        artifact_paths: dict = {}
+        async with RpaVideoSession(
+            skill_slug=SKILL_SLUG,
+            skill_data_dir=data_dir,
+            batch_id=batch_id,
+            title="技能仿真演示",
+        ) as video:
+            video.add_step("正在打开目标页面")
+            try:
+                async with async_playwright() as pw:
+                    context = await launch_persistent_browser(
+                        pw,
+                        profile_dir=profile_dir,
+                        headless=config.get_bool("OPENCLAW_BROWSER_HEADLESS"),
+                    )
+                    page = context.pages[0] if context.pages else await context.new_page()
+                    await page.goto(url, wait_until="domcontentloaded", timeout=60_000)
+                    await anti_detect.human_mouse_wiggle(page)
+
+                    search = page.locator('input[type="search"], input[name="q"]').first
+                    if await search.count() > 0:
+                        demo_text = items[0].label if items else "openclaw-demo"
+                        video.step("正在填写演示内容")
+                        await anti_detect.human_type(page, search, demo_text)
+                        await anti_detect.human_delay_short()
+
+                    video.step("正在等待人工验证（如有）")
+                    if not await wait_for_captcha_pass(
+                        page,
+                        timeout_sec=config.get_int("HUMAN_WAIT_TIMEOUT", 180),
+                    ):
+                        shot = await artifacts.capture_failure(
+                            page, data_dir, batch_id, "captcha"
+                        )
+                        if shot:
+                            artifact_paths["captcha"] = shot
+                        return ExampleResult(
+                            ok=False,
+                            serial_no=None,
+                            error_msg=errors.ERR_CAPTCHA_NEED_HUMAN,
+                            artifacts={**artifact_paths, **video.to_artifact_dict()},
+                        )
+
+                    video.step("演示任务完成")
+                    await context.close()
+            except Exception as exc:
+                video.step("演示异常结束")
+                return ExampleResult(
+                    ok=False,
+                    serial_no=None,
+                    error_msg=str(exc),
+                    artifacts={**artifact_paths, **video.to_artifact_dict()},
+                )
+
+        arts = {"mode": "sim_rpa", "url": url, **artifact_paths, **video.to_artifact_dict()}
+        if video.output_video_path:
+            arts["video_path"] = video.output_video_path
+        return ExampleResult(
+            ok=True,
+            serial_no=f"SIM-{len(items)}",
+            error_msg=None,
+            artifacts=arts,
+        )

scripts/service/example_browser_rpa.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""浏览器 RPA 最小用法示例（复制到新 skill 后按需改）。
+
+演示标准流程：
+1. config.ensure_env_file 首次落盘 .env
+2. launch_persistent_browser 启动 stealth persistent context
+3. 拟人操作（mouse_wiggle / human_type / human_click）
+4. 验证码 HITL 等待
+5. 失败 capture_failure 截图
+
+运行（需系统 Chrome/Edge + playwright 包）：
+    python scripts/service/example_browser_rpa.py
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+
+_scripts_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _scripts_dir not in sys.path:
+    sys.path.insert(0, _scripts_dir)
+
+from jiangchang_skill_core import config
+from jiangchang_skill_core.rpa import (
+    anti_detect,
+    artifacts,
+    errors,
+    launch_persistent_browser,
+    wait_for_captcha_pass,
+)
+from util.constants import SKILL_SLUG
+from util.runtime_paths import get_skill_data_dir, get_skill_root
+
+
+async def demo() -> int:
+    example_path = os.path.join(get_skill_root(), ".env.example")
+    config.ensure_env_file(SKILL_SLUG, example_path)
+
+    url = config.get("TARGET_BASE_URL") or "https://example.com"
+    data_dir = get_skill_data_dir()
+    profile_dir = os.path.join(data_dir, "browser-profile")
+    os.makedirs(profile_dir, exist_ok=True)
+    batch_id = "demo"
+
+    try:
+        from playwright.async_api import async_playwright
+    except ImportError:
+        print("ERROR: playwright not installed")
+        return 1
+
+    async with async_playwright() as pw:
+        context = await launch_persistent_browser(
+            pw,
+            profile_dir=profile_dir,
+            headless=config.get_bool("OPENCLAW_BROWSER_HEADLESS"),
+        )
+        page = context.pages[0] if context.pages else await context.new_page()
+        try:
+            await page.goto(url, wait_until="domcontentloaded", timeout=60_000)
+            await anti_detect.human_mouse_wiggle(page)
+
+            search = page.locator('input[type="search"], input[name="q"]').first
+            if await search.count() > 0:
+                await anti_detect.human_type(page, search, "openclaw-rpa-demo")
+                await anti_detect.human_click(page, selector='input[type="search"]')
+
+            if not await wait_for_captcha_pass(
+                page,
+                timeout_sec=config.get_int("HUMAN_WAIT_TIMEOUT", 180),
+            ):
+                shot = await artifacts.capture_failure(page, data_dir, batch_id, "captcha")
+                print(errors.ERR_CAPTCHA_NEED_HUMAN, shot or "")
+                return 1
+
+            print("OK browser RPA demo finished")
+            return 0
+        except Exception as exc:
+            shot = await artifacts.capture_failure(page, data_dir, batch_id, "error")
+            print(f"ERROR: {exc}", shot or "")
+            return 1
+        finally:
+            await context.close()
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(demo()))

tests/README.md

@@ -39,6 +39,8 @@
 - [ ] 至少有 1 个缺必填字段 / 非法输入测试。
 - [ ] 如果有 adapter，至少覆盖 timeout / unauthorized / invalid response（可用 fake 模拟）。
 - [ ] 如果有解析 / 计算 / 校验类业务，至少保留 1 组 golden fixture（脱敏）。
+- [ ] 如果技能有 Python 三方依赖，`requirements.txt` 已声明且版本约束合理（尽量收窄范围）。
+- [ ] `health` 能在依赖缺失或原生运行库未就绪时给出清晰、可操作的错误（非裸 traceback）。
 - [ ] 真实 API / 真实 RPA 测试只放 `tests/integration/`，并且默认 `.sample`，不默认运行。
 
 ---
@@ -70,7 +72,7 @@ python tests/run_tests.py test_cli_smoke
 | **subprocess 真实入口** ``python scripts/main.py`` | ``test_entrypoint_subprocess.py`` |
 | 运行路径与 **CLAW_*/JIANGCHANG_*** 隔离 | ``test_runtime_paths.py`` |
 | ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` |
-| DB / ``publish_logs`` 骨架冒烟 | ``test_db_smoke.py`` |
+| DB / ``task_logs`` 骨架冒烟 | ``test_db_smoke.py`` |
 | **adapter profile / 外呼授权策略** | ``test_adapter_profile_policy.py`` + ``adapter_test_utils.py``（非 ``test_`` 前缀，不自动当用例收集） |
 
 ### 1.3 数据隔离（``_support.IsolatedDataRoot``）

tests/desktop/README.md

@@ -32,3 +32,32 @@ pytest tests/desktop/test_desktop_smoke.py -v -s
 - **桌面 E2E**：仅在本目录用 **pytest** 维护，二者互补、互不替代。
 
 若仅需验证 **service 业务逻辑**、契约与 golden 回归，**不要**写 desktop E2E；desktop 只用于验证**宿主 UI、路由、`jiangchang://` 协议、真实用户交互链路**等与桌面宿主强相关的行为。
+
+## 模板选择指南
+
+| 场景 | 用哪个模板 |
+|---|---|
+| 纯问答 skill（不上传附件、不跑长流程） | `test_desktop_smoke.py.sample` |
+| 需要上传附件 / RPA 长流程 / Agent 多轮工具调用 | `test_desktop_smoke_with_attachment.py.sample` |
+
+录屏入口都用 `record_screencast.py.sample`，里面 `TEST_FILE_NAME` 常量切换测试文件。
+
+## 已知 SDK 坑（来自 disburse-payroll-icbc 实战）
+
+1. **`client.send_file()` 在 v2.0.17+ 抛 NotImplementedError**——附件场景必须用
+   `jiangchang_desktop_sdk.e2e_helpers.drop_file_into_composer`（DataTransfer 拖拽）。
+2. **`client.ask()` 内部 `user_msg_index<0` 死循环**——如果匠厂启动慢 / gateway 握手有
+   延迟，导致 15s 内没抓到 user 节点，ask() 会锁死后续 900s 等待。建议长流程 / 附件场景
+   绕开 ask()，用 `jiangchang_desktop_sdk.e2e_helpers` 里的 `send_prompt_via_composer`
+   + `wait_for_user_message` + `wait_for_agent_complete` 自己控时。
+3. **字幕关键词建议用 ASCII**——虽然 platform-kit 已修了 subprocess 编码乱码，
+   保险起见字幕脚本里关键词用 `step: 1/` 这种 ASCII 锚而非中文短语。
+4. **`client.bring_to_front()` 已升级**——v?? 起会 ctypes ShowWindow(SW_MAXIMIZE) +
+   SetForegroundWindow，能真正铺满桌面。dev 模式下也能用（不依赖 jiangchang:// 协议）。
+
+## 录屏新能力（platform-kit v??）
+
+- `run_screencast(..., activate_window=True)` 自动激活+最大化匠厂窗口
+- `run_screencast(..., monitor_index=1)` 多显示器场景显式录主屏
+- 字幕脚本支持三元组 `(keyword, text, duration)` 给个别字幕指定独立时长，
+  覆盖长等待场景中间没日志输出的"字幕真空"

tests/desktop/conftest.py

@@ -2,11 +2,13 @@
 """
 可选桌面 E2E：pytest 收集时自动加载，负责把 jiangchang_desktop_sdk 加入 sys.path。
 
-查找顺序（与 content-manager 对齐思路，去掉业务仓专属路径假设）：
+查找顺序：
 1. 已 pip install 则直接 import；
-2. 环境变量 JIANGCHANG_KIT_ROOT → sdk/jiangchang-desktop-sdk/src；
+2. 环境变量 JIANGCHANG_KIT_ROOT → sdk/jiangchang-desktop-sdk/src（旧版布局）；
-3. 自本文件上溯：skill 根 → workspace 根 → OpenClaw 根，拼 jiangchang-platform-kit/sdk/.../src；
+3. 环境变量 JIANGCHANG_KIT_ROOT → src（新版 platform-kit 布局）；
-4. 都找不到则 pytest.exit 给出可操作的排障说明。
+4. 自本文件上溯：skill 根 → workspace 根 → OpenClaw 根，拼路径；
+5. 兜底 D:\\AI\\jiangchang-platform-kit\\src（若存在）；
+6. 都找不到则 pytest.exit 给出可操作的排障说明。
 """
 from __future__ import annotations
 
@@ -31,12 +33,22 @@ def _bootstrap_jiangchang_desktop_sdk() -> None:
     candidates = []
     if env_kit_root:
         candidates.append(os.path.join(env_kit_root, "sdk", "jiangchang-desktop-sdk", "src"))
+        candidates.append(os.path.join(env_kit_root, "src"))
     candidates.append(
         os.path.join(workspace_root, "jiangchang-platform-kit", "sdk", "jiangchang-desktop-sdk", "src")
     )
     candidates.append(
         os.path.join(open_claw_root, "jiangchang-platform-kit", "sdk", "jiangchang-desktop-sdk", "src")
     )
+    candidates.append(
+        os.path.join(workspace_root, "jiangchang-platform-kit", "src")
+    )
+    candidates.append(
+        os.path.join(open_claw_root, "jiangchang-platform-kit", "src")
+    )
+    default_kit_src = r"D:\AI\jiangchang-platform-kit\src"
+    if os.path.isdir(default_kit_src):
+        candidates.append(default_kit_src)
 
     tried = []
     for cand in candidates:
@@ -49,10 +61,10 @@ def _bootstrap_jiangchang_desktop_sdk() -> None:
 
     pytest.exit(
         "无法定位 jiangchang_desktop_sdk。请任选一种方式解决：\n"
-        "  1) pip install -e <path-to>/jiangchang-platform-kit/sdk/jiangchang-desktop-sdk\n"
+        "  1) pip install -e <path-to>/jiangchang-platform-kit\n"
         "  2) 设置环境变量 JIANGCHANG_KIT_ROOT=<path-to>/jiangchang-platform-kit\n"
-        "  3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置，例如：\n"
+        "     （SDK 新版位置：<root>\\src；旧版：<root>\\sdk\\jiangchang-desktop-sdk\\src）\n"
-        f"     {workspace_root}/jiangchang-platform-kit/ 或 {open_claw_root}/jiangchang-platform-kit/\n"
+        "  3) 将 jiangchang-platform-kit 与本仓库放在同一 workspace 父级可推断的位置\n"
         "已尝试的候选路径：\n  - " + "\n  - ".join(tried),
         returncode=4,
     )

tests/desktop/record_screencast.py.sample

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""
+Screencast 录制入口（模板）。
+
+启用方式：
+  1. 复制为 record_screencast.py
+  2. 修改 SKILL_SLUG 和 SUBTITLE_SCRIPT
+  3. 决定 test_file 指向哪个测试（默认 test_desktop_smoke.py；
+     有附件场景的用 test_desktop_smoke_with_attachment.py）
+  4. 运行：python tests/desktop/record_screencast.py
+
+依赖：
+  1. pip install mss
+  2. 系统已安装 ffmpeg 并在 PATH 中
+  3. jiangchang-platform-kit 可推断路径（见 _bootstrap 逻辑）
+  4. MEDIA_ASSETS_ROOT 环境变量指向 media-assets 仓库（或使用默认路径）
+
+平台 kit v?? 之后 run_screencast 自动激活并最大化匠厂窗口、支持指定显示器、
+字幕支持每条独立时长。本模板已经利用这些能力，新 skill 复制后无需再内嵌
+ctypes 激活代码。
+"""
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+
+def _bootstrap_screencast() -> None:
+    """将 jiangchang-platform-kit/tools 加入 sys.path，使 screencast 包可导入。"""
+    here = Path(__file__).resolve()
+    skill_root = here.parents[2]
+    workspace_root = skill_root.parent
+    open_claw_root = workspace_root.parent
+
+    candidates = [
+        workspace_root / "jiangchang-platform-kit" / "tools",
+        open_claw_root / "jiangchang-platform-kit" / "tools",
+        Path(r"D:\AI\jiangchang-platform-kit") / "tools",
+    ]
+    env_kit = os.environ.get("JIANGCHANG_KIT_ROOT", "")
+    if env_kit:
+        candidates.insert(0, Path(env_kit) / "tools")
+
+    for cand in candidates:
+        if (cand / "screencast" / "__init__.py").exists():
+            if str(cand) not in sys.path:
+                sys.path.insert(0, str(cand))
+            return
+
+    raise RuntimeError(
+        "找不到 jiangchang-platform-kit/tools/screencast。\n"
+        "请设置 JIANGCHANG_KIT_ROOT 环境变量，或将 platform-kit 放在 workspace "
+        "同级目录。"
+    )
+
+
+_bootstrap_screencast()
+
+from screencast import run_screencast  # noqa: E402
+
+# ============================================================
+# ↓↓↓ 复制后只需修改下面这些 ↓↓↓
+# ============================================================
+
+SKILL_SLUG = "your-skill-slug"   # 替换为本技能 slug
+
+# 字幕脚本设计要点（来自 disburse-payroll-icbc 实战经验）：
+# 1. **关键词全用 ASCII**：保险起见绕开 Windows cp936/UTF-8 编码偶发问题。
+#    虽然 platform-kit 已修了 subprocess 编码，ASCII 关键词仍是最稳的。
+# 2. **用 "step: N/X" 模式作触发锚**：在测试代码里加 _logger.info("step: 1/5 ...")
+#    日志作为字幕触发点，能精确控制字幕出现时机，比依赖 SDK 内部日志更可靠。
+# 3. **关键词每条只触发一次**，所以"step: 1/" 这种唯一前缀比"准备数据"更稳。
+# 4. **三元组支持每条独立时长**：(关键词, 文案, 显示秒数)；
+#    long wait 场景可以让某条字幕停留更久（如 30s）覆盖中间无日志输出的等待期。
+SUBTITLE_SCRIPT = [
+    # (keyword, text)  二元组用默认时长（5s）
+    ("ensure_app_running",  "启动匠厂智能体平台"),
+    ("wait_gateway_ready",  "网关就绪"),
+    ("step: 1/",            "准备测试数据"),
+    ("step: 2/",            "执行技能操作"),
+    # (keyword, text, duration)  三元组指定独立时长
+    ("step: 3/",            "技能在后端持续执行，请稍候...", 30.0),
+    ("step: 4/",            "结果汇总返回"),
+    ("PASSED",              "测试通过 ✓"),
+]
+
+# 测试文件选择（按需切换）：
+#   - test_desktop_smoke.py: 纯问答 skill 用这个
+#   - test_desktop_smoke_with_attachment.py: 需要上传附件的 skill 用这个
+TEST_FILE_NAME = "test_desktop_smoke.py"
+
+# 音乐子目录（按 skill 风格选）：
+#   - "music/corporate": 商务严肃（金融、办公场景）
+#   - "music/calm": 舒缓（流程演示、配置类）
+#   - "music/upbeat": 活力（成功汇报、营销视频）
+#   - "music"（默认）: 三个子目录混选
+MUSIC_SUBDIR = "music/corporate"
+
+# ============================================================
+
+if __name__ == "__main__":
+    _here = Path(__file__).resolve()
+    test_file = str(_here.parent / TEST_FILE_NAME)
+    output_dir = str(_here.parent / "artifacts" / "recordings")
+
+    run_screencast(
+        skill_slug=SKILL_SLUG,
+        subtitle_script=SUBTITLE_SCRIPT,
+        pytest_args=[test_file, "-v", "-s"],
+        output_dir=output_dir,
+        music_subdir=MUSIC_SUBDIR,
+        # platform-kit 自动激活+最大化窗口（Windows ctypes / 其他系统协议），
+        # 想关掉传 activate_window=False
+        # monitor_index=1 显式录主屏（多显示器场景推荐）
+    )

tests/desktop/test_desktop_smoke.py.sample

@@ -1,12 +1,29 @@
 # -*- coding: utf-8 -*-
 """
-可选桌面 E2E 示例（默认不执行：文件名后缀为 .sample）。
+可选桌面 E2E 示例（默认不执行：文件名后缀为 .sample）—— **简单问答场景**。
 
 启用方式：
   复制本文件为 test_desktop_smoke.py，按目标技能修改提问文案与断言。
   运行：pytest tests/desktop/test_desktop_smoke.py -v -s
 
 依赖：pytest、jiangchang_desktop_sdk、匠厂桌面宿主已就绪。
+
+【何时用此模板】
+- 技能只需要发一句话给 Main Agent，等回复，做断言（纯问答）
+- 不涉及上传附件 / 不涉及 RPA 长流程
+
+【何时不要用此模板，改用 test_desktop_smoke_with_attachment.py.sample】
+- 需要上传文件（Excel / PDF / 图片）作为附件
+- 技能内部跑 RPA / 长流程（> 2 分钟），需要精细等待
+
+【已知 SDK 限制（来自 disburse-payroll-icbc 实战）】
+- client.ask() 内部对"找到 user 消息节点"只给 15 秒窗口，找不到就锁死
+  user_msg_index=-1，后续等待 assistant 永远 return None。如果你的场景里
+  匠厂启动较慢 / gateway 握手有延迟，client.ask() 可能假死到 timeout。
+  此模板仍用 ask()，因为它对**简单场景**够用。复杂场景请用 _with_attachment.py.sample
+  里的"绕开 ask()"模式。
+- client.send_file() 在 v2.0.17+ 抛 NotImplementedError，**绝不要调用它**。
+  需要附件请用 _with_attachment.py.sample 里的拖拽方案。
 """
 from __future__ import annotations
 

tests/desktop/test_desktop_smoke_with_attachment.py.sample

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+"""
+可选桌面 E2E 示例（默认不执行：文件名后缀为 .sample）—— **附件上传 + 长流程场景**。
+
+启用方式：
+  1. 复制为 test_desktop_smoke_with_attachment.py
+  2. 修改 SKILL_SLUG / PROMPT / 断言为你技能的真实业务
+  3. 准备样例附件，或动态生成（参考 _prepare_sample_attachment）
+  4. 运行：pytest tests/desktop/test_desktop_smoke_with_attachment.py -v -s
+
+依赖：
+  - pytest / jiangchang_desktop_sdk / 匠厂桌面宿主已就绪
+  - 通过 conftest.py 自动 bootstrap，jiangchang_desktop_sdk.e2e_helpers 模块在 sys.path
+
+【为什么用此模板而不是 test_desktop_smoke.py.sample】
+- client.send_file() 在 v2.0.17+ 抛 NotImplementedError，必须用拖拽方式
+- client.ask() 在 user_idx<0 时会锁死（已知 SDK 缺陷），本模板绕开 ask()，
+  在测试侧自己用 jiangchang_desktop_sdk.e2e_helpers 做等待
+
+【字幕配合】
+- 关键节点都打了 _logger.info("step: N/7 ...")，对应 record_screencast.py 的
+  SUBTITLE_SCRIPT 用 step: 关键词触发字幕
+"""
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+import pytest
+
+from jiangchang_desktop_sdk import JiangchangDesktopClient
+from jiangchang_desktop_sdk.e2e_helpers import (
+    drop_file_into_composer,
+    send_prompt_via_composer,
+    wait_for_agent_complete,
+    wait_for_attachment_ready,
+    wait_for_composer_enabled,
+    wait_for_user_message,
+)
+
+# ============================================================
+# ↓↓↓ 复制后修改这些 ↓↓↓
+# ============================================================
+
+SKILL_SLUG = "your-skill-slug"
+
+PROMPT = (
+    "请使用 /your-skill-slug 处理我刚才上传的文件。"
+    "完成后请用清单方式汇总：[改成你技能要求的字段]。"
+)
+
+# 业务断言：根据你的技能输出修改
+EXPECTED_KEYWORDS_IN_REPLY = ["关键词1", "关键词2"]
+
+# ============================================================
+
+_logger = logging.getLogger("desktop-e2e-with-attachment-sample")
+if not _logger.handlers:
+    _logger.setLevel(logging.INFO)
+    _h = logging.StreamHandler()
+    _h.setFormatter(logging.Formatter("[E2E] %(message)s"))
+    _logger.addHandler(_h)
+
+_ARTIFACT_DIR = os.path.join(os.path.dirname(__file__), "artifacts")
+
+
+def _prepare_sample_attachment(tmp_path: Path) -> Path:
+    """在 tmp_path 下生成一个样例附件（按 skill 真实需要修改）。
+
+    示例：生成一个 5 行的 xlsx。你的 skill 可能需要 CSV / PDF / 图片等其他格式。
+    """
+    from openpyxl import Workbook  # noqa: WPS433
+    xlsx = tmp_path / "sample.xlsx"
+    wb = Workbook()
+    ws = wb.active
+    ws.append(["列1", "列2", "列3"])
+    for i in range(1, 6):
+        ws.append([f"值{i}A", f"值{i}B", i * 100])
+    wb.save(str(xlsx))
+    return xlsx
+
+
+@pytest.fixture
+def client(request):
+    c = JiangchangDesktopClient()
+    _logger.info("ensure_app_running")
+    c.ensure_app_running(wait_timeout_s=45)
+    c.bring_to_front()
+    c.wait_gateway_ready(timeout_ms=30000)
+    yield c
+    try:
+        rep = getattr(request.node, "rep_call", None)
+        if rep is not None and rep.failed:
+            os.makedirs(_ARTIFACT_DIR, exist_ok=True)
+            paths = c.snapshot(_ARTIFACT_DIR, tag=request.node.name)
+            _logger.error("failure snapshot: %s", paths)
+    finally:
+        c.disconnect()
+
+
+def test_desktop_with_attachment(client: JiangchangDesktopClient, tmp_path: Path) -> None:
+    _logger.info("step: 1/7 准备样例附件")
+    attachment_path = _prepare_sample_attachment(tmp_path)
+
+    client.new_task()
+    page = client.get_page()
+    client.bring_to_front()
+
+    _logger.info("step: 2/7 拖拽附件到聊天框")
+    drop_file_into_composer(page, attachment_path)
+
+    _logger.info("step: 3/7 等附件 stage-buffer 完成")
+    wait_for_attachment_ready(page, attachment_path.name)
+
+    wait_for_composer_enabled(page)
+    client.bring_to_front()
+    _logger.info("step: 4/7 键入并发送指令")
+    send_prompt_via_composer(page, PROMPT)
+
+    _logger.info("step: 5/7 Main Agent 接收，等待路由到技能")
+    user_idx = wait_for_user_message(page, PROMPT, timeout_s=60.0)
+
+    _logger.info("step: 6/7 技能在后端执行（可能 2-3 分钟）")
+    client.bring_to_front()
+    reply = wait_for_agent_complete(
+        page, user_idx,
+        overall_timeout_s=1200.0,
+        stable_seconds=5.0,
+    )
+
+    _logger.info("step: 7/7 校验回复内容")
+    _logger.info("reply length=%d", len(reply))
+
+    assert reply, "assistant 回复为空"
+    for kw in EXPECTED_KEYWORDS_IN_REPLY:
+        assert kw in reply, f"期望关键词 {kw!r} 未出现在回复中；片段：{reply[:400]!r}"