重构项目路径

src/jiangchang_desktop_sdk/__init__.py

@@ -0,0 +1,38 @@
+from .exceptions import (
+    AppNotFoundError,
+    AssertError,
+    ConnectionError,
+    GatewayDownError,
+    JiangchangDesktopError,
+    LaunchError,
+    TimeoutError,
+)
+from .types import AskOptions, AssertOptions, JiangchangMessage, LaunchOptions
+
+__all__ = [
+    "JiangchangDesktopClient",
+    "JiangchangMessage",
+    "AskOptions",
+    "LaunchOptions",
+    "AssertOptions",
+    "JiangchangDesktopError",
+    "AppNotFoundError",
+    "ConnectionError",
+    "TimeoutError",
+    "AssertError",
+    "LaunchError",
+    "GatewayDownError",
+]
+__version__ = "0.1.0"
+
+
+def __getattr__(name: str):
+    if name == "JiangchangDesktopClient":
+        from .client import JiangchangDesktopClient
+
+        return JiangchangDesktopClient
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+    return sorted({*__all__, "__version__"})

src/jiangchang_desktop_sdk/exceptions.py

@@ -0,0 +1,27 @@
+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
+
+class GatewayDownError(JiangchangDesktopError):
+    """Gateway 在等待过程中被检测到已停止/退出时抛出，便于测试立刻失败而不是空等超时。"""
+    pass

src/jiangchang_desktop_sdk/testing/__init__.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+"""匠厂桌面 E2E 与集成测试常用工具子包。
+
+提供：
+  - ``SkillInfo`` / ``discover_skill_root`` / ``parse_skill_md``
+  - ``skill_healthcheck``
+  - ``HostAPIClient`` / ``HostAPIError``
+
+**不提供任何会话清理能力**：测试完全模拟真实用户的 UI 操作，
+真实用户从不会去后台删会话，测试也不该走底层 API / 文件系统去清。
+"""
+from .config import SkillInfo, discover_skill_root, parse_skill_md
+from .healthcheck import HealthCheckError, skill_healthcheck
+from .host_api import HostAPIClient, HostAPIError
+
+__all__ = [
+    "SkillInfo",
+    "discover_skill_root",
+    "parse_skill_md",
+    "HealthCheckError",
+    "skill_healthcheck",
+    "HostAPIClient",
+    "HostAPIError",
+]

src/jiangchang_desktop_sdk/testing/config.py

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+"""Skill 元信息读取。
+
+负责：
+  - 定位 skill 根目录（含 SKILL.md）；
+  - 解析 SKILL.md 的 YAML frontmatter，抽出 slug / version / name 等关键字段。
+
+为避免硬性依赖 PyYAML，这里用一个**足够用**的轻量正则 parser：只抽取
+我们真正关心的字段（顶层 `name` / `version` / `author`，嵌套 `openclaw.slug`
+与 `openclaw.category`）。更复杂的 YAML 结构请自行用 PyYAML 解析。
+"""
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+SKILL_MD = "SKILL.md"
+SKILL_ROOT_ENV = "JIANGCHANG_E2E_SKILL_ROOT"
+
+
+@dataclass
+class SkillInfo:
+    root: str
+    slug: str
+    name: str
+    version: str
+    author: str = ""
+    category: str = ""
+
+
+def discover_skill_root(start: Optional[str] = None) -> str:
+    """定位 skill 根目录。
+
+    优先级：
+      1. 环境变量 ``JIANGCHANG_E2E_SKILL_ROOT``（由调用方或测试入口提前注入）；
+      2. 从 ``start``（默认 cwd）向上回溯，直到找到含 ``SKILL.md`` 的目录。
+    """
+    env = (os.environ.get(SKILL_ROOT_ENV) or "").strip()
+    if env:
+        env_abs = os.path.abspath(env)
+        if os.path.isfile(os.path.join(env_abs, SKILL_MD)):
+            return env_abs
+    cur = Path(start or os.getcwd()).resolve()
+    for parent in [cur, *cur.parents]:
+        if (parent / SKILL_MD).exists():
+            return str(parent)
+    raise FileNotFoundError(
+        f"未能在 {start or os.getcwd()} 及其父目录中找到 {SKILL_MD}。"
+        f"请设置环境变量 {SKILL_ROOT_ENV}=<skill 根目录绝对路径>。"
+    )
+
+
+_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
+
+
+def _extract_frontmatter(text: str) -> str:
+    match = _FRONTMATTER_RE.match(text)
+    if not match:
+        raise ValueError("SKILL.md 缺少 YAML frontmatter（--- ... ---）。")
+    return match.group(1)
+
+
+def _strip_quotes(value: str) -> str:
+    v = value.strip()
+    if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'):
+        return v[1:-1]
+    return v
+
+
+def parse_skill_md(skill_root: str) -> SkillInfo:
+    """解析 SKILL.md 的 frontmatter，返回 SkillInfo。"""
+    md_path = Path(skill_root) / SKILL_MD
+    if not md_path.exists():
+        raise FileNotFoundError(f"SKILL.md 不存在：{md_path}")
+    text = md_path.read_text(encoding="utf-8")
+    fm = _extract_frontmatter(text)
+
+    top: dict[str, str] = {}
+    openclaw: dict[str, str] = {}
+
+    lines = fm.splitlines()
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        if not line.strip() or line.lstrip().startswith("#"):
+            i += 1
+            continue
+        # 顶层 key: value
+        m = re.match(r"^([A-Za-z_][\w-]*):\s*(.*)$", line)
+        if m:
+            key, rest = m.group(1), m.group(2)
+            if rest.strip() == "":
+                # 嵌套块，收集 2 空格缩进的子项
+                i += 1
+                sub: dict[str, str] = {}
+                while i < len(lines):
+                    sub_line = lines[i]
+                    if sub_line.strip() == "" or sub_line.lstrip().startswith("#"):
+                        i += 1
+                        continue
+                    if not sub_line.startswith("  "):
+                        break
+                    sm = re.match(r"^\s+([A-Za-z_][\w-]*):\s*(.*)$", sub_line)
+                    if sm:
+                        sub[sm.group(1)] = _strip_quotes(sm.group(2))
+                    i += 1
+                if key == "metadata":
+                    # openclaw 在 metadata 下再下一层：
+                    #   metadata:
+                    #     openclaw:
+                    #       slug: ...
+                    # 上面 sub 已经拿到的只是 "openclaw" 这个 key。为稳妥起见，回头
+                    # 用一个更宽松的扫描：直接找所有含 `slug:` `category:` 的行。
+                    pass
+                continue
+            top[key] = _strip_quotes(rest)
+        i += 1
+
+    # openclaw.slug / category 用宽松扫描兜底（无论嵌套几层）
+    for line in lines:
+        stripped = line.lstrip()
+        if stripped.startswith("slug:") and "slug" not in openclaw:
+            openclaw["slug"] = _strip_quotes(stripped.split(":", 1)[1])
+        elif stripped.startswith("category:") and "category" not in openclaw:
+            openclaw["category"] = _strip_quotes(stripped.split(":", 1)[1])
+
+    return SkillInfo(
+        root=os.path.abspath(skill_root),
+        slug=openclaw.get("slug", "") or Path(skill_root).name,
+        name=top.get("name", Path(skill_root).name),
+        version=top.get("version", "0.0.0"),
+        author=top.get("author", ""),
+        category=openclaw.get("category", ""),
+    )

src/jiangchang_desktop_sdk/testing/healthcheck.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+"""技能健康检查。
+
+检查项：
+  1. SKILL.md 成功解析，含 slug / version。
+  2. ``python scripts/main.py health`` 命令退出码为 0（每个 skill 都约定实现）。
+"""
+from __future__ import annotations
+
+import logging
+import os
+import subprocess
+import sys
+
+from .config import SkillInfo
+
+logger = logging.getLogger("jiangchang_desktop_sdk.testing.healthcheck")
+
+
+class HealthCheckError(RuntimeError):
+    """健康检查失败。"""
+
+
+def skill_healthcheck(
+    skill_info: SkillInfo,
+    *,
+    run_cli: bool = True,
+    timeout_s: float = 15.0,
+) -> None:
+    """对一个 skill 跑本地健康检查。失败抛 ``HealthCheckError``。"""
+    if not skill_info.slug:
+        raise HealthCheckError(f"SKILL.md 未定义 openclaw.slug：{skill_info.root}")
+    if not skill_info.version or skill_info.version == "0.0.0":
+        raise HealthCheckError(f"SKILL.md 未定义 version：{skill_info.root}")
+
+    if not run_cli:
+        return
+
+    main_py = os.path.join(skill_info.root, "scripts", "main.py")
+    if not os.path.isfile(main_py):
+        logger.warning("skill_healthcheck: %s 不存在，跳过 CLI health 检查", main_py)
+        return
+
+    cmd = [sys.executable, main_py, "health"]
+    logger.debug("skill_healthcheck: 运行 %s (cwd=%s)", cmd, skill_info.root)
+    try:
+        proc = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=timeout_s,
+            cwd=skill_info.root,
+            encoding="utf-8",
+            errors="replace",
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise HealthCheckError(
+            f"skill_healthcheck: '{' '.join(cmd)}' 超时 {timeout_s}s"
+        ) from exc
+
+    if proc.returncode != 0:
+        raise HealthCheckError(
+            f"skill_healthcheck: '{' '.join(cmd)}' 退出码 {proc.returncode}\n"
+            f"stdout: {proc.stdout[:600]}\n"
+            f"stderr: {proc.stderr[:600]}"
+        )
+    logger.info(
+        "skill_healthcheck: %s v%s OK", skill_info.slug, skill_info.version
+    )

src/jiangchang_desktop_sdk/testing/host_api.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+"""匠厂 Host HTTP API 的 Python 轻封装。
+
+使用 stdlib urllib，避免为 SDK 引入新运行时依赖。当前只暴露本模块
+真正需要的端点（健康探活、agent 列表），**不包含任何会话删除/会话
+文件系统操作**——测试侧要保持"只通过 UI 操作，不走底层接口"的原则，
+后续如需访问其他 host-api 端点请在此处按需添加读型方法。
+
+Host API 默认端口：`13210`（见 jiangchang/electron/utils/config.ts 中 `CLAWX_HOST_API`）。
+可通过环境变量 ``CLAWX_PORT_CLAWX_HOST_API`` 或本类的 ``port`` 参数覆盖。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import urllib.error
+import urllib.request
+from typing import Any, Dict, Optional
+
+DEFAULT_PORT = 13210
+DEFAULT_HOST = "127.0.0.1"
+
+logger = logging.getLogger("jiangchang_desktop_sdk.testing.host_api")
+
+
+class HostAPIError(RuntimeError):
+    """Host API 调用失败（网络错误或非 2xx 响应）。"""
+
+
+class HostAPIClient:
+    def __init__(
+        self,
+        host: str = DEFAULT_HOST,
+        port: Optional[int] = None,
+        timeout: float = 10.0,
+    ) -> None:
+        self.host = host
+        env_port = os.environ.get("CLAWX_PORT_CLAWX_HOST_API")
+        self.port = port or (int(env_port) if env_port else DEFAULT_PORT)
+        self.timeout = timeout
+
+    @property
+    def base_url(self) -> str:
+        return f"http://{self.host}:{self.port}"
+
+    def _request(self, method: str, path: str, body: Any = None) -> Any:
+        url = f"{self.base_url}{path}"
+        data = None
+        headers = {"Accept": "application/json"}
+        if body is not None:
+            data = json.dumps(body).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        req = urllib.request.Request(url, data=data, method=method, headers=headers)
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                raw = resp.read().decode("utf-8")
+                if not raw:
+                    return None
+                try:
+                    return json.loads(raw)
+                except json.JSONDecodeError:
+                    return raw
+        except urllib.error.HTTPError as exc:
+            payload: Any
+            try:
+                payload = json.loads(exc.read().decode("utf-8"))
+            except Exception:  # noqa: BLE001
+                payload = str(exc)
+            raise HostAPIError(f"{method} {path} -> HTTP {exc.code}: {payload}") from exc
+        except urllib.error.URLError as exc:
+            raise HostAPIError(f"{method} {path} -> {exc}") from exc
+
+    def ping(self) -> bool:
+        try:
+            self.list_agents()
+            return True
+        except HostAPIError:
+            return False
+
+    def list_agents(self) -> Dict[str, Any]:
+        return self._request("GET", "/api/agents") or {}

src/jiangchang_desktop_sdk/types.py

@@ -0,0 +1,48 @@
+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"
+    # 拟人化输入：每个字符间延迟（毫秒）。0 = 直接 fill()。
+    typing_delay_ms: int = 25
+    # 发送方式：True=按 Enter 键；False=点击发送按钮
+    use_enter_key: bool = True
+    # 每次 ask 前是否自动新建任务（避免上下文污染）
+    new_task: bool = True
+    # 流式输出判稳阈值（秒）：助手消息文本连续该秒数无增长视为完成
+    stable_seconds: float = 3.0
+    # 轮询间隔（秒）
+    poll_interval: float = 0.5
+
+@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

src/jiangchang_skill_core/__init__.py

@@ -0,0 +1,43 @@
+from .client import EntitlementClient
+from .errors import EntitlementDeniedError, EntitlementError, EntitlementServiceError
+from .guard import enforce_entitlement
+from .models import EntitlementResult
+from .runtime_env import (
+    apply_cli_local_defaults,
+    get_data_root,
+    get_sibling_skills_root,
+    get_skills_root,
+    get_user_id,
+    platform_default_data_root,
+)
+from .unified_logging import (
+    attach_unified_file_handler,
+    ensure_trace_for_process,
+    get_skill_log_file_path,
+    get_skill_logger,
+    get_unified_logs_dir,
+    setup_skill_logging,
+    subprocess_env_with_trace,
+)
+
+__all__ = [
+    "EntitlementClient",
+    "EntitlementDeniedError",
+    "EntitlementError",
+    "EntitlementResult",
+    "EntitlementServiceError",
+    "apply_cli_local_defaults",
+    "attach_unified_file_handler",
+    "enforce_entitlement",
+    "ensure_trace_for_process",
+    "get_data_root",
+    "get_sibling_skills_root",
+    "get_skills_root",
+    "get_skill_log_file_path",
+    "get_skill_logger",
+    "get_unified_logs_dir",
+    "get_user_id",
+    "platform_default_data_root",
+    "setup_skill_logging",
+    "subprocess_env_with_trace",
+]

src/jiangchang_skill_core/client.py

@@ -0,0 +1,63 @@
+import os
+from typing import Any
+
+import requests
+
+from .errors import EntitlementServiceError
+from .models import EntitlementResult
+
+
+class EntitlementClient:
+    def __init__(
+        self,
+        base_url: str | None = None,
+        api_key: str | None = None,
+        timeout_seconds: int | None = None,
+    ) -> None:
+        self.base_url = (base_url or os.getenv("JIANGCHANG_AUTH_BASE_URL", "")).rstrip("/")
+        self.api_key = api_key or os.getenv("JIANGCHANG_AUTH_API_KEY", "")
+        self.timeout_seconds = timeout_seconds or int(
+            os.getenv("JIANGCHANG_AUTH_TIMEOUT_SECONDS", "5")
+        )
+        if not self.base_url:
+            raise EntitlementServiceError("missing JIANGCHANG_AUTH_BASE_URL")
+
+    def check_entitlement(
+        self,
+        user_id: str,
+        skill_slug: str,
+        trace_id: str = "",
+        context: dict[str, Any] | None = None,
+    ) -> EntitlementResult:
+        url = f"{self.base_url}/api/entitlements/check"
+        payload = {
+            "user_id": user_id,
+            "skill_slug": skill_slug,
+            "trace_id": trace_id,
+            "context": context or {},
+        }
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["Authorization"] = f"Bearer {self.api_key}"
+        try:
+            res = requests.post(
+                url,
+                json=payload,
+                headers=headers,
+                timeout=self.timeout_seconds,
+            )
+        except requests.RequestException as exc:
+            raise EntitlementServiceError(f"entitlement request failed: {exc}") from exc
+
+        if res.status_code != 200:
+            raise EntitlementServiceError(f"entitlement http status {res.status_code}")
+        try:
+            body = res.json()
+        except ValueError as exc:
+            raise EntitlementServiceError("entitlement response is not json") from exc
+
+        data = body.get("data") or {}
+        allow = bool(data.get("allow", False))
+        reason = str(data.get("reason") or body.get("msg") or "")
+        expire_at = str(data.get("expire_at") or "")
+        return EntitlementResult(allow=allow, reason=reason, expire_at=expire_at, raw=body)

src/jiangchang_skill_core/errors.py

@@ -0,0 +1,10 @@
+class EntitlementError(Exception):
+    pass
+
+
+class EntitlementDeniedError(EntitlementError):
+    pass
+
+
+class EntitlementServiceError(EntitlementError):
+    pass

src/jiangchang_skill_core/guard.py

@@ -0,0 +1,22 @@
+from .client import EntitlementClient
+from .errors import EntitlementDeniedError
+from .models import EntitlementResult
+
+
+def enforce_entitlement(
+    user_id: str,
+    skill_slug: str,
+    trace_id: str = "",
+    context: dict | None = None,
+    client: EntitlementClient | None = None,
+) -> EntitlementResult:
+    c = client or EntitlementClient()
+    result = c.check_entitlement(
+        user_id=user_id,
+        skill_slug=skill_slug,
+        trace_id=trace_id,
+        context=context or {},
+    )
+    if not result.allow:
+        raise EntitlementDeniedError(result.reason or "skill not purchased or expired")
+    return result

src/jiangchang_skill_core/models.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class EntitlementResult:
+    allow: bool
+    reason: str = ""
+    expire_at: str = ""
+    raw: dict[str, Any] | None = None

src/jiangchang_skill_core/runtime_env.py

@@ -0,0 +1,136 @@
+"""
+JIANGCHANG_* 数据根与用户目录：解析规则 + 可选本地 CLI 默认值。
+
+宿主在 skills.entries 与 Gateway 中注入：
+- PATH 前缀：{JIANGCHANG_DATA_ROOT}/python-runtime/.venv/(Scripts|bin)
+- VIRTUAL_ENV、JIANGCHANG_PYTHON_EXE
+Playwright 使用本机 Chrome/Edge（launch 时 channel=chrome|msedge），不依赖宿主下载的 Chromium 包。
+技能勿在仓库内维护独立 .venv；共享逻辑请使用本仓库发布的 ``jiangchang-platform-kit`` 包，
+其余运行时依赖（如 Playwright、模型 SDK）由宿主或技能各自的环境策略决定。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+# 发版/嵌入宿主前改为 False，或仅通过环境变量 JIANGCHANG_CLI_LOCAL_DEV=1 开启
+CLI_LOCAL_DEV_ENABLED = True
+
+# 本地开发且未设置 JIANGCHANG_USER_ID 时注入（与宿主约定一致即可修改）
+DEFAULT_LOCAL_USER_ID = "10032"
+
+# Windows 下未设置 JIANGCHANG_DATA_ROOT 时的默认盘路径
+WIN_DEFAULT_DATA_ROOT = r"D:\jiangchang-data"
+
+# 匠厂桌面宿主应用根（未设置 JIANGCHANG_APP_ROOT 时 Windows 兜底；技能一般在 {APP}/skills/ 下并列）
+WIN_DEFAULT_JIANGCHANG_APP_ROOT = r"D:\AI\jiangchang"
+
+
+def platform_default_data_root() -> str:
+    if sys.platform == "win32":
+        return WIN_DEFAULT_DATA_ROOT
+    return os.path.join(os.path.expanduser("~"), ".jiangchang-data")
+
+
+def get_data_root() -> str:
+    env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
+    if env:
+        return env
+    return platform_default_data_root()
+
+
+def get_user_id() -> str:
+    return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon"
+
+
+def _looks_like_skills_root(path: str) -> bool:
+    if not path or not os.path.isdir(path):
+        return False
+    for marker in (
+        "llm-manager",
+        "content-manager",
+        "account-manager",
+        "sohu-publisher",
+        "toutiao-publisher",
+        "logistics-tracker",
+        "api-key-vault",
+    ):
+        if os.path.isdir(os.path.join(path, marker)):
+            return True
+    return False
+
+
+def get_skills_root() -> str:
+    """
+    并列技能安装目录（其下为「技能 slug」子目录）。
+
+    优先级：
+    1) JIANGCHANG_SKILLS_ROOT
+    2) CLAW_SKILLS_ROOT
+    3) JIANGCHANG_APP_ROOT：若 {APP}/skills 存在且像技能根则用之；否则若 APP 下已有技能子目录则 APP 即根
+    4) Windows：默认 D:\\AI\\jiangchang 同上规则
+    5) 其他平台：~/.openclaw/skills
+    """
+    for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
+        v = (os.getenv(key) or "").strip()
+        if v:
+            return os.path.normpath(v)
+
+    app = (os.getenv("JIANGCHANG_APP_ROOT") or "").strip()
+    if sys.platform == "win32" and not app:
+        app = WIN_DEFAULT_JIANGCHANG_APP_ROOT
+    if app:
+        nested = os.path.join(app, "skills")
+        if _looks_like_skills_root(nested):
+            return os.path.normpath(nested)
+        if _looks_like_skills_root(app):
+            return os.path.normpath(app)
+
+    if sys.platform == "win32":
+        nested = os.path.join(WIN_DEFAULT_JIANGCHANG_APP_ROOT, "skills")
+        if _looks_like_skills_root(nested):
+            return os.path.normpath(nested)
+        if _looks_like_skills_root(WIN_DEFAULT_JIANGCHANG_APP_ROOT):
+            return os.path.normpath(WIN_DEFAULT_JIANGCHANG_APP_ROOT)
+
+    return os.path.normpath(os.path.join(os.path.expanduser("~"), ".openclaw", "skills"))
+
+
+def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str:
+    """
+    编排子进程调用兄弟技能时用：先根据本技能 scripts 目录推断并列根；若目录下能识别出兄弟技能
+    则用之（开发仓 D:\\OpenClaw 与安装目录 ~/.openclaw/skills 均满足「技能根之上一级 = 并列根」），
+    避免全局 JIANGCHANG_SKILLS_ROOT / CLAW_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。
+    推断失败时再读上述环境变量，最后回落 get_skills_root()。
+    """
+    if skill_scripts_dir:
+        scripts = os.path.abspath(skill_scripts_dir)
+        skill_root = os.path.dirname(scripts)
+        inferred = os.path.dirname(skill_root)
+        if _looks_like_skills_root(inferred):
+            return os.path.normpath(inferred)
+
+    for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
+        v = (os.getenv(key) or "").strip()
+        if v:
+            return os.path.normpath(v)
+
+    return get_skills_root()
+
+
+def apply_cli_local_defaults() -> None:
+    """
+    在 CLI 最早阶段调用（main.py 在 import 业务包之前）。
+    宿主已设置 JIANGCHANG_* 时不会覆盖。
+    """
+    enabled = CLI_LOCAL_DEV_ENABLED
+    if not enabled:
+        v = (os.getenv("JIANGCHANG_CLI_LOCAL_DEV") or "").strip().lower()
+        enabled = v in ("1", "true", "yes", "on")
+    if not enabled:
+        return
+    if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip():
+        os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
+    if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
+        os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()

src/jiangchang_skill_core/unified_logging.py

@@ -0,0 +1,156 @@
+"""
+统一文件日志：{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log
+按日轮转；行内带 trace_id 与 skill_slug，便于跨技能排查。
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+import uuid
+from logging.handlers import TimedRotatingFileHandler
+from typing import Optional
+
+from .runtime_env import get_data_root, get_user_id
+
+_skill_slug: str = ""
+_logger_name: str = ""
+
+
+def get_unified_logs_dir() -> str:
+    path = os.path.join(get_data_root(), get_user_id(), "logs")
+    os.makedirs(path, exist_ok=True)
+    return path
+
+
+def get_skill_log_file_path() -> str:
+    """与 setup 写入的主日志文件一致（供终端提示等）。"""
+    override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip()
+    if override:
+        parent = os.path.dirname(os.path.abspath(override))
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        return os.path.abspath(override)
+    return os.path.join(get_unified_logs_dir(), "jiangchang.log")
+
+
+def ensure_trace_for_process() -> str:
+    """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。"""
+    existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
+    if existing:
+        return existing
+    tid = uuid.uuid4().hex[:12]
+    os.environ["JIANGCHANG_TRACE_ID"] = tid
+    return tid
+
+
+def subprocess_env_with_trace(environ: Optional[dict] = None) -> dict:
+    """subprocess.run(..., env=...) 时使用，保证带上当前 trace。"""
+    ensure_trace_for_process()
+    tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
+    if not tid:
+        tid = ensure_trace_for_process()
+    base = os.environ if environ is None else environ
+    return {**base, "JIANGCHANG_TRACE_ID": tid}
+
+
+class _SkillContextFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.trace_id = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or "-"
+        record.skill_slug = _skill_slug or "-"
+        return True
+
+
+_FORMAT = (
+    "%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
+)
+_DATEFMT = "%Y-%m-%dT%H:%M:%S"
+
+
+def _log_level_from_env() -> int:
+    v = (os.getenv("JIANGCHANG_LOG_LEVEL") or "INFO").strip().upper()
+    return getattr(logging, v, None) or logging.INFO
+
+
+def _backup_count() -> int:
+    try:
+        n = int((os.getenv("JIANGCHANG_LOG_BACKUP_COUNT") or "30").strip())
+        return max(1, min(n, 365))
+    except ValueError:
+        return 30
+
+
+def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
+    """
+    幂等：为指定 logger 挂载统一文件日志与可选 stderr。
+    须在进程早期调用（如 CLI main 在业务日志之前）。
+    """
+    global _skill_slug, _logger_name
+    ensure_trace_for_process()
+    _skill_slug = skill_slug
+    _logger_name = logger_name
+
+    log = logging.getLogger(logger_name)
+    if log.handlers:
+        return
+    log.setLevel(_log_level_from_env())
+    path = get_skill_log_file_path()
+    fh = TimedRotatingFileHandler(
+        path,
+        when="midnight",
+        interval=1,
+        backupCount=_backup_count(),
+        encoding="utf-8",
+        delay=True,
+    )
+    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
+    fh.setFormatter(fmt)
+    fh.addFilter(_SkillContextFilter())
+    log.addHandler(fh)
+    if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
+        "1",
+        "true",
+        "yes",
+        "on",
+    ):
+        sh = logging.StreamHandler(sys.stderr)
+        sh.setLevel(logging.WARNING)
+        sh.setFormatter(fmt)
+        sh.addFilter(_SkillContextFilter())
+        log.addHandler(sh)
+    log.propagate = False
+
+
+def get_skill_logger() -> logging.Logger:
+    if not _logger_name:
+        raise RuntimeError("get_skill_logger: call setup_skill_logging first")
+    return logging.getLogger(_logger_name)
+
+
+def attach_unified_file_handler(
+    log_path: str,
+    *,
+    skill_slug: str,
+    logger_name: str,
+    level: int = logging.DEBUG,
+) -> logging.Logger:
+    """
+    独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。
+    """
+    global _skill_slug
+    ensure_trace_for_process()
+    _skill_slug = skill_slug
+    lg = logging.getLogger(logger_name)
+    lg.handlers.clear()
+    lg.setLevel(level)
+    parent = os.path.dirname(os.path.abspath(log_path))
+    if parent:
+        os.makedirs(parent, exist_ok=True)
+    fh = logging.FileHandler(log_path, encoding="utf-8")
+    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
+    fh.setFormatter(fmt)
+    fh.addFilter(_SkillContextFilter())
+    lg.addHandler(fh)
+    lg.propagate = False
+    return lg