加入匠厂桌面测试

Made-with: Cursor

.github/workflows/reusable-release-frontend.yaml

@@ -0,0 +1,115 @@
+# Standalone reusable workflow for Node/Vite static sites.
+# Does not reference reusable-release-skill.yaml; skill workflows remain unchanged.
+#
+# Node：请在 Runner 上预装 Node 20+（勿用 actions/setup-node，避免每次从 GitHub 拉包）。
+# npm：通过 npm_registry 输入使用国内镜像（默认 npmmirror）。
+# Checkout：不用 actions/checkout（host 模式下 JS Action 仍会进无 Node 的容器）；改用 git + github.token。
+#   重要：act_runner 会把 run: 脚本写到 $GITHUB_WORKSPACE/workflow/*.sh，
+#   所以 checkout 绝对不能在 workspace 根上 rm -rf ./*，否则会把自己这个脚本一起删了。
+#   解决办法：克隆到 $GITHUB_WORKSPACE/_src 子目录，后续 step 都在该子目录操作。
+# 浅拉不要用「裸 SHA」作 fetch 参数（Gitea 上易失败）；按分支 clone 再对齐 SHA。
+
+name: Reusable Frontend Deploy
+
+on:
+  workflow_call:
+    inputs:
+      deploy_path:
+        description: "Target directory for static files (trailing slash optional)"
+        required: false
+        type: string
+        default: "/www/wwwroot/sandbox/web"
+      build_command:
+        required: false
+        type: string
+        default: "npm run build:jc2009"
+      npm_registry:
+        description: "npm registry (e.g. npmmirror for CN)"
+        required: false
+        type: string
+        default: "https://registry.npmmirror.com"
+      runs_on:
+        description: "Runner label; must match a registered runner"
+        required: false
+        type: string
+        default: "ubuntu-latest"
+      chown_www:
+        description: "Run chown -R www:www after sync (requires permission on runner)"
+        required: false
+        type: boolean
+        default: true
+
+jobs:
+  build-and-deploy:
+    runs-on: ${{ inputs.runs_on }}
+    defaults:
+      run:
+        shell: bash
+        working-directory: ${{ github.workspace }}/_src
+    env:
+      NPM_CONFIG_REGISTRY: ${{ inputs.npm_registry }}
+    steps:
+      - name: Checkout
+        # 这一步在 workspace 根执行，不能用默认的 _src（此时还不存在）
+        working-directory: ${{ github.workspace }}
+        env:
+          GITEA_HOST: git.jc2009.com
+          GITEA_TOKEN: ${{ github.token }}
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          set -euo pipefail
+          git config --global --add safe.directory '*'
+          REPO="${{ github.repository }}"
+          SHA="${{ github.sha }}"
+          BRANCH="${{ github.ref_name }}"
+          TOKEN="${GITEA_TOKEN:-${GITHUB_TOKEN:-}}"
+          test -n "$TOKEN" || { echo "Checkout: empty token (github.token not available to this job)"; exit 1; }
+          URL="https://x-access-token:${TOKEN}@${GITEA_HOST}/${REPO}.git"
+
+          SRC_DIR="${GITHUB_WORKSPACE}/_src"
+          # 只清理 _src，不动 workspace 根（保护 runner 写入的 workflow/*.sh）
+          rm -rf "$SRC_DIR"
+          git clone --depth 1 --branch "$BRANCH" "$URL" "$SRC_DIR"
+          cd "$SRC_DIR"
+          if [ "$(git rev-parse HEAD)" != "$SHA" ]; then
+            git fetch --depth 200 origin "refs/heads/${BRANCH}"
+            git checkout --force "$SHA"
+          fi
+          echo "Checked out $(git rev-parse HEAD) to $SRC_DIR"
+
+      - name: Check Node.js (pre-installed on runner)
+        run: |
+          set -e
+          command -v node >/dev/null || { echo "Runner 上未找到 node，请先安装 Node 20+"; exit 1; }
+          command -v npm  >/dev/null || { echo "Runner 上未找到 npm"; exit 1; }
+          node -v
+          npm  -v
+          NODE_MAJOR=$(node -p "process.versions.node.split('.')[0]")
+          test "$NODE_MAJOR" -ge 20 || { echo "需要 Node 20 及以上，当前: $(node -v)"; exit 1; }
+
+      - name: Install dependencies
+        run: npm ci
+
+      - name: Build
+        run: ${{ inputs.build_command }}
+
+      - name: Deploy (sync dist to web root)
+        run: |
+          set -euo pipefail
+          test -d dist
+          DEST="${{ inputs.deploy_path }}"
+          DEST="${DEST%/}"
+          mkdir -p "$DEST"
+          # 宝塔面板会自动生成 .user.ini(并加 immutable 属性)和 .htaccess,跳过它们
+          find "$DEST" -mindepth 1 -maxdepth 1 \
+            ! -name ".user.ini" \
+            ! -name ".htaccess" \
+            -exec rm -rf {} +
+          cp -a dist/. "$DEST/"
+
+      - name: Set ownership for Nginx (Baota www)
+        run: |
+          if [ "${{ inputs.chown_www }}" != "true" ]; then exit 0; fi
+          DEST="${{ inputs.deploy_path }}"
+          DEST="${DEST%/}"
+          chown -R www:www "$DEST" || chown -R nginx:nginx "$DEST" || true

.github/workflows/reusable-release-skill.yaml

@@ -3,6 +3,11 @@ name: Reusable Skill Release
 on:
   workflow_call:
     inputs:
+      runs_on:
+        description: "Runner label; must match a registered runner (use host runner for pip/python on same machine as Node frontend)"
+        required: false
+        type: string
+        default: ubuntu-latest
       artifact_platform:
         required: false
         type: string
@@ -30,23 +35,30 @@ on:
 
 jobs:
   build-and-deploy:
-    runs-on: ubuntu-latest
+    runs-on: ${{ inputs.runs_on }}
+    defaults:
+      run:
+        shell: bash
     env:
       ARTIFACT_PLATFORM: ${{ inputs.artifact_platform }}
       PYARMOR_PLATFORM: ${{ inputs.pyarmor_platform }}
       PIP_BREAK_SYSTEM_PACKAGES: "1"
+      # Prefer self-built Python 3.12 under /usr/local (Alibaba Cloud Linux host); keep system paths as fallback.
+      # 显式前缀，避免部分 Runner 未注入 env.PATH 时丢失系统路径
+      PATH: /usr/local/bin:/usr/local/python3.12/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin
       # PyArmor 交叉平台加密时会内部执行 pip 安装 pyarmor.cli.core.* 等包；不设则默认走 files.pythonhosted.org，国内 CI 易超时。
       PIP_INDEX_URL: https://pypi.tuna.tsinghua.edu.cn/simple
       PIP_EXTRA_INDEX_URL: https://mirrors.aliyun.com/pypi/simple https://mirrors.cloud.tencent.com/pypi/simple https://mirrors.huaweicloud.com/repository/pypi/simple
       PIP_DEFAULT_TIMEOUT: "180"
       PIP_TRUSTED_HOST: pypi.tuna.tsinghua.edu.cn mirrors.aliyun.com mirrors.cloud.tencent.com mirrors.huaweicloud.com files.pythonhosted.org pypi.org
     steps:
-      - uses: http://120.25.191.12:3000/admin/actions-checkout@v4
+      - uses: https://git.jc2009.com/admin/actions-checkout@v4
 
       # Pin PyArmor 8.5.3 — matches desktop bundles; 9.x trial is stricter in CI。
       # 镜像由 job env（PIP_INDEX_URL / PIP_EXTRA_INDEX_URL）统一指定，与 Encrypt 步骤中 PyArmor 内部 pip 一致。
+      # 使用 python3.12 -m pip，避免仅存在 python3(3.6) 或裸 pip 不在 PATH 的宿主机/容器。
       - name: Setup Tools
-        run: pip install "pyarmor==8.5.3" requests python-frontmatter --break-system-packages
+        run: python3.12 -m pip install "pyarmor==8.5.3" requests python-frontmatter --break-system-packages
 
       - name: Register PyArmor (optional)
         env:
@@ -55,12 +67,13 @@ jobs:
           if [ -z "${PYARMOR_REG_B64}" ]; then
             echo "PyArmor: no PYARMOR_REG_B64 secret — trial mode (very large single .py modules may fail to obfuscate)."
           else
-            python -c "import os,base64,pathlib,subprocess; p=pathlib.Path('/tmp/pyarmor-reg.zip'); p.write_bytes(base64.standard_b64decode(os.environ['PYARMOR_REG_B64'])); subprocess.run(['pyarmor','reg',str(p)],check=True); p.unlink(missing_ok=True)"
+            python3.12 -c "import os,base64,pathlib,subprocess; os.environ['PATH']='/usr/local/bin:/usr/local/python3.12/bin:'+os.environ.get('PATH',''); p=pathlib.Path('/tmp/pyarmor-reg.zip'); p.write_bytes(base64.standard_b64decode(os.environ['PYARMOR_REG_B64'])); subprocess.run(['pyarmor','reg',str(p)],check=True); p.unlink(missing_ok=True)"
           fi
 
       # 递归加密整个 scripts/（含 cli、service、db、util 等子包）；产物保留与源码一致的 scripts/ 层级，入口为 scripts/main.py。
       - name: Encrypt Source Code
         run: |
+          export PATH="/usr/local/bin:/usr/local/python3.12/bin:${PATH:-}"
           mkdir -p dist/package/scripts
           set -euo pipefail
           test -d scripts
@@ -72,11 +85,30 @@ jobs:
       - name: Parse Metadata and Pack
         id: build_task
         run: |
-          python -c "
+          python3.12 -c "
           import frontmatter, os, json, shutil
           post = frontmatter.load('SKILL.md')
           metadata = dict(post.metadata or {})
-          metadata['readme_md'] = (post.content or '').strip()
+          skill_readme_md = (post.content or '').strip()
+          skill_description = metadata.get('description')
+          metadata['readme_md'] = skill_readme_md
+          readme_path = os.path.join('references', 'README.md')
+          if os.path.isfile(readme_path):
+              try:
+                  readme_post = frontmatter.load(readme_path)
+                  body = (readme_post.content or '').strip()
+                  rm_meta = readme_post.metadata or {}
+                  desc = rm_meta.get('description')
+                  if desc is not None and str(desc).strip():
+                      metadata['description'] = str(desc).strip()
+                  if body:
+                      metadata['readme_md'] = body
+              except Exception:
+                  pass
+          if not (metadata.get('readme_md') or '').strip():
+              metadata['readme_md'] = skill_readme_md
+          if skill_description is not None and not str(metadata.get('description', '') or '').strip():
+              metadata['description'] = skill_description
           openclaw_meta = metadata.get('metadata', {}).get('openclaw', {})
           slug = (openclaw_meta.get('slug') or metadata.get('slug') or metadata.get('name') or '').strip()
           if not slug:
@@ -102,7 +134,7 @@ jobs:
           METADATA_JSON: ${{ steps.build_task.outputs.metadata }}
           SYNC_URL: ${{ inputs.sync_url }}
         run: |
-          python -c "
+          python3.12 -c "
           import requests, json, os
           metadata = json.loads(os.environ['METADATA_JSON'])
           res = requests.post(os.environ['SYNC_URL'], json=metadata)
@@ -121,7 +153,7 @@ jobs:
           ARTIFACT_PLATFORM: ${{ steps.build_task.outputs.artifact_platform }}
           UPLOAD_URL: ${{ inputs.upload_url }}
         run: |
-          python -c "
+          python3.12 -c "
           import requests, os
           slug = os.environ['SLUG']
           version = os.environ['VERSION']
@@ -148,7 +180,7 @@ jobs:
           VERSION: ${{ steps.build_task.outputs.version }}
           PRUNE_URL: ${{ inputs.prune_url }}
         run: |
-          python -c "
+          python3.12 -c "
           import requests, os
           payload = {
             'name': os.environ['SLUG'],

python-runtime/README.md

@@ -9,7 +9,8 @@
 `{JIANGCHANG_DATA_ROOT}/python-runtime/`
 
 并在该目录执行 `uv sync` 生成 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv/`。  
-Playwright 浏览器缓存目录：`{JIANGCHANG_DATA_ROOT}/playwright-browsers/`（由宿主设置 `PLAYWRIGHT_BROWSERS_PATH`）。
+
+**Playwright**：仅安装 Python 包 `playwright` 即可；技能通过 **`channel=chrome` / `msedge`** 使用用户本机已安装的 Chrome 或 Edge。**不要**在宿主流程里执行 `playwright install chromium`（避免下载约 170MB 的自带 Chromium）。
 
 ## 维护流程
 

sdk/jiangchang-desktop-sdk/pyproject.toml

@@ -0,0 +1,13 @@
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "jiangchang-desktop-sdk"
+version = "0.1.0"
+description = "匠厂桌面应用自动化测试 SDK"
+requires-python = ">=3.10"
+dependencies = ["playwright>=1.42.0"]
+
+[tool.setuptools.packages.find]
+where = ["src"]

sdk/jiangchang-desktop-sdk/src/jiangchang_desktop_sdk/__init__.py

@@ -0,0 +1,15 @@
+from .client import JiangchangDesktopClient
+from .types import JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
+from .exceptions import AppNotFoundError, TimeoutError, AssertError
+
+__all__ = [
+    "JiangchangDesktopClient",
+    "JiangchangMessage",
+    "AskOptions",
+    "LaunchOptions", 
+    "AssertOptions",
+    "AppNotFoundError",
+    "TimeoutError",
+    "AssertError",
+]
+__version__ = "0.1.0"

sdk/jiangchang-desktop-sdk/src/jiangchang_desktop_sdk/client.py

@@ -0,0 +1,588 @@
+"""
+JiangchangDesktopClient — 匠厂桌面应用自动化测试 SDK
+"""
+import os
+import time
+import json
+import urllib.request
+import signal
+import re
+from typing import Optional, List
+
+from playwright.sync_api import sync_playwright, Browser, Page
+
+from .types import JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
+from .exceptions import AppNotFoundError, ConnectionError, TimeoutError as JcTimeoutError, AssertError, LaunchError
+
+
+class JiangchangDesktopClient:
+    """
+    主要自动化客户端。
+    
+    用法示例：
+    
+    ```python
+    client = JiangchangDesktopClient()
+    # 连接到已运行的匠厂桌面应用（通过 CDP，端口 9222）
+    client.connect()
+    
+    # 提问并等待回复
+    answer = client.ask("帮我计算，上海到洛杉矶一个40HQ的海运费")
+    print(answer)
+    
+    # 断言回复内容
+    client.assert_contains("USD")
+    
+    client.disconnect()
+    ```
+    """
+
+    def __init__(self):
+        self._playwright = None          # playwright.sync_playwright 实例
+        self._browser: Optional[Browser] = None
+        self._page: Optional[Page] = None
+        self._connected = False
+        self._owns_browser = False       # 是否由我们启动的浏览器
+        self._electron_process = None    # subprocess.Popen 启动的 Electron 进程
+
+    def _get_default_executable_path(self) -> str:
+        """从环境变量获取 Electron 可执行文件路径"""
+        path = os.environ.get("JIANGCHANG_E2E_APP_PATH")
+        if not path:
+            raise AppNotFoundError(
+                "未找到 Electron 可执行文件路径。"
+                "请设置环境变量 JIANGCHANG_E2E_APP_PATH，"
+                "或调用 launch_app() 时传入 executable_path 参数。"
+            )
+        return path
+
+    def _get_default_cdp_port(self) -> int:
+        """从环境变量获取 CDP 端口，默认 9222"""
+        port = os.environ.get("JIANGCHANG_E2E_CDP_PORT", "9222")
+        return int(port)
+
+    def _discover_main_page_ws_url(self, cdp_port: int) -> str:
+        """
+        通过 CDP HTTP 端点发现匠厂主窗口的 WebSocket 调试 URL。
+        CDP HTTP 端点格式：http://localhost:9222/json 返回目标列表，
+        每个目标包含 webSocketDebuggerUrl 字段。
+        """
+        json_url = f"http://localhost:{cdp_port}/json"
+        try:
+            with urllib.request.urlopen(json_url, timeout=10) as resp:
+                targets = json.load(resp)
+        except urllib.error.URLError as exc:
+            raise ConnectionError(
+                f"无法访问 CDP 端点 {json_url}：{exc}。"
+                "请确认匠厂应用正在运行，并使用了 --remote-debugging-port=9222 参数。"
+            ) from exc
+
+        if not targets:
+            raise ConnectionError(
+                f"CDP 端点 {json_url} 没有找到任何调试目标。"
+                "请确认匠厂应用正在运行，并使用了 --remote-debugging-port=9222 参数。"
+            )
+
+        # 优先选择标题含"匠厂"、"OpenClaw"、"ClawX"的页面
+        for t in targets:
+            title = t.get("title", "")
+            if any(kw in title for kw in ["匠厂", "OpenClaw", "ClawX"]):
+                ws_url = t.get("webSocketDebuggerUrl")
+                if ws_url:
+                    return ws_url
+
+        # 没有匹配就用第一个可用目标
+        first = targets[0]
+        ws_url = first.get("webSocketDebuggerUrl")
+        if not ws_url:
+            raise ConnectionError("目标页面没有 WebSocket 调试 URL")
+        return ws_url
+
+    def connect(self, url: Optional[str] = None) -> None:
+        """
+        连接到已运行的匠厂桌面应用（通过 Chrome DevTools Protocol）。
+        
+        url: 可选，CDP HTTP 端点，默认 http://localhost:9222
+             也可通过环境变量 JIANGCHANG_E2E_CDP_PORT 设置端口
+        """
+        if self._connected:
+            return
+
+        cdp_port = self._get_default_cdp_port()
+        
+        try:
+            self._playwright = sync_playwright().start()
+        except Exception as exc:
+            raise LaunchError(f"启动 Playwright 失败: {exc}") from exc
+
+        try:
+            # 发现 WebSocket URL
+            ws_url = self._discover_main_page_ws_url(cdp_port)
+            
+            # 通过 WebSocket 连接到已有浏览器
+            self._browser = self._playwright.chromium.connect_over_cdp(ws_url)
+            
+            # 获取页面
+            contexts = self._browser.contexts
+            if not contexts:
+                raise ConnectionError("CDP 连接成功，但没有找到浏览器上下文")
+            
+            pages = contexts[0].pages
+            if not pages:
+                raise ConnectionError("CDP 连接成功，但没有找到页面")
+            
+            self._page = pages[0]
+            self._connected = True
+            self._owns_browser = False
+            
+        except ConnectionError:
+            self._cleanup()
+            raise
+        except Exception as exc:
+            self._cleanup()
+            raise ConnectionError(f"连接失败: {exc}") from exc
+
+    def launch_app(self, options: Optional[LaunchOptions] = None) -> None:
+        """
+        启动匠厂桌面应用（直接运行 Electron 可执行文件）。
+        
+        options: 启动选项。如果 executable_path 未指定，
+                 从环境变量 JIANGCHANG_E2E_APP_PATH 读取。
+        """
+        if self._connected:
+            self.disconnect()
+        
+        opts = options or LaunchOptions()
+        executable_path = opts.executable_path or self._get_default_executable_path()
+        cdp_port = opts.cdp_port
+        startup_timeout = opts.startup_timeout
+        
+        if not os.path.exists(executable_path):
+            raise AppNotFoundError(f"Electron 可执行文件不存在: {executable_path}")
+
+        try:
+            self._playwright = sync_playwright().start()
+        except Exception as exc:
+            raise LaunchError(f"启动 Playwright 失败: {exc}") from exc
+
+        try:
+            # 直接启动匠厂 Electron 程序，带远程调试端口
+            import subprocess
+            self._electron_process = subprocess.Popen(
+                [executable_path, f"--remote-debugging-port={cdp_port}"],
+                detached=True,
+                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
+            )
+            
+            # 等应用启动
+            time.sleep(3)
+            
+            # 再通过 CDP 连接上去
+            self.connect()
+            self._owns_browser = False
+            
+            # 等待页面加载
+            self._wait_for_page_ready(startup_timeout)
+            
+        except AppNotFoundError:
+            self._cleanup()
+            raise
+        except Exception as exc:
+            self._cleanup()
+            raise LaunchError(f"启动应用失败: {exc}") from exc
+
+    def _wait_for_page_ready(self, timeout_ms: int) -> None:
+        """等待页面有标题（说明主窗口已加载）"""
+        start = time.time()
+        deadline = start + timeout_ms / 1000
+        
+        while time.time() < deadline:
+            if self._page and self._page.title():
+                return
+            time.sleep(0.5)
+        
+        raise JcTimeoutError(f"页面加载超时（{timeout_ms}ms）")
+
+    def _cleanup(self) -> None:
+        """清理所有资源"""
+        # 如果是我们启动的 Electron 进程，先关闭它
+        if self._electron_process is not None:
+            try:
+                if os.name == "nt":
+                    # Windows：发送 CTRL_BREAK_EVENT 然后 terminate
+                    self._electron_process.send_signal(signal.CTRL_BREAK_EVENT if hasattr(signal, 'CTRL_BREAK_EVENT') else None)
+                    self._electron_process.wait(timeout=5)
+                self._electron_process.terminate()
+                self._electron_process.wait(timeout=5)
+            except Exception:
+                try:
+                    self._electron_process.kill()
+                except Exception:
+                    pass
+            self._electron_process = None
+
+        if self._page:
+            try:
+                self._page.close()
+            except Exception:
+                pass
+            self._page = None
+        
+        if self._browser:
+            try:
+                if self._owns_browser:
+                    self._browser.close()
+                else:
+                    self._browser.disconnect()
+            except Exception:
+                pass
+            self._browser = None
+        
+        if self._playwright:
+            try:
+                self._playwright.stop()
+            except Exception:
+                pass
+            self._playwright = None
+        
+        self._connected = False
+        self._electron_process = None
+
+    # ─── 辅助选择器 ───────────────────────────────────────────────
+
+    def _get_chat_input_locator(self):
+        """获取聊天输入框 locator，优先用 data-jcid，回退到类名/位置选择器"""
+        page = self.get_page()
+        # 优先：data-jcid="chat-input"
+        locator = page.locator('[data-jcid="chat-input"]')
+        if locator.count() > 0:
+            return locator
+        # Fallback：查找 textarea（通常在 chat 输入区）
+        locator = page.locator('textarea').last
+        if locator.count() > 0:
+            return locator
+        raise ConnectionError("找不到聊天输入框（textarea），请确认 data-jcid 属性已添加")
+
+    def _get_send_button_locator(self):
+        """获取发送按钮 locator"""
+        page = self.get_page()
+        locator = page.locator('[data-jcid="send-button"]')
+        if locator.count() > 0:
+            return locator
+        # Fallback：找 type="submit" 的 Button，或 textarea 同级的发送按钮
+        locator = page.locator('button[type="submit"]')
+        if locator.count() > 0:
+            return locator
+        # Fallback：找最后一个主要操作按钮（排除 icon-only 小按钮）
+        locator = page.locator('button:not([aria-label])').last
+        if locator.count() > 0:
+            return locator
+        raise ConnectionError("找不到发送按钮，请确认 data-jcid 属性已添加")
+
+    def _get_message_list_locator(self):
+        """获取消息列表容器 locator"""
+        page = self.get_page()
+        locator = page.locator('[data-jcid="message-list"]')
+        if locator.count() > 0:
+            return locator
+        # Fallback：找消息容器（class 含 space-y 或 max-w 的 div）
+        locator = page.locator('[class*="space-y-3"]')
+        if locator.count() > 0:
+            return locator
+        # Fallback：找 chat 主区域
+        locator = page.locator('main, [class*="chat"]').first
+        if locator.count() > 0:
+            return locator
+        raise ConnectionError("找不到消息列表容器")
+
+    # ─── 核心交互方法 ─────────────────────────────────────────────
+
+    def _wait_for_streaming_done(self, timeout: int = 120000) -> None:
+        """
+        等待 AI 流式输出完成。
+        
+        优先检测 window.__jc_sending__ 标志（需要 UI 注入），
+        回退到轮询消息列表内容变化。
+        """
+        page = self.get_page()
+
+        # 策略 1：window.__jc_sending__ 标志（由 UI runtime 注入）
+        try:
+            page.wait_for_function(
+                "() => window.__jc_sending__ === false",
+                timeout=timeout,
+            )
+            return
+        except Exception:
+            pass  # 标志不存在，尝试策略 2
+
+        # 策略 2：轮询消息列表内容，直到连续 3 次无变化（稳定）
+        last_content = ""
+        stable_count = 0
+        stable_threshold = 3
+        poll_interval = 2.0  # 秒
+        max_wait = timeout / 1000
+
+        start = time.time()
+        while time.time() - start < max_wait:
+            try:
+                msg_list = self._get_message_list_locator()
+                content = msg_list.inner_text()
+            except Exception:
+                content = ""
+
+            if content == last_content:
+                stable_count += 1
+                if stable_count >= stable_threshold:
+                    return  # 连续稳定则认为完成
+            else:
+                stable_count = 0
+                last_content = content
+
+            time.sleep(poll_interval)
+
+        raise JcTimeoutError(f"等待 AI 回复超时（{timeout}ms）")
+
+    def wait_for_response(self, timeout: int = 120000) -> None:
+        """
+        等待当前 AI 回复完成。
+        
+        timeout: 超时毫秒数，默认 120000（2分钟）
+        """
+        if not self._connected:
+            raise ConnectionError("未连接到应用")
+        self._wait_for_streaming_done(timeout)
+
+    def ask(self, question: str, options: Optional[AskOptions] = None) -> str:
+        """
+        向应用提问，等待回复完成，返回 assistant 回复文本。
+        
+        步骤：
+        1. 清空并填写输入框
+        2. 点击发送按钮
+        3. 等待流式输出完成（__jc_sending__ 标志 或 轮询）
+        4. 返回最新一条 assistant 消息内容
+        """
+        if not self._connected:
+            raise ConnectionError("未连接到应用，请先调用 connect() 或 launch_app()")
+
+        opts = options or AskOptions()
+        page = self.get_page()
+
+        # 1. 填写问题到输入框
+        input_locator = self._get_chat_input_locator()
+        input_locator.wait_for(state="visible", timeout=10000)
+        input_locator.click()
+        input_locator.fill(question)
+
+        # 2. 点击发送
+        send_btn = self._get_send_button_locator()
+        send_btn.wait_for(state="visible", timeout=5000)
+        send_btn.click()
+
+        # 3. 等待回复完成
+        self._wait_for_streaming_done(opts.timeout)
+
+        # 4. 读取最后一条 assistant 消息
+        messages = self.read()
+        assistant_msgs = [m for m in messages if m.role == "assistant"]
+        if not assistant_msgs:
+            return ""
+        return assistant_msgs[-1].content
+
+    def send_file(self, file_path: str, message: Optional[str] = None) -> None:
+        """
+        上传附件到当前会话。
+
+        步骤：
+        1. 点击附件按钮（data-jcid="attach-button" 或 fallback）
+        2. 通过 input[type="file"] 设置文件路径
+        3. 如果提供了 message，填入输入框并发送
+
+        file_path: 要上传的文件绝对路径（如 D:/test.pdf）
+        message: 可选，上传后附带的消息文本。如果提供，上传后自动触发发送
+        """
+        if not self._connected:
+            raise ConnectionError("未连接到应用")
+
+        page = self.get_page()
+
+        # 1. 找到附件按钮
+        attach_btn = page.locator('[data-jcid="attach-button"]')
+        if attach_btn.count() == 0:
+            # Fallback：找包含附件/paperclip 等文字的按钮
+            attach_btn = page.locator('button').filter(
+                has_text=re.compile(r"附件|上传|paperclip|attach", re.I)
+            )
+
+        # 2. 尝试找到文件 input[type="file"]
+        file_input = page.locator('input[type="file"]')
+        if file_input.count() == 0 and attach_btn.count() > 0:
+            # 点击附件按钮可能会触发隐藏的 file input 显示
+            attach_btn.first.click()
+            time.sleep(0.5)
+            file_input = page.locator('input[type="file"]')
+
+        if file_input.count() == 0:
+            raise AssertError(
+                "找不到文件上传 input[type='file']。"
+                "请确认附件功能已实现，或 UI 已添加 data-jcid=\"attach-button\" 属性。"
+            )
+
+        # 3. 设置文件
+        abs_path = os.path.abspath(file_path)
+        if not os.path.exists(abs_path):
+            raise AppNotFoundError(f"上传文件不存在: {abs_path}")
+
+        file_input.set_input_files(abs_path)
+        time.sleep(1)  # 等待附件预览渲染
+
+        # 4. 如果提供了 message，自动发送
+        if message:
+            input_locator = self._get_chat_input_locator()
+            input_locator.click()
+            input_locator.fill(message)
+            send_btn = self._get_send_button_locator()
+            send_btn.click()
+
+    def read(self) -> List[JiangchangMessage]:
+        """
+        从页面 DOM 提取当前可见的所有消息。
+        
+        优先通过 data-jcid 属性识别，回退到 DOM 结构推断。
+        返回按时间顺序排列的 JiangchangMessage 列表。
+        """
+        if not self._connected:
+            raise ConnectionError("未连接到应用")
+
+        page = self.get_page()
+        messages: List[JiangchangMessage] = []
+
+        try:
+            msg_list = self._get_message_list_locator()
+            msg_list.wait_for(state="visible", timeout=10000)
+        except Exception:
+            return messages  # 消息列表还没出现，返回空
+
+        # 策略 1：通过 data-jcid="message-xxx" 查找每条消息
+        items = page.locator('[data-jcid^="message-"]')
+        if items.count() > 0:
+            for i in range(items.count()):
+                el = items.nth(i)
+                try:
+                    role = el.get_attribute("data-jcid-role") or "assistant"
+                    content = el.inner_text()
+                    is_error = el.get_attribute("data-jcid-error") == "true"
+                    tool_name = el.get_attribute("data-jcid-tool")
+                    messages.append(JiangchangMessage(
+                        id=el.get_attribute("data-jcid") or str(i),
+                        role=role,
+                        content=content,
+                        timestamp=time.time(),
+                        is_error=is_error,
+                        tool_name=tool_name,
+                    ))
+                except Exception:
+                    continue
+            return messages
+
+        # 策略 2：按 DOM 结构推断（通过 class 特征区分 user/assistant）
+        # user 消息通常 class 含 "user" 或气泡颜色为蓝色
+        # assistant 消息通常 class 含 "assistant" 或气泡颜色为灰色
+        raw_items = msg_list.locator("> *").all()
+        for idx, el in enumerate(raw_items):
+            try:
+                classes = el.get_attribute("class") or ""
+                text = el.inner_text()
+                if not text.strip():
+                    continue
+
+                # 简单启发式判断角色
+                if "user" in classes.lower():
+                    role = "user"
+                elif "assistant" in classes.lower() or "bot" in classes.lower():
+                    role = "assistant"
+                elif idx % 2 == 0:  # 交替：偶数=user，奇数=assistant
+                    role = "user"
+                else:
+                    role = "assistant"
+
+                messages.append(JiangchangMessage(
+                    id=f"msg-{idx}",
+                    role=role,
+                    content=text,
+                    timestamp=time.time(),
+                ))
+            except Exception:
+                continue
+
+        return messages
+
+    def assert_contains(self, expected: str, options: Optional[AssertOptions] = None) -> None:
+        """
+        断言最新一条 assistant 回复中包含指定内容。
+        
+        options.match_mode 支持：
+        - 'contains'（默认）：子串包含
+        - 'regex'：正则表达式匹配
+        - 'exact'：完全相等
+        """
+        if not self._connected:
+            raise ConnectionError("未连接到应用")
+
+        opts = options or AssertOptions()
+        messages = self.read()
+
+        # 找目标消息
+        assistant_msgs = [m for m in messages if m.role == "assistant"]
+        if not assistant_msgs:
+            raise AssertError(
+                f"未找到任何 assistant 消息。期望包含：{expected!r}",
+                expected=expected,
+                actual="（无 assistant 消息）",
+            )
+
+        # 默认取最后一条，可通过 message_index 自定义
+        target_idx = opts.message_index if opts.message_index != -1 else -1
+        target = assistant_msgs[target_idx]
+        actual = target.content
+
+        # 执行匹配
+        matched = False
+        if opts.match_mode == "exact":
+            matched = (actual.strip() == expected.strip())
+        elif opts.match_mode == "regex":
+            import re
+            matched = bool(re.search(expected, actual))
+        else:  # contains（默认）
+            matched = expected in actual
+
+        if not matched:
+            raise AssertError(
+                f"断言失败。期望包含：{expected!r}\n实际内容：{actual[:200]!r}",
+                expected=expected,
+                actual=actual[:200],
+            )
+
+    def disconnect(self) -> None:
+        """断开连接"""
+        if not self._connected:
+            return
+        self._cleanup()
+
+    def is_connected(self) -> bool:
+        """返回当前是否已连接"""
+        return self._connected
+
+    def get_page(self) -> Page:
+        """获取 Playwright Page 对象，供高级用法"""
+        if not self._page:
+            raise ConnectionError("未连接到应用，请先调用 connect() 或 launch_app()")
+        return self._page
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.disconnect()
+        return False

sdk/jiangchang-desktop-sdk/src/jiangchang_desktop_sdk/exceptions.py

@@ -0,0 +1,23 @@
+class JiangchangDesktopError(Exception):
+    """基类"""
+    pass
+
+class AppNotFoundError(JiangchangDesktopError):
+    """Electron 可执行文件找不到时抛出"""
+    pass
+
+class ConnectionError(JiangchangDesktopError, ConnectionError):
+    """连接桌面应用失败时抛出"""
+    pass
+
+class TimeoutError(JiangchangDesktopError, TimeoutError):
+    """操作超时"""
+    pass
+
+class AssertError(JiangchangDesktopError):
+    """断言失败时抛出，消息要包含期望值和实际值"""
+    pass
+
+class LaunchError(JiangchangDesktopError):
+    """应用启动失败时抛出"""
+    pass

sdk/jiangchang-desktop-sdk/src/jiangchang_desktop_sdk/types.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass
+from typing import Optional, List
+
+@dataclass
+class JiangchangMessage:
+    """单条会话消息"""
+    id: str
+    role: str  # 'user' | 'assistant' | 'system' | 'tool'
+    content: str
+    timestamp: float
+    is_error: bool = False
+    tool_call_id: Optional[str] = None
+    tool_name: Optional[str] = None
+    tool_status: Optional[str] = None  # 'running' | 'completed' | 'error'
+    thinking_content: Optional[str] = None
+
+@dataclass
+class AskOptions:
+    """ask() 方法的选项"""
+    timeout: int = 120000        # 毫秒
+    wait_for_tools: bool = True
+    agent_id: str = "main"
+
+@dataclass
+class LaunchOptions:
+    """launch_app() 方法的选项"""
+    executable_path: Optional[str] = None  # 默认从 JIANGCHANG_E2E_APP_PATH 环境变量读取
+    cdp_port: int = 9222
+    startup_timeout: int = 30000
+    headless: bool = False       # 是否无头模式运行
+
+@dataclass
+class AssertOptions:
+    """assert_contains() 方法的选项"""
+    timeout: int = 5000
+    match_mode: str = "contains"  # 'contains' | 'regex' | 'exact'
+    message_index: int = -1        # -1 表示最后一条 assistant 消息
+    include_tools: bool = False

sdk/jiangchang-desktop-sdk/tests/test_client.py

@@ -0,0 +1,36 @@
+import pytest
+from jiangchang_desktop_sdk import JiangchangDesktopClient, JiangchangMessage, AskOptions, LaunchOptions, AssertOptions
+from jiangchang_desktop_sdk.exceptions import JiangchangDesktopError, AppNotFoundError, ConnectionError, TimeoutError, AssertError
+
+def test_client_init():
+    """实例化 JiangchangDesktopClient，验证 _connected = False"""
+    client = JiangchangDesktopClient()
+    assert client._connected is False
+    assert client._browser is None
+    assert client._page is None
+
+def test_types_dataclasses():
+    """验证 JiangchangMessage, AskOptions 等 dataclass 可以正常实例化"""
+    msg = JiangchangMessage(id="1", role="user", content="hello", timestamp=123.456)
+    assert msg.id == "1"
+    assert msg.role == "user"
+    
+    ask_opts = AskOptions(timeout=1000)
+    assert ask_opts.timeout == 1000
+    
+    launch_opts = LaunchOptions(headless=True)
+    assert launch_opts.headless is True
+    
+    assert_opts = AssertOptions(match_mode="exact")
+    assert assert_opts.match_mode == "exact"
+
+def test_exceptions_are_exceptions():
+    """验证所有异常类都是 Exception 的子类"""
+    assert issubclass(JiangchangDesktopError, Exception)
+    assert issubclass(AppNotFoundError, JiangchangDesktopError)
+    assert issubclass(ConnectionError, JiangchangDesktopError)
+    assert issubclass(ConnectionError, ConnectionError)
+    assert issubclass(TimeoutError, JiangchangDesktopError)
+    # Note: in Python 3.10+, TimeoutError is a built-in. 
+    # Our definition: class TimeoutError(JiangchangDesktopError, TimeoutError):
+    assert issubclass(AssertError, JiangchangDesktopError)

sdk/jiangchang_skill_core/runtime_env.py

@@ -5,7 +5,8 @@ JIANGCHANG_* 数据根与用户目录：解析规则 + 可选本地 CLI 默认
 
 宿主在 skills.entries 与 Gateway 中注入：
 - PATH 前缀：{JIANGCHANG_DATA_ROOT}/python-runtime/.venv/(Scripts|bin)
-- PLAYWRIGHT_BROWSERS_PATH、VIRTUAL_ENV、JIANGCHANG_PYTHON_EXE
+- VIRTUAL_ENV、JIANGCHANG_PYTHON_EXE
+Playwright 使用本机 Chrome/Edge（launch 时 channel=chrome|msedge），不依赖宿主下载的 Chromium 包。
 技能勿在仓库内维护独立 .venv；依赖以 jiangchang-platform-kit/python-runtime 锁文件为准。
 """