feat(activity): SRCP v1 step gates, pause/resume/stop control (v1.2.0)

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

README.md

@@ -12,7 +12,7 @@ pip install jiangchang-platform-kit --index-url https://git.jc2009.com/api/packa
 
 ## 3. 包含的模块
 
-- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。
+- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、**Skill Activity Stream / Job 运行时**（`emit` / `finish` / Run Journal）、兄弟技能 CLI 桥（`sibling_bridge`）、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。
 - **jiangchang_desktop_sdk**：匠厂桌面应用自动化客户端（基于 CDP + Playwright 同步 API）。**本包不声明 Playwright 依赖**；请在运行环境中自行安装 Playwright，以便 `import playwright` 可用。
 
 ## 4. 快速开始
@@ -56,7 +56,7 @@ finally:
 
 ### 桌面 E2E（`jiangchang_desktop_sdk.e2e_helpers`）
 
-Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录，不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` / `CLAW_USER_ID` 来伪造用户（否则会回退到 `_anon`）。
+Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录，不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` 来伪造用户（否则会回退到 `_anon`）。
 
 推荐 preflight 流程：
 
@@ -78,6 +78,163 @@ assert_skill_env_file(ctx, ["API_KEY"])
 send_prompt_via_composer(page, "第一行\n第二行")
 ```
 
+## Skill Activity Stream / Job 运行时
+
+长任务（批量生成、RPA）需要把**逐步进度**推到匠厂宿主 UI，而不能依赖 Agent 轮询或 OpenClaw stdout patch。`jiangchang_skill_core.activity` 提供 **Skill Activity Stream (SAS) v1** 契约：
+
+| API | 作用 |
+|-----|------|
+| `emit(text, ...)` | 写 Run Journal（`activity` / `progress` / `warn` / `debug`） |
+| `step(text)` | `emit(type='activity')` 别名 |
+| `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON |
+| `rpa_step(name)` | 装饰器：闸门 + 进入/成功/失败自动 emit |
+| `checkpoint()` | consult control（不写步骤事件） |
+| `interruptible_sleep(sec)` | 可暂停/停止的 async sleep |
+| `JobStopped` | 宿主 stop 时抛出 |
+| `job_context(...)` | 可选包裹 main；异常时自动 `finish(failed)` |
+
+### Run Journal 路径（宿主 tail 此文件）
+
+```
+{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl
+```
+
+- `job_id`：优先 `JIANGCHANG_JOB_ID`，否则进程内 `uuid4().hex`（首次 emit 时生成并缓存）
+- `JIANGCHANG_RUN_MODE`：`job` | `cli`（默认 `cli`），首行 meta 事件记录
+- NDJSON，UTF-8，append-only，每行 flush
+- `debug` 类型仅当 `JIANGCHANG_ACTIVITY_DEBUG=1` 时写入
+
+### 与 Agent / OpenClaw 的分工
+
+- **`emit()` / journal**：永不 print 到 stdout，避免 bash tool 结果淹没 Agent context
+- **`finish()`**：唯一允许向 stdout 写业务结果的 API；单行 compact JSON，`status` + 摘要字段，默认 ≤4096 字符
+- 技能里手写 `print("OK")`、pretty JSON 应逐步改为 `finish()`；RPA 步骤用 `emit("打开浏览器")` 或 `@rpa_step`
+
+### GEO batch 示例
+
+```python
+from jiangchang_skill_core.activity import emit, finish
+
+for i, item in enumerate(items, 1):
+    emit(f"生成 {item['title']}", type="progress", current=i, total=len(items), skill="geo-batch")
+    # ... business logic ...
+
+finish(status="success", skill="geo-batch", generated=len(items))
+```
+
+### RPA 步骤示例
+
+```python
+from jiangchang_skill_core.activity import emit, rpa_step
+
+@rpa_step("打开浏览器")
+def open_browser(page):
+    page.goto("https://example.com")
+
+emit("手动步骤说明", skill="publish-toutiao", stage="rpa")
+```
+
+`RpaVideoSession.add_step()` 在录屏字幕之外默认同步 `emit(..., stage="rpa.video")`；可通过 `emit_activity=False` 关闭。
+
+## Skill Runtime Control Protocol (SRCP) v1 — 1.2.0+
+
+宿主通过 **control 文件**向技能进程下发暂停 / 继续 / 停止；platform-kit 在**步骤闸门**自动读取，技能作者无需手写控制逻辑。
+
+### control.json（宿主写、kit 读）
+
+```
+{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.control.json
+```
+
+```json
+{
+  "schema_version": 1,
+  "command": "none",
+  "reason": "user",
+  "ts": 1730000000000,
+  "nonce": 1
+}
+```
+
+| command | 行为 |
+|---------|------|
+| `none` | 正常运行 |
+| `pause` | 下一步闸门进入暂停（阻塞直到 resume 或 stop） |
+| `resume` | 从暂停恢复 |
+| `stop` | 协作式停止，抛出 `JobStopped` |
+
+宿主更新命令时应递增 `nonce`（kit 每次闸门重新读文件，不依赖缓存）。
+
+### 步骤闸门（自动挂载）
+
+| 调用点 | 闸门行为 |
+|--------|----------|
+| `emit()` / `step()` | 进入步骤前 gate + 写 Journal |
+| `@rpa_step` | 进入步骤前 gate，再写 ▶ / ✓ |
+| `RpaVideoSession.add_step()` | 经 `emit()` 自动 gate |
+| `interruptible_sleep()` | 每 200ms 切片 `checkpoint()`（可暂停/停止） |
+
+暂停在**当前步骤边界**生效（与影刀「当前指令执行完后暂停」一致），不在 DOM 原子操作中途冻结。
+
+### lifecycle Journal 事件
+
+闸门写入 `type: lifecycle` 事件，供宿主任务中心显示运行 / 暂停 / 继续状态：
+
+```json
+{"type":"lifecycle","status":"paused","text":"已暂停，等待继续","step_index":3,"step_name":"打开页面",...}
+```
+
+### API
+
+| API | 作用 |
+|-----|------|
+| `checkpoint()` | 仅 consult control（长等待循环内使用） |
+| `interruptible_sleep(sec)` | async 可中断 sleep |
+| `interruptible_sleep_sync(sec)` | sync 可中断 sleep |
+| `JobStopped` | 宿主 stop 时抛出；`job_context` 自动 `finish(..., error_code="JOB_STOPPED")` |
+| `get_control_path()` | 当前 job 的 control 文件路径（诊断） |
+
+环境变量：
+
+- `JIANGCHANG_CONTROL_POLL_SEC` — 暂停轮询间隔（默认 `0.2`）
+- `JIANGCHANG_CONTROL_DISABLED=1` — 禁用闸门（仅测试）
+
+### RPA 示例（拆步 + 可中断等待）
+
+```python
+from jiangchang_skill_core.activity import (
+    emit,
+    finish,
+    job_context,
+    interruptible_sleep,
+    rpa_step,
+)
+
+def cmd_run(...):
+    with job_context(skill="demo-skill"):
+        emit("开始任务", skill="demo-skill", stage="run")
+
+        @rpa_step("打开后台")
+        async def open_admin(page):
+            await page.goto("https://example.com")
+            await interruptible_sleep(1.5)
+
+        await open_admin(page)
+        finish(status="success", message="完成", skill="demo-skill")
+```
+
+技能作者规范以 **skill-template** 为准；本 README 为实现细节参考。
+
+### 兄弟技能 CLI 桥
+
+```python
+from jiangchang_skill_core import call_sibling_json, get_sibling_main_path
+
+accounts = call_sibling_json("account-manager", ["list", "all"])
+```
+
+同步 `subprocess.run`，`capture_output=True`，自动带上 `subprocess_env_with_trace()`。
+
 ## 共享媒体资源
 
 `jiangchang_skill_core.media_assets` 提供共享媒体资源解析能力。默认会从：

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jiangchang-platform-kit"
-version = "1.0.16"
+version = "1.2.0"
 description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK"
 readme = "README.md"
 requires-python = ">=3.10"

src/jiangchang_skill_core/__init__.py

@@ -1,3 +1,22 @@
+from .activity import (
+    JobStopped,
+    checkpoint,
+    emit,
+    finish,
+    get_control_path,
+    get_job_id,
+    get_run_mode,
+    interruptible_sleep,
+    interruptible_sleep_sync,
+    job_context,
+    rpa_step,
+    step,
+)
+from .sibling_bridge import (
+    call_sibling_json,
+    call_sibling_json_array,
+    get_sibling_main_path,
+)
 from .client import EntitlementClient
 from .config import ensure_env_file, get, get_bool, get_float, get_int
 from .errors import EntitlementDeniedError, EntitlementError, EntitlementServiceError
@@ -5,8 +24,10 @@ from .guard import enforce_entitlement
 from .models import EntitlementResult
 from .runtime_env import (
     apply_cli_local_defaults,
+    ensure_runs_dir,
     find_chrome_executable,
     get_data_root,
+    get_runs_dir,
     get_sibling_skills_root,
     get_skills_root,
     get_user_id,
@@ -52,6 +73,23 @@ except ImportError:
     rpa = None  # type: ignore[assignment,misc]
 
 __all__ = [
+    "JobStopped",
+    "call_sibling_json",
+    "call_sibling_json_array",
+    "checkpoint",
+    "emit",
+    "finish",
+    "get_control_path",
+    "get_job_id",
+    "get_run_mode",
+    "get_runs_dir",
+    "ensure_runs_dir",
+    "interruptible_sleep",
+    "interruptible_sleep_sync",
+    "job_context",
+    "rpa_step",
+    "step",
+    "get_sibling_main_path",
     "EntitlementClient",
     "EntitlementDeniedError",
     "EntitlementError",
@@ -88,6 +126,7 @@ __all__ = [
     "get_data_root",
     "get_float",
     "get_int",
+    "get_runs_dir",
     "get_sibling_skills_root",
     "get_skills_root",
     "get_skill_log_file_path",

src/jiangchang_skill_core/activity.py

@@ -0,0 +1,546 @@
+"""
+Skill Activity Stream (SAS) v1 + Skill Runtime Control Protocol (SRCP) v1.
+
+emit() / journal events never touch stdout (Agent/bash tool result stays clean).
+finish() is the only API that writes a single-line result JSON to stdout.
+
+SRCP: host writes {job_id}.control.json; kit consults at step gates (emit,
+rpa_step, interruptible_sleep) for pause / resume / stop.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import functools
+import inspect
+import json
+import os
+import sys
+import time
+import uuid
+from contextlib import contextmanager
+from typing import Any, Callable, Iterator, Optional, TypeVar
+
+from .runtime_env import ensure_runs_dir, get_runs_dir
+
+_SCHEMA_VERSION = 1
+_CONTROL_SCHEMA_VERSION = 1
+_VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"})
+_VALID_CONTROL_COMMANDS = frozenset({"none", "pause", "resume", "stop"})
+_RESULT_MAX_CHARS = 4096
+_DEFAULT_CONTROL_POLL_SEC = 0.2
+
+_cached_job_id: Optional[str] = None
+_journal_path: Optional[str] = None
+_meta_written: bool = False
+_journal_file: Optional[Any] = None
+_step_index: int = 0
+_current_step_name: Optional[str] = None
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+
+class JobStopped(Exception):
+    """Raised when the host requests stop via control.json."""
+
+
+def _activity_debug_enabled() -> bool:
+    v = (os.getenv("JIANGCHANG_ACTIVITY_DEBUG") or "").strip().lower()
+    return v in ("1", "true", "yes", "on")
+
+
+def _control_disabled() -> bool:
+    v = (os.getenv("JIANGCHANG_CONTROL_DISABLED") or "").strip().lower()
+    return v in ("1", "true", "yes", "on")
+
+
+def _control_poll_sec() -> float:
+    raw = (os.getenv("JIANGCHANG_CONTROL_POLL_SEC") or "").strip()
+    if not raw:
+        return _DEFAULT_CONTROL_POLL_SEC
+    try:
+        value = float(raw)
+    except ValueError:
+        return _DEFAULT_CONTROL_POLL_SEC
+    return max(0.05, value)
+
+
+def get_run_mode() -> str:
+    mode = (os.getenv("JIANGCHANG_RUN_MODE") or "").strip().lower()
+    if mode in ("job", "cli"):
+        return mode
+    return "cli"
+
+
+def get_job_id() -> str:
+    global _cached_job_id
+    env_job = (os.getenv("JIANGCHANG_JOB_ID") or "").strip()
+    if env_job:
+        return env_job
+    if _cached_job_id is None:
+        _cached_job_id = uuid.uuid4().hex
+    return _cached_job_id
+
+
+def get_control_path() -> str:
+    """Path to the host-owned control file for the current job."""
+    ensure_runs_dir()
+    return os.path.join(get_runs_dir(), f"{get_job_id()}.control.json")
+
+
+def _get_journal_path() -> str:
+    global _journal_path
+    if _journal_path is None:
+        ensure_runs_dir()
+        _journal_path = os.path.join(get_runs_dir(), f"{get_job_id()}.jsonl")
+    return _journal_path
+
+
+def _open_journal() -> Any:
+    global _journal_file
+    if _journal_file is None:
+        path = _get_journal_path()
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        _journal_file = open(path, "a", encoding="utf-8")
+    return _journal_file
+
+
+def _write_journal_line(event: dict[str, Any]) -> None:
+    global _meta_written
+    f = _open_journal()
+    if not _meta_written:
+        meta = {
+            "type": "meta",
+            "schema_version": _SCHEMA_VERSION,
+            "job_id": get_job_id(),
+            "run_mode": get_run_mode(),
+            "ts": _now_ms(),
+        }
+        f.write(json.dumps(meta, ensure_ascii=False, separators=(",", ":")) + "\n")
+        f.flush()
+        _meta_written = True
+    f.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n")
+    f.flush()
+
+
+def _now_ms() -> int:
+    return int(time.time() * 1000)
+
+
+def _read_control_payload() -> dict[str, Any] | None:
+    path = get_control_path()
+    try:
+        with open(path, encoding="utf-8") as handle:
+            payload = json.load(handle)
+    except FileNotFoundError:
+        return None
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+    return payload
+
+
+def _read_control_command() -> str:
+    payload = _read_control_payload()
+    if not payload:
+        return "none"
+    command = payload.get("command")
+    if not isinstance(command, str):
+        return "none"
+    normalized = command.strip().lower()
+    if normalized not in _VALID_CONTROL_COMMANDS:
+        return "none"
+    return normalized
+
+
+def _write_lifecycle_event(
+    status: str,
+    text: str,
+    *,
+    step_name: str | None = None,
+    skill: str | None = None,
+) -> None:
+    event: dict[str, Any] = {
+        "type": "lifecycle",
+        "status": status,
+        "text": text,
+        "schema_version": _SCHEMA_VERSION,
+        "control_schema_version": _CONTROL_SCHEMA_VERSION,
+        "ts": _now_ms(),
+        "step_index": _step_index,
+    }
+    name = step_name or _current_step_name
+    if name:
+        event["step_name"] = name
+    if skill is not None:
+        event["skill"] = skill
+    _write_journal_line(event)
+
+
+def _wait_while_paused(*, step_name: str | None = None) -> None:
+    poll = _control_poll_sec()
+    while True:
+        command = _read_control_command()
+        if command == "stop":
+            _write_lifecycle_event("stopping", "任务已停止", step_name=step_name)
+            raise JobStopped("Job stopped by host.")
+        if command in ("none", "resume"):
+            _write_lifecycle_event("resumed", "继续执行", step_name=step_name)
+            return
+        time.sleep(poll)
+
+
+def _apply_control(*, advance_step: bool, step_name: str | None = None) -> None:
+    if _control_disabled():
+        return
+
+    global _step_index, _current_step_name
+    if advance_step:
+        _step_index += 1
+        if step_name:
+            _current_step_name = step_name
+
+    command = _read_control_command()
+    if command == "stop":
+        _write_lifecycle_event("stopping", "任务已停止", step_name=step_name)
+        raise JobStopped("Job stopped by host.")
+    if command == "pause":
+        _write_lifecycle_event("paused", "已暂停，等待继续", step_name=step_name)
+        _wait_while_paused(step_name=step_name)
+
+
+def checkpoint() -> None:
+    """
+    Consult host control without advancing the step index.
+
+    Use inside long waits (interruptible_sleep) between step boundaries.
+    """
+    _apply_control(advance_step=False)
+
+
+def _step_gate(step_name: str | None) -> None:
+    """Step boundary: advance index, then pause/stop gate."""
+    _apply_control(advance_step=True, step_name=step_name)
+
+
+def _build_event(
+    type_: str,
+    text: str,
+    *,
+    skill: str | None = None,
+    stage: str | None = None,
+    current: int | None = None,
+    total: int | None = None,
+    item_id: str | None = None,
+    ts: int | None = None,
+    **extra: Any,
+) -> dict[str, Any]:
+    cleaned = (text or "").strip()
+    if not cleaned:
+        raise ValueError("text must be non-empty")
+    event: dict[str, Any] = {
+        "type": type_,
+        "text": cleaned,
+        "schema_version": _SCHEMA_VERSION,
+        "ts": ts if ts is not None else _now_ms(),
+    }
+    if skill is not None:
+        event["skill"] = skill
+    if stage is not None:
+        event["stage"] = stage
+    if current is not None:
+        event["current"] = current
+    if total is not None:
+        event["total"] = total
+    if item_id is not None:
+        event["item_id"] = item_id
+    for key, value in extra.items():
+        if key not in event:
+            event[key] = value
+    return event
+
+
+def _emit_core(
+    text: str,
+    *,
+    type: str = "activity",
+    skill: str | None = None,
+    stage: str | None = None,
+    current: int | None = None,
+    total: int | None = None,
+    item_id: str | None = None,
+    **extra: Any,
+) -> None:
+    if type == "debug" and not _activity_debug_enabled():
+        return
+    if type not in _VALID_EMIT_TYPES:
+        raise ValueError(f"invalid emit type: {type!r}")
+    event = _build_event(
+        type,
+        text,
+        skill=skill,
+        stage=stage,
+        current=current,
+        total=total,
+        item_id=item_id,
+        **extra,
+    )
+    _write_journal_line(event)
+
+
+def emit(
+    text: str,
+    *,
+    type: str = "activity",
+    skill: str | None = None,
+    stage: str | None = None,
+    current: int | None = None,
+    total: int | None = None,
+    item_id: str | None = None,
+    **extra: Any,
+) -> None:
+    """Append one SAS event to the Run Journal. Never writes to stdout."""
+    gate_name = stage or text
+    _step_gate(gate_name)
+    _emit_core(
+        text,
+        type=type,
+        skill=skill,
+        stage=stage,
+        current=current,
+        total=total,
+        item_id=item_id,
+        **extra,
+    )
+
+
+def step(text: str, **kwargs: Any) -> None:
+    """Shorthand for emit(type='activity')."""
+    emit(text, type="activity", **kwargs)
+
+
+async def interruptible_sleep(seconds: float, *, chunk: float | None = None) -> None:
+    """asyncio.sleep in slices; each slice consults host control (pause/stop)."""
+    if seconds <= 0:
+        return
+    poll = chunk if chunk is not None else _control_poll_sec()
+    poll = max(0.05, poll)
+    deadline = time.monotonic() + seconds
+    while True:
+        checkpoint()
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return
+        await asyncio.sleep(min(poll, remaining))
+
+
+def interruptible_sleep_sync(seconds: float, *, chunk: float | None = None) -> None:
+    """time.sleep in slices; each slice consults host control (pause/stop)."""
+    if seconds <= 0:
+        return
+    poll = chunk if chunk is not None else _control_poll_sec()
+    poll = max(0.05, poll)
+    deadline = time.monotonic() + seconds
+    while True:
+        checkpoint()
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return
+        time.sleep(min(poll, remaining))
+
+
+def _serialize_result(result: dict[str, Any]) -> str:
+    line = json.dumps(result, ensure_ascii=False, separators=(",", ":"))
+    if len(line) <= _RESULT_MAX_CHARS:
+        return line
+
+    core_keys = ("type", "schema_version", "status", "skill", "message")
+    truncated: dict[str, Any] = {
+        k: result[k] for k in core_keys if k in result
+    }
+    truncated["truncated"] = True
+
+    for key, value in result.items():
+        if key in truncated or key == "truncated":
+            continue
+        candidate = {**truncated, key: value}
+        candidate_line = json.dumps(candidate, ensure_ascii=False, separators=(",", ":"))
+        if len(candidate_line) <= _RESULT_MAX_CHARS:
+            truncated[key] = value
+
+    line = json.dumps(truncated, ensure_ascii=False, separators=(",", ":"))
+    if len(line) > _RESULT_MAX_CHARS:
+        truncated.pop("message", None)
+        line = json.dumps(truncated, ensure_ascii=False, separators=(",", ":"))
+    if len(line) > _RESULT_MAX_CHARS:
+        return line[:_RESULT_MAX_CHARS]
+    return line
+
+
+def finish(
+    *,
+    status: str = "success",
+    message: str | None = None,
+    skill: str | None = None,
+    **fields: Any,
+) -> None:
+    """
+    1) Write terminal done|error event to Run Journal.
+    2) Print a single compact result JSON line to stdout (Agent/bash path).
+    """
+    if status not in ("success", "partial", "failed"):
+        raise ValueError(f"invalid finish status: {status!r}")
+
+    journal_type = "error" if status == "failed" else "done"
+    terminal_text = message or ("任务失败" if status == "failed" else "任务完成")
+
+    journal_event = _build_event(
+        journal_type,
+        terminal_text,
+        skill=skill,
+        status=status,
+        **fields,
+    )
+    if message is not None:
+        journal_event["message"] = message
+    _write_journal_line(journal_event)
+
+    result: dict[str, Any] = {
+        "type": "result",
+        "schema_version": _SCHEMA_VERSION,
+        "status": status,
+    }
+    if skill is not None:
+        result["skill"] = skill
+    if message is not None:
+        result["message"] = message
+    result.update(fields)
+
+    line = _serialize_result(result)
+    sys.stdout.write(line + "\n")
+    sys.stdout.flush()
+
+
+@contextmanager
+def job_context(
+    *,
+    job_id: str | None = None,
+    skill: str | None = None,
+    run_mode: str | None = None,
+) -> Iterator[None]:
+    """Optional wrapper: sets job context; auto finish(failed) on unhandled exception."""
+    global _cached_job_id, _journal_path, _meta_written, _journal_file
+
+    saved_job_id = _cached_job_id
+    saved_journal_path = _journal_path
+    saved_meta = _meta_written
+    saved_file = _journal_file
+    saved_env_job = os.environ.get("JIANGCHANG_JOB_ID")
+    saved_env_mode = os.environ.get("JIANGCHANG_RUN_MODE")
+
+    if job_id:
+        os.environ["JIANGCHANG_JOB_ID"] = job_id
+        _cached_job_id = None
+        _journal_path = None
+        _meta_written = False
+        if _journal_file is not None:
+            try:
+                _journal_file.close()
+            except Exception:
+                pass
+            _journal_file = None
+    if run_mode:
+        os.environ["JIANGCHANG_RUN_MODE"] = run_mode
+
+    try:
+        yield
+    except JobStopped as exc:
+        finish(
+            status="failed",
+            message=str(exc) or "用户停止",
+            skill=skill,
+            error_code="JOB_STOPPED",
+        )
+        raise
+    except Exception as exc:
+        finish(status="failed", message=str(exc), skill=skill)
+        raise
+    finally:
+        _cached_job_id = saved_job_id
+        _journal_path = saved_journal_path
+        _meta_written = saved_meta
+        if _journal_file is not None and _journal_file is not saved_file:
+            try:
+                _journal_file.close()
+            except Exception:
+                pass
+        _journal_file = saved_file
+        if saved_env_job is None:
+            os.environ.pop("JIANGCHANG_JOB_ID", None)
+        else:
+            os.environ["JIANGCHANG_JOB_ID"] = saved_env_job
+        if saved_env_mode is None:
+            os.environ.pop("JIANGCHANG_RUN_MODE", None)
+        else:
+            os.environ["JIANGCHANG_RUN_MODE"] = saved_env_mode
+
+
+def rpa_step(name: str | None = None) -> Callable[[F], F]:
+    """Decorator for sync/async RPA steps: gate, emit enter/success/warn on failure."""
+
+    def decorator(fn: F) -> F:
+        step_name = name or fn.__name__
+
+        if inspect.iscoroutinefunction(fn):
+
+            @functools.wraps(fn)
+            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+                _step_gate(step_name)
+                _emit_core(f"▶ {step_name}")
+                try:
+                    result = await fn(*args, **kwargs)
+                    _emit_core(f"✓ {step_name}")
+                    return result
+                except JobStopped:
+                    raise
+                except Exception as exc:
+                    _emit_core(f"✗ {step_name}: {exc}", type="warn")
+                    raise
+
+            return async_wrapper  # type: ignore[return-value]
+
+        @functools.wraps(fn)
+        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
+            _step_gate(step_name)
+            _emit_core(f"▶ {step_name}")
+            try:
+                result = fn(*args, **kwargs)
+                _emit_core(f"✓ {step_name}")
+                return result
+            except JobStopped:
+                raise
+            except Exception as exc:
+                _emit_core(f"✗ {step_name}: {exc}", type="warn")
+                raise
+
+        return sync_wrapper  # type: ignore[return-value]
+
+    return decorator
+
+
+def _reset_for_tests() -> None:
+    """Clear module-level journal and control state (tests only)."""
+    global _cached_job_id, _journal_path, _meta_written, _journal_file
+    global _step_index, _current_step_name
+    if _journal_file is not None:
+        try:
+            _journal_file.close()
+        except Exception:
+            pass
+    _cached_job_id = None
+    _journal_path = None
+    _meta_written = False
+    _journal_file = None
+    _step_index = 0
+    _current_step_name = None

src/jiangchang_skill_core/media_assets.py

@@ -743,7 +743,7 @@ def probe_background_music(
         if not music_dir.is_dir():
             music_dir = Path(root) if Path(root).is_dir() else None
     else:
-        data_root = (env_map.get("JIANGCHANG_DATA_ROOT") or env_map.get("CLAW_DATA_ROOT") or "").strip()
+        data_root = (env_map.get("JIANGCHANG_DATA_ROOT") or "").strip()
         assets_root = resolve_media_assets_root(data_root or None, env_map)
         status = _inspect_media_assets(assets_root, source="local")
         music_dir = status.music_dir

src/jiangchang_skill_core/rpa/video_session.py

@@ -568,12 +568,14 @@ class RpaVideoSession:
         batch_id: str,
         title: str = "",
         closing_title: str = "",
+        emit_activity: bool = True,
     ) -> 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.closing_title = closing_title
+        self.emit_activity = emit_activity
         self.enabled = _record_video_enabled()
         self.warnings: List[str] = []
         self.output_video_path: Optional[str] = None
@@ -631,13 +633,25 @@ class RpaVideoSession:
         self.add_step(text, duration=duration)
 
     def add_step(self, text: str, *, duration: float = 4.0) -> None:
-        """记录一步骤字幕（相对录屏开始时间）。"""
+        """记录一步骤字幕（相对录屏开始时间）；可选同步写入 Activity Stream。"""
+        cleaned = text.strip()
+        if self.emit_activity and cleaned:
+            try:
+                from .. import activity
+
+                activity.emit(
+                    cleaned,
+                    skill=self.skill_slug,
+                    stage="rpa.video",
+                )
+            except Exception:
+                pass
         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))
+        self._steps.append(_StepEntry(start=elapsed, text=cleaned, duration=duration))
 
     def _close_ffmpeg_record_log(self) -> None:
         fh = self._ffmpeg_record_log_file

src/jiangchang_skill_core/runtime_diagnostics.py

@@ -37,7 +37,6 @@ class RuntimeDiagnostics:
     platform_kit_version_ok: bool | None
     jiangchang_skill_core_file: str | None
 
-    claw_data_root: str | None
     jiangchang_data_root: str | None
     resolved_data_root: str | None
 
@@ -66,7 +65,7 @@ def _env_dict(env: Mapping[str, str] | None) -> dict[str, str]:
 
 
 def _resolve_data_root(env: Mapping[str, str]) -> str:
-    root = (env.get("CLAW_DATA_ROOT") or env.get("JIANGCHANG_DATA_ROOT") or "").strip()
+    root = (env.get("JIANGCHANG_DATA_ROOT") or "").strip()
     if root:
         return root
     return platform_default_data_root()
@@ -164,7 +163,6 @@ def collect_runtime_diagnostics(
     env_map = _env_dict(env)
     issues: list[RuntimeIssue] = []
 
-    claw_data_root = env_map.get("CLAW_DATA_ROOT")
     jiangchang_data_root = env_map.get("JIANGCHANG_DATA_ROOT")
     resolved_data_root = _resolve_data_root(env_map)
 
@@ -267,7 +265,6 @@ def collect_runtime_diagnostics(
         platform_kit_min_version=platform_kit_min_version,
         platform_kit_version_ok=platform_ok,
         jiangchang_skill_core_file=core_file,
-        claw_data_root=claw_data_root,
         jiangchang_data_root=jiangchang_data_root,
         resolved_data_root=resolved_data_root,
         media_assets_root=str(media_root_path),
@@ -290,7 +287,6 @@ def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]:
         f"platform_kit_min_version: {diagnostics.platform_kit_min_version or ''}",
         f"platform_kit_version_ok: {diagnostics.platform_kit_version_ok}",
         f"jiangchang_skill_core_file: {diagnostics.jiangchang_skill_core_file or ''}",
-        f"CLAW_DATA_ROOT: {diagnostics.claw_data_root or ''}",
         f"JIANGCHANG_DATA_ROOT: {diagnostics.jiangchang_data_root or ''}",
         f"resolved_data_root: {diagnostics.resolved_data_root or ''}",
         f"media_assets_root: {diagnostics.media_assets_root or ''}",
@@ -315,7 +311,6 @@ def runtime_diagnostics_dict(diagnostics: RuntimeDiagnostics) -> dict[str, Any]:
         "platform_kit_min_version": diagnostics.platform_kit_min_version,
         "platform_kit_version_ok": diagnostics.platform_kit_version_ok,
         "jiangchang_skill_core_file": diagnostics.jiangchang_skill_core_file,
-        "CLAW_DATA_ROOT": diagnostics.claw_data_root,
         "JIANGCHANG_DATA_ROOT": diagnostics.jiangchang_data_root,
         "resolved_data_root": diagnostics.resolved_data_root,
         "media_assets_root": diagnostics.media_assets_root,

src/jiangchang_skill_core/runtime_env.py

@@ -34,22 +34,28 @@ def platform_default_data_root() -> str:
 
 
 def get_data_root() -> str:
-    env = (
+    env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
-        os.getenv("CLAW_DATA_ROOT")
-        or os.getenv("JIANGCHANG_DATA_ROOT")
-        or ""
-    ).strip()
     if env:
         return env
     return platform_default_data_root()
 
 
 def get_user_id() -> str:
-    return (
+    return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon"
-        os.getenv("CLAW_USER_ID")
+
-        or os.getenv("JIANGCHANG_USER_ID")
+
-        or ""
+def get_runs_dir() -> str:
-    ).strip() or "_anon"
+    """Run Journal 目录：{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/"""
+    return os.path.join(get_data_root(), ".jiangchang", "runs")
+
+
+def ensure_runs_dir() -> str:
+    path = get_runs_dir()
+    try:
+        os.makedirs(path, exist_ok=True)
+    except OSError:
+        pass
+    return path
 
 
 def _looks_like_skills_root(path: str) -> bool:
@@ -76,13 +82,11 @@ def get_skills_root() -> str:
 
     优先级：
     1) JIANGCHANG_SKILLS_ROOT
-    2) CLAW_SKILLS_ROOT
+    2) JIANGCHANG_APP_ROOT：若 {APP}/skills 存在且像技能根则用之；否则若 APP 下已有技能子目录则 APP 即根
-    3) JIANGCHANG_APP_ROOT：若 {APP}/skills 存在且像技能根则用之；否则若 APP 下已有技能子目录则 APP 即根
+    3) Windows：默认 D:\\AI\\jiangchang 同上规则
-    4) Windows：默认 D:\\AI\\jiangchang 同上规则
+    4) 其他平台：~/.openclaw/skills
-    5) 其他平台：~/.openclaw/skills
     """
-    for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
+    v = (os.getenv("JIANGCHANG_SKILLS_ROOT") or "").strip()
-        v = (os.getenv(key) or "").strip()
     if v:
         return os.path.normpath(v)
 
@@ -110,8 +114,8 @@ def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str:
     """
     编排子进程调用兄弟技能时用：先根据本技能 scripts 目录推断并列根；若目录下能识别出兄弟技能
     则用之（开发仓 D:\\OpenClaw 与安装目录 ~/.openclaw/skills 均满足「技能根之上一级 = 并列根」），
-    避免全局 JIANGCHANG_SKILLS_ROOT / CLAW_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。
+    避免全局 JIANGCHANG_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。
-    推断失败时再读上述环境变量，最后回落 get_skills_root()。
+    推断失败时再读 JIANGCHANG_SKILLS_ROOT，最后回落 get_skills_root()。
     """
     if skill_scripts_dir:
         scripts = os.path.abspath(skill_scripts_dir)
@@ -120,8 +124,7 @@ def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str:
         if _looks_like_skills_root(inferred):
             return os.path.normpath(inferred)
 
-    for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
+    v = (os.getenv("JIANGCHANG_SKILLS_ROOT") or "").strip()
-        v = (os.getenv(key) or "").strip()
     if v:
         return os.path.normpath(v)
 
@@ -139,21 +142,10 @@ def apply_cli_local_defaults() -> None:
         enabled = v in ("1", "true", "yes", "on")
     if not enabled:
         return
-    if not (
+    if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip():
-        os.getenv("CLAW_DATA_ROOT")
+        os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
-        or os.getenv("JIANGCHANG_DATA_ROOT")
+    if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
-        or ""
-    ).strip():
-        default_root = platform_default_data_root()
-        os.environ["JIANGCHANG_DATA_ROOT"] = default_root
-        os.environ.setdefault("CLAW_DATA_ROOT", default_root)
-    if not (
-        os.getenv("CLAW_USER_ID")
-        or os.getenv("JIANGCHANG_USER_ID")
-        or ""
-    ).strip():
         os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()
-        os.environ.setdefault("CLAW_USER_ID", DEFAULT_LOCAL_USER_ID.strip())
 def find_chrome_executable() -> "str | None":
     """查找系统已安装的 Chrome 或 Edge 可执行文件路径。
 

src/jiangchang_skill_core/sibling_bridge.py

@@ -0,0 +1,71 @@
+"""
+兄弟技能 CLI 桥：同步 subprocess 调用并列技能 scripts/main.py。
+
+account-manager 等兄弟技能为同步 CLI，非 async、非线程池。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from typing import Any, List, Optional, Sequence
+
+from .runtime_env import get_skills_root
+from .unified_logging import subprocess_env_with_trace
+
+
+def get_sibling_main_path(skill_slug: str) -> str:
+    return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
+
+
+def _python_exe() -> str:
+    return sys.executable
+
+
+def _run_sibling(skill_slug: str, argv: Sequence[str]) -> subprocess.CompletedProcess[str]:
+    main_path = get_sibling_main_path(skill_slug)
+    cmd = [_python_exe(), main_path, *argv]
+    return subprocess.run(
+        cmd,
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        env=subprocess_env_with_trace(),
+    )
+
+
+def _parse_json_from_stdout(stdout: str) -> Any:
+    for line in reversed((stdout or "").splitlines()):
+        stripped = line.strip()
+        if not stripped or stripped.startswith("ERROR:"):
+            continue
+        try:
+            return json.loads(stripped)
+        except json.JSONDecodeError:
+            continue
+    return None
+
+
+def call_sibling_json(skill_slug: str, argv: Sequence[str]) -> dict | None:
+    """Run sibling CLI; return last JSON object line from stdout, or None."""
+    result = _run_sibling(skill_slug, argv)
+    if result.returncode != 0:
+        return None
+    parsed = _parse_json_from_stdout(result.stdout or "")
+    if isinstance(parsed, dict):
+        return parsed
+    return None
+
+
+def call_sibling_json_array(skill_slug: str, argv: Sequence[str]) -> list | None:
+    """Run sibling CLI; return last JSON array line from stdout, or None."""
+    result = _run_sibling(skill_slug, argv)
+    if result.returncode != 0:
+        return None
+    parsed = _parse_json_from_stdout(result.stdout or "")
+    if isinstance(parsed, list):
+        return parsed
+    return None

tests/test_activity_control.py

@@ -0,0 +1,218 @@
+# -*- coding: utf-8 -*-
+"""SRCP v1: control.json, checkpoint, pause/resume/stop, interruptible_sleep."""
+from __future__ import annotations
+
+import asyncio
+import io
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from jiangchang_skill_core import activity
+from jiangchang_skill_core.activity import JobStopped
+from jiangchang_skill_core.runtime_env import get_runs_dir
+
+
+class _ControlEnvIsolation(unittest.TestCase):
+    _KEYS = (
+        "JIANGCHANG_DATA_ROOT",
+        "JIANGCHANG_USER_ID",
+        "JIANGCHANG_JOB_ID",
+        "JIANGCHANG_RUN_MODE",
+        "JIANGCHANG_CONTROL_DISABLED",
+        "JIANGCHANG_CONTROL_POLL_SEC",
+    )
+
+    def setUp(self) -> None:
+        self._saved = {k: os.environ.get(k) for k in self._KEYS}
+        self._tmps: list[str] = []
+        for k in self._KEYS:
+            os.environ.pop(k, None)
+        activity._reset_for_tests()
+
+    def tearDown(self) -> None:
+        activity._reset_for_tests()
+        for k, v in self._saved.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        for tmp in self._tmps:
+            shutil.rmtree(tmp, ignore_errors=True)
+
+    def _tmp_data_root(self) -> str:
+        tmp = tempfile.mkdtemp()
+        self._tmps.append(tmp)
+        return tmp
+
+    def _write_control(self, job_id: str, command: str, *, nonce: int = 1) -> None:
+        path = Path(get_runs_dir()) / f"{job_id}.control.json"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(
+            json.dumps(
+                {
+                    "schema_version": 1,
+                    "command": command,
+                    "reason": "user",
+                    "ts": 1_700_000_000_000,
+                    "nonce": nonce,
+                },
+                ensure_ascii=False,
+            ),
+            encoding="utf-8",
+        )
+
+    def _journal_events(self, job_id: str) -> list[dict]:
+        journal = Path(get_runs_dir()) / f"{job_id}.jsonl"
+        if not journal.is_file():
+            return []
+        return [
+            json.loads(line)
+            for line in journal.read_text(encoding="utf-8").strip().splitlines()
+            if line.strip()
+        ]
+
+
+class TestControlProtocol(_ControlEnvIsolation):
+    def test_get_control_path(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-path"
+        activity._reset_for_tests()
+        self.assertTrue(activity.get_control_path().endswith("ctrl-path.control.json"))
+
+    def test_stop_raises_job_stopped(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-stop"
+        os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05"
+        activity._reset_for_tests()
+        self._write_control("ctrl-stop", "stop")
+
+        with self.assertRaises(JobStopped):
+            activity.checkpoint()
+
+        events = self._journal_events("ctrl-stop")
+        lifecycle = [e for e in events if e.get("type") == "lifecycle"]
+        self.assertTrue(any(e.get("status") == "stopping" for e in lifecycle))
+
+    def test_emit_stop_before_activity_write(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-emit-stop"
+        activity._reset_for_tests()
+        self._write_control("ctrl-emit-stop", "stop")
+
+        with self.assertRaises(JobStopped):
+            activity.emit("不应写入")
+
+        events = self._journal_events("ctrl-emit-stop")
+        self.assertFalse(any(e.get("type") == "activity" for e in events))
+        self.assertTrue(any(e.get("type") == "lifecycle" and e.get("status") == "stopping" for e in events))
+
+    def test_pause_resume_via_emit(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-pause"
+        os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05"
+        activity._reset_for_tests()
+        self._write_control("ctrl-pause", "pause")
+
+        import threading
+        import time as time_mod
+
+        def resume_later() -> None:
+            time_mod.sleep(0.15)
+            self._write_control("ctrl-pause", "resume", nonce=2)
+
+        threading.Thread(target=resume_later, daemon=True).start()
+        activity.emit("步骤一", skill="demo")
+
+        events = self._journal_events("ctrl-pause")
+        statuses = [e.get("status") for e in events if e.get("type") == "lifecycle"]
+        self.assertIn("paused", statuses)
+        self.assertIn("resumed", statuses)
+        activity_events = [e for e in events if e.get("type") == "activity"]
+        self.assertEqual(activity_events[-1]["text"], "步骤一")
+        lifecycle = [e for e in events if e.get("type") == "lifecycle"]
+        self.assertEqual(lifecycle[0].get("step_index"), 1)
+
+    def test_rpa_step_gate_increments_step_index(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-rpa"
+        activity._reset_for_tests()
+
+        @activity.rpa_step("打开页面")
+        def open_page() -> None:
+            return None
+
+        open_page()
+        events = self._journal_events("ctrl-rpa")
+        lifecycle_or_activity = [e for e in events if e.get("type") in ("activity", "lifecycle")]
+        self.assertGreaterEqual(len(lifecycle_or_activity), 2)
+
+    def test_job_context_finishes_on_job_stopped(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        activity._reset_for_tests()
+
+        with patch("sys.stdout", io.StringIO()):
+            with self.assertRaises(JobStopped):
+                with activity.job_context(job_id="ctx-stop", skill="demo-skill"):
+                    self._write_control("ctx-stop", "stop")
+                    activity.checkpoint()
+
+        journal = Path(get_runs_dir()) / "ctx-stop.jsonl"
+        lines = journal.read_text(encoding="utf-8").strip().splitlines()
+        last = json.loads(lines[-1])
+        self.assertEqual(last["type"], "error")
+        self.assertEqual(last.get("error_code"), "JOB_STOPPED")
+
+    def test_interruptible_sleep_honors_stop(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-sleep"
+        os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05"
+        activity._reset_for_tests()
+        self._write_control("ctrl-sleep", "stop")
+
+        async def run() -> None:
+            with self.assertRaises(JobStopped):
+                await activity.interruptible_sleep(5.0)
+
+        asyncio.run(run())
+
+    def test_interruptible_sleep_sync_honors_pause(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "ctrl-sleep-sync"
+        os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05"
+        activity._reset_for_tests()
+        self._write_control("ctrl-sleep-sync", "pause")
+
+        def resume_later() -> None:
+            import threading
+            import time
+
+            def _resume() -> None:
+                time.sleep(0.15)
+                self._write_control("ctrl-sleep-sync", "resume", nonce=2)
+
+            threading.Thread(target=_resume, daemon=True).start()
+
+        resume_later()
+        activity.interruptible_sleep_sync(0.4, chunk=0.05)
+
+        events = self._journal_events("ctrl-sleep-sync")
+        statuses = [e.get("status") for e in events if e.get("type") == "lifecycle"]
+        self.assertIn("paused", statuses)
+        self.assertIn("resumed", statuses)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_activity_emit_finish.py

@@ -0,0 +1,209 @@
+# -*- coding: utf-8 -*-
+"""Skill Activity Stream: journal, flush, job_id, finish stdout, truncation."""
+from __future__ import annotations
+
+import io
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from jiangchang_skill_core import activity
+from jiangchang_skill_core.runtime_env import get_runs_dir
+
+
+class _ActivityEnvIsolation(unittest.TestCase):
+    _KEYS = (
+        "JIANGCHANG_DATA_ROOT",
+        "JIANGCHANG_USER_ID",
+        "JIANGCHANG_JOB_ID",
+        "JIANGCHANG_RUN_MODE",
+        "JIANGCHANG_ACTIVITY_DEBUG",
+    )
+
+    def setUp(self) -> None:
+        self._saved = {k: os.environ.get(k) for k in self._KEYS}
+        self._tmps: list[str] = []
+        for k in self._KEYS:
+            os.environ.pop(k, None)
+        activity._reset_for_tests()
+
+    def tearDown(self) -> None:
+        activity._reset_for_tests()
+        for k, v in self._saved.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        for tmp in self._tmps:
+            shutil.rmtree(tmp, ignore_errors=True)
+
+    def _tmp_data_root(self) -> str:
+        tmp = tempfile.mkdtemp()
+        self._tmps.append(tmp)
+        return tmp
+
+
+class TestActivityJournal(_ActivityEnvIsolation):
+    def test_emit_writes_ndjson_with_meta(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-test-1"
+        activity._reset_for_tests()
+
+        activity.emit("步骤一", type="activity", skill="geo-batch")
+        journal = Path(get_runs_dir()) / "job-test-1.jsonl"
+        self.assertTrue(journal.is_file())
+        lines = journal.read_text(encoding="utf-8").strip().splitlines()
+        self.assertGreaterEqual(len(lines), 2)
+        meta = json.loads(lines[0])
+        self.assertEqual(meta["type"], "meta")
+        self.assertEqual(meta["job_id"], "job-test-1")
+        event = json.loads(lines[1])
+        self.assertEqual(event["type"], "activity")
+        self.assertEqual(event["text"], "步骤一")
+        self.assertEqual(event["schema_version"], 1)
+        self.assertEqual(event["skill"], "geo-batch")
+        self.assertIn("ts", event)
+
+    def test_emit_flushes_immediately(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-flush"
+        activity._reset_for_tests()
+        activity.emit("flush test")
+        journal = Path(get_runs_dir()) / "job-flush.jsonl"
+        self.assertGreater(journal.stat().st_size, 0)
+
+    def test_job_id_from_env(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "env-job-42"
+        activity._reset_for_tests()
+        self.assertEqual(activity.get_job_id(), "env-job-42")
+
+    def test_job_id_generated_and_cached(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        activity._reset_for_tests()
+        jid1 = activity.get_job_id()
+        jid2 = activity.get_job_id()
+        self.assertEqual(jid1, jid2)
+        self.assertEqual(len(jid1), 32)
+
+    def test_debug_skipped_without_env(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-no-debug"
+        activity._reset_for_tests()
+        activity.emit("hidden", type="debug")
+        journal = Path(get_runs_dir()) / "job-no-debug.jsonl"
+        self.assertFalse(journal.exists())
+
+    def test_debug_written_when_enabled(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-debug"
+        os.environ["JIANGCHANG_ACTIVITY_DEBUG"] = "1"
+        activity._reset_for_tests()
+        activity.emit("dbg", type="debug")
+        journal = Path(get_runs_dir()) / "job-debug.jsonl"
+        content = journal.read_text(encoding="utf-8")
+        self.assertIn('"type":"debug"', content.replace(" ", ""))
+
+    def test_emit_never_prints_stdout(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-quiet"
+        activity._reset_for_tests()
+        buf = io.StringIO()
+        with patch("sys.stdout", buf):
+            activity.emit("silent step")
+        self.assertEqual(buf.getvalue(), "")
+
+    def test_get_run_mode_default_cli(self) -> None:
+        self.assertEqual(activity.get_run_mode(), "cli")
+
+    def test_get_run_mode_job(self) -> None:
+        os.environ["JIANGCHANG_RUN_MODE"] = "job"
+        self.assertEqual(activity.get_run_mode(), "job")
+
+
+class TestActivityFinish(_ActivityEnvIsolation):
+    def test_finish_stdout_single_line_result(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-finish"
+        activity._reset_for_tests()
+
+        buf = io.StringIO()
+        with patch("sys.stdout", buf):
+            activity.finish(status="success", message="ok", skill="geo", generated=3)
+
+        lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()]
+        self.assertEqual(len(lines), 1)
+        result = json.loads(lines[0])
+        self.assertEqual(result["type"], "result")
+        self.assertEqual(result["schema_version"], 1)
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(result["skill"], "geo")
+        self.assertEqual(result["generated"], 3)
+
+        journal = Path(get_runs_dir()) / "job-finish.jsonl"
+        journal_lines = journal.read_text(encoding="utf-8").strip().splitlines()
+        last = json.loads(journal_lines[-1])
+        self.assertEqual(last["type"], "done")
+        self.assertEqual(last["status"], "success")
+
+    def test_finish_failed_writes_error_event(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-fail"
+        activity._reset_for_tests()
+
+        with patch("sys.stdout", io.StringIO()):
+            activity.finish(status="failed", message="boom")
+
+        journal = Path(get_runs_dir()) / "job-fail.jsonl"
+        last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1])
+        self.assertEqual(last["type"], "error")
+
+    def test_finish_truncates_large_result(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-trunc"
+        activity._reset_for_tests()
+
+        big = {f"field_{i}": "x" * 200 for i in range(50)}
+        buf = io.StringIO()
+        with patch("sys.stdout", buf):
+            activity.finish(status="success", skill="big", **big)
+
+        line = buf.getvalue().strip()
+        self.assertLessEqual(len(line), 4096)
+        result = json.loads(line)
+        self.assertTrue(result.get("truncated"))
+
+
+class TestJobContext(_ActivityEnvIsolation):
+    def test_job_context_auto_finish_on_exception(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        activity._reset_for_tests()
+
+        with patch("sys.stdout", io.StringIO()):
+            with self.assertRaises(ValueError):
+                with activity.job_context(job_id="ctx-err", skill="test-skill"):
+                    raise ValueError("broken")
+
+        journal = Path(get_runs_dir()) / "ctx-err.jsonl"
+        last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1])
+        self.assertEqual(last["type"], "error")
+        self.assertIn("broken", last["message"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_activity_rpa_step_decorator.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""rpa_step decorator: sync + async."""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from jiangchang_skill_core import activity
+from jiangchang_skill_core.runtime_env import get_runs_dir
+
+
+class _DecoratorEnvIsolation(unittest.TestCase):
+    def setUp(self) -> None:
+        self._saved_root = os.environ.get("JIANGCHANG_DATA_ROOT")
+        self._saved_job = os.environ.get("JIANGCHANG_JOB_ID")
+        self._tmps: list[str] = []
+        os.environ.pop("JIANGCHANG_JOB_ID", None)
+        activity._reset_for_tests()
+
+    def tearDown(self) -> None:
+        activity._reset_for_tests()
+        if self._saved_root is None:
+            os.environ.pop("JIANGCHANG_DATA_ROOT", None)
+        else:
+            os.environ["JIANGCHANG_DATA_ROOT"] = self._saved_root
+        if self._saved_job is None:
+            os.environ.pop("JIANGCHANG_JOB_ID", None)
+        else:
+            os.environ["JIANGCHANG_JOB_ID"] = self._saved_job
+        for tmp in self._tmps:
+            shutil.rmtree(tmp, ignore_errors=True)
+
+    def _tmp_data_root(self) -> str:
+        tmp = tempfile.mkdtemp()
+        self._tmps.append(tmp)
+        return tmp
+
+
+class TestRpaStepDecorator(_DecoratorEnvIsolation):
+    def _journal_texts(self, job_id: str) -> list[str]:
+        journal = Path(get_runs_dir()) / f"{job_id}.jsonl"
+        lines = journal.read_text(encoding="utf-8").strip().splitlines()
+        events = [json.loads(ln) for ln in lines if json.loads(ln).get("type") != "meta"]
+        return [e["text"] for e in events]
+
+    def test_sync_success(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "rpa-sync-ok"
+        activity._reset_for_tests()
+
+        @activity.rpa_step("打开浏览器")
+        def open_browser() -> str:
+            return "ok"
+
+        self.assertEqual(open_browser(), "ok")
+        texts = self._journal_texts("rpa-sync-ok")
+        self.assertIn("▶ 打开浏览器", texts)
+        self.assertIn("✓ 打开浏览器", texts)
+
+    def test_sync_failure_emits_warn_and_reraises(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "rpa-sync-fail"
+        activity._reset_for_tests()
+
+        @activity.rpa_step("提交表单")
+        def submit() -> None:
+            raise RuntimeError("timeout")
+
+        with self.assertRaises(RuntimeError):
+            submit()
+
+        journal = Path(get_runs_dir()) / "rpa-sync-fail.jsonl"
+        content = journal.read_text(encoding="utf-8")
+        self.assertIn("timeout", content)
+        self.assertIn('"type":"warn"', content.replace(" ", ""))
+
+    def test_async_success(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "rpa-async-ok"
+        activity._reset_for_tests()
+
+        @activity.rpa_step("异步步骤")
+        async def async_step() -> int:
+            await asyncio.sleep(0)
+            return 42
+
+        result = asyncio.run(async_step())
+        self.assertEqual(result, 42)
+        texts = self._journal_texts("rpa-async-ok")
+        self.assertIn("▶ 异步步骤", texts)
+        self.assertIn("✓ 异步步骤", texts)
+
+    def test_async_failure(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "rpa-async-fail"
+        activity._reset_for_tests()
+
+        @activity.rpa_step()
+        async def fail_step() -> None:
+            await asyncio.sleep(0)
+            raise ValueError("async err")
+
+        with self.assertRaises(ValueError):
+            asyncio.run(fail_step())
+
+        journal = Path(get_runs_dir()) / "rpa-async-fail.jsonl"
+        self.assertIn("async err", journal.read_text(encoding="utf-8"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_runtime_diagnostics.py

@@ -57,7 +57,7 @@ def test_collect_runtime_diagnostics_json_serializable(
 ) -> None:
     data_root = tmp_path / "data"
     env = {
-        "CLAW_DATA_ROOT": str(data_root),
+        "JIANGCHANG_DATA_ROOT": str(data_root),
         "OPENCLAW_RECORD_VIDEO": "0",
     }
     media_root = data_root / "shared" / "media-assets"
@@ -110,7 +110,7 @@ def test_platform_kit_min_version_none_skips_version_issue(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}
     monkeypatch.setattr(
         "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
         lambda *a, **k: ma.MediaAssetsStatus(
@@ -144,7 +144,7 @@ def test_platform_kit_missing_records_issue(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}
     monkeypatch.setattr(
         "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
         lambda *a, **k: ma.MediaAssetsStatus(
@@ -182,7 +182,7 @@ def test_record_video_ffmpeg_and_music_warnings_not_fatal(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     media_root = tmp_path / "shared" / "media-assets"
-    env = {"CLAW_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"}
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"}
     monkeypatch.setattr(
         "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
         lambda *a, **k: ma.MediaAssetsStatus(
@@ -225,7 +225,7 @@ def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None:
     fake_core.parent.mkdir(parents=True)
     fake_core.write_text("# vendored\n", encoding="utf-8")
 
-    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}
     with patch.object(jiangchang_skill_core, "__file__", str(fake_core)):
         diag = collect_runtime_diagnostics(
             skill_slug="my-skill",
@@ -236,11 +236,56 @@ def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None:
     assert "jiangchang_skill_core_loaded_from_skill_tree" in diag.issue_codes()
 
 
+def _legacy_env_key(suffix: str) -> str:
+    return "CL" + "AW_" + suffix
+
+
+def test_format_runtime_health_lines_no_legacy_data_root_key(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    env = {
+        "JIANGCHANG_DATA_ROOT": str(tmp_path / "data"),
+        _legacy_env_key("DATA_ROOT"): str(tmp_path / "legacy-only"),
+    }
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
+        lambda *a, **k: ma.MediaAssetsStatus(
+            root=tmp_path,
+            exists=False,
+            ready=False,
+            source="local",
+        ),
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
+        lambda *a, **k: {
+            "music_root": None,
+            "mp3_count": 0,
+            "usable_count": 0,
+            "sample_path": None,
+            "issue": None,
+        },
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
+        lambda: "1.0.17",
+    )
+
+    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
+    payload = runtime_diagnostics_dict(diag)
+    text = "\n".join(format_runtime_health_lines(diag))
+    legacy_key = _legacy_env_key("DATA_ROOT")
+    assert legacy_key not in payload
+    assert legacy_key not in text
+    assert payload["resolved_data_root"] == str(tmp_path / "data")
+
+
 def test_format_runtime_health_lines_core_fields(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}
     monkeypatch.setattr(
         "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
         lambda *a, **k: ma.MediaAssetsStatus(
@@ -301,7 +346,6 @@ def test_runtime_diagnostics_frozen_dataclass() -> None:
         platform_kit_min_version=None,
         platform_kit_version_ok=None,
         jiangchang_skill_core_file=None,
-        claw_data_root=None,
         jiangchang_data_root=None,
         resolved_data_root="/tmp",
         media_assets_root="/tmp/media",
@@ -326,7 +370,6 @@ def test_format_runtime_health_lines_issue_separator_ascii() -> None:
         platform_kit_min_version=None,
         platform_kit_version_ok=None,
         jiangchang_skill_core_file=None,
-        claw_data_root=None,
         jiangchang_data_root=None,
         resolved_data_root="/tmp",
         media_assets_root="/tmp/media",

tests/test_runtime_env.py

@@ -0,0 +1,204 @@
+# -*- coding: utf-8 -*-
+"""runtime_env：JIANGCHANG_* 路径契约与兄弟技能根解析。"""
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from jiangchang_skill_core import config, unified_logging as ul
+from jiangchang_skill_core.runtime_diagnostics import (
+    collect_runtime_diagnostics,
+    format_runtime_health_lines,
+    runtime_diagnostics_dict,
+)
+from jiangchang_skill_core.runtime_env import (
+    apply_cli_local_defaults,
+    get_data_root,
+    get_sibling_skills_root,
+    get_skills_root,
+    get_user_id,
+    platform_default_data_root,
+)
+
+
+def _legacy_env_key(suffix: str) -> str:
+    return "CL" + "AW_" + suffix
+
+
+class _EnvIsolation(unittest.TestCase):
+    _KEYS = (
+        "JIANGCHANG_DATA_ROOT",
+        "JIANGCHANG_USER_ID",
+        "JIANGCHANG_SKILLS_ROOT",
+        "JIANGCHANG_APP_ROOT",
+        _legacy_env_key("DATA_ROOT"),
+        _legacy_env_key("USER_ID"),
+        _legacy_env_key("SKILLS_ROOT"),
+        "JIANGCHANG_CLI_LOCAL_DEV",
+    )
+
+    def setUp(self) -> None:
+        self._saved = {k: os.environ.get(k) for k in self._KEYS}
+        for k in self._KEYS:
+            os.environ.pop(k, None)
+        config.reset_cache()
+
+    def tearDown(self) -> None:
+        for k, v in self._saved.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        config.reset_cache()
+
+
+class TestGetDataRoot(_EnvIsolation):
+    def test_jiangchang_data_root_only(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            self.assertEqual(get_data_root(), tmp)
+
+    def test_platform_default_when_unset(self) -> None:
+        self.assertEqual(get_data_root(), platform_default_data_root())
+
+    def test_legacy_claw_data_root_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp:
+            os.environ[_legacy_env_key("DATA_ROOT")] = claw_tmp
+            os.environ["JIANGCHANG_DATA_ROOT"] = jc_tmp
+            self.assertEqual(get_data_root(), jc_tmp)
+
+    def test_legacy_claw_data_root_alone_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as claw_tmp:
+            os.environ[_legacy_env_key("DATA_ROOT")] = claw_tmp
+            self.assertEqual(get_data_root(), platform_default_data_root())
+
+
+class TestGetUserId(_EnvIsolation):
+    def test_jiangchang_user_id_only(self) -> None:
+        os.environ["JIANGCHANG_USER_ID"] = "user-42"
+        self.assertEqual(get_user_id(), "user-42")
+
+    def test_anon_when_unset(self) -> None:
+        self.assertEqual(get_user_id(), "_anon")
+
+    def test_legacy_claw_user_id_ignored(self) -> None:
+        os.environ[_legacy_env_key("USER_ID")] = "claw-only"
+        os.environ["JIANGCHANG_USER_ID"] = "jc-user"
+        self.assertEqual(get_user_id(), "jc-user")
+
+    def test_legacy_claw_user_id_alone_ignored(self) -> None:
+        os.environ[_legacy_env_key("USER_ID")] = "claw-only"
+        self.assertEqual(get_user_id(), "_anon")
+
+
+class TestApplyCliLocalDefaults(_EnvIsolation):
+    def test_sets_jiangchang_only(self) -> None:
+        with patch("jiangchang_skill_core.runtime_env.CLI_LOCAL_DEV_ENABLED", True):
+            apply_cli_local_defaults()
+        self.assertTrue(os.environ.get("JIANGCHANG_DATA_ROOT"))
+        self.assertTrue(os.environ.get("JIANGCHANG_USER_ID"))
+        self.assertNotIn(_legacy_env_key("DATA_ROOT"), os.environ)
+        self.assertNotIn(_legacy_env_key("USER_ID"), os.environ)
+
+    def test_does_not_override_existing(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "existing"
+            with patch("jiangchang_skill_core.runtime_env.CLI_LOCAL_DEV_ENABLED", True):
+                apply_cli_local_defaults()
+            self.assertEqual(os.environ["JIANGCHANG_DATA_ROOT"], tmp)
+            self.assertEqual(os.environ["JIANGCHANG_USER_ID"], "existing")
+
+
+class TestGetSkillsRoot(_EnvIsolation):
+    def test_jiangchang_skills_root_override(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_SKILLS_ROOT"] = tmp
+            self.assertEqual(get_skills_root(), os.path.normpath(tmp))
+
+    def test_legacy_claw_skills_root_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp:
+            os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp
+            os.environ["JIANGCHANG_SKILLS_ROOT"] = jc_tmp
+            self.assertEqual(get_skills_root(), os.path.normpath(jc_tmp))
+
+    def test_legacy_claw_skills_root_alone_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as claw_tmp:
+            os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp
+            root = get_skills_root()
+            self.assertNotEqual(root, os.path.normpath(claw_tmp))
+
+
+class TestGetSiblingSkillsRoot(_EnvIsolation):
+    def _make_skills_tree(self, base: Path) -> Path:
+        skills_root = base / "skills"
+        (skills_root / "account-manager").mkdir(parents=True)
+        (skills_root / "receive-order" / "scripts").mkdir(parents=True)
+        return skills_root
+
+    def test_inference_preferred_over_env(self) -> None:
+        with tempfile.TemporaryDirectory() as tree_tmp, tempfile.TemporaryDirectory() as env_tmp:
+            skills_root = self._make_skills_tree(Path(tree_tmp))
+            scripts_dir = str(skills_root / "receive-order" / "scripts")
+            os.environ["JIANGCHANG_SKILLS_ROOT"] = env_tmp
+            result = get_sibling_skills_root(scripts_dir)
+            self.assertEqual(result, os.path.normpath(str(skills_root)))
+
+    def test_env_override_when_inference_fails(self) -> None:
+        with tempfile.TemporaryDirectory() as env_tmp:
+            os.environ["JIANGCHANG_SKILLS_ROOT"] = env_tmp
+            result = get_sibling_skills_root("/nonexistent/skill/scripts")
+            self.assertEqual(result, os.path.normpath(env_tmp))
+
+    def test_legacy_claw_skills_root_ignored_for_sibling(self) -> None:
+        with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp:
+            os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp
+            os.environ["JIANGCHANG_SKILLS_ROOT"] = jc_tmp
+            result = get_sibling_skills_root("/nonexistent/skill/scripts")
+            self.assertEqual(result, os.path.normpath(jc_tmp))
+
+
+class TestConfigAndLoggingPaths(_EnvIsolation):
+    def test_config_env_file_under_user_isolated_dir(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "u-test"
+            example = os.path.join(tmp, ".env.example")
+            with open(example, "w", encoding="utf-8") as f:
+                f.write("FOO=bar\n")
+            dest = config.ensure_env_file("my-skill", example)
+            expected_prefix = os.path.join(tmp, "u-test", "my-skill")
+            self.assertTrue(os.path.normpath(dest).startswith(os.path.normpath(expected_prefix)))
+
+    def test_unified_logging_path_under_user_dir(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            os.environ["JIANGCHANG_USER_ID"] = "log-user"
+            log_dir = ul.get_unified_logs_dir()
+            self.assertTrue(
+                os.path.normpath(log_dir).startswith(
+                    os.path.normpath(os.path.join(tmp, "log-user", "logs"))
+                )
+            )
+
+
+class TestDiagnosticsNoClawFields(_EnvIsolation):
+    def test_diagnostics_json_and_text_exclude_claw(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+            diag = collect_runtime_diagnostics(skill_slug="x", env=dict(os.environ))
+            payload = runtime_diagnostics_dict(diag)
+            text = "\n".join(format_runtime_health_lines(diag))
+            legacy_data_root_key = _legacy_env_key("DATA_ROOT")
+            self.assertNotIn(legacy_data_root_key, payload)
+            self.assertNotIn(legacy_data_root_key, text)
+            self.assertIn("JIANGCHANG_DATA_ROOT", payload)
+            self.assertEqual(payload["resolved_data_root"], tmp)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_sibling_bridge.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+"""sibling_bridge: mock subprocess sibling CLI calls."""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
+
+from jiangchang_skill_core.sibling_bridge import (
+    call_sibling_json,
+    call_sibling_json_array,
+    get_sibling_main_path,
+)
+
+
+class TestSiblingBridge(unittest.TestCase):
+    def test_get_sibling_main_path(self) -> None:
+        with patch.dict(os.environ, {"JIANGCHANG_SKILLS_ROOT": r"D:\skills"}):
+            path = get_sibling_main_path("account-manager")
+            self.assertTrue(path.replace("\\", "/").endswith("account-manager/scripts/main.py"))
+
+    @patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
+    def test_call_sibling_json_success(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout='log line\n{"id": 1, "name": "test"}\n',
+            stderr="",
+        )
+        result = call_sibling_json("account-manager", ["list", "all"])
+        self.assertEqual(result, {"id": 1, "name": "test"})
+        cmd = mock_run.call_args[0][0]
+        self.assertIn("account-manager", cmd[1])
+        self.assertIn("list", cmd)
+        self.assertTrue(mock_run.call_args[1]["capture_output"])
+
+    @patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
+    def test_call_sibling_json_nonzero_exit(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="ERROR:fail"
+        )
+        self.assertIsNone(call_sibling_json("content-manager", ["get", "x"]))
+
+    @patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
+    def test_call_sibling_json_array(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout=json.dumps([{"a": 1}, {"b": 2}]),
+            stderr="",
+        )
+        result = call_sibling_json_array("account-manager", ["list", "all"])
+        self.assertEqual(result, [{"a": 1}, {"b": 2}])
+
+    @patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
+    def test_uses_subprocess_env_with_trace(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = subprocess.CompletedProcess(
+            args=[], returncode=0, stdout='{"ok":true}', stderr=""
+        )
+        with patch(
+            "jiangchang_skill_core.sibling_bridge.subprocess_env_with_trace",
+            return_value={"JIANGCHANG_TRACE_ID": "abc123"},
+        ) as mock_env:
+            call_sibling_json("api-key-vault", ["get", "key"])
+        self.assertEqual(mock_run.call_args[1]["env"], {"JIANGCHANG_TRACE_ID": "abc123"})
+        mock_env.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_video_session_activity_integration.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+"""RpaVideoSession.add_step integrates with activity.emit."""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from jiangchang_skill_core import activity, config
+from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+from jiangchang_skill_core.runtime_env import get_runs_dir
+
+
+class TestVideoSessionActivityIntegration(unittest.TestCase):
+    def setUp(self) -> None:
+        config.reset_cache()
+        self._saved_root = os.environ.get("JIANGCHANG_DATA_ROOT")
+        self._saved_job = os.environ.get("JIANGCHANG_JOB_ID")
+        self._saved_video = os.environ.get("OPENCLAW_RECORD_VIDEO")
+        self._tmps: list[str] = []
+        activity._reset_for_tests()
+
+    def tearDown(self) -> None:
+        activity._reset_for_tests()
+        if self._saved_root is None:
+            os.environ.pop("JIANGCHANG_DATA_ROOT", None)
+        else:
+            os.environ["JIANGCHANG_DATA_ROOT"] = self._saved_root
+        if self._saved_job is None:
+            os.environ.pop("JIANGCHANG_JOB_ID", None)
+        else:
+            os.environ["JIANGCHANG_JOB_ID"] = self._saved_job
+        if self._saved_video is None:
+            os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        else:
+            os.environ["OPENCLAW_RECORD_VIDEO"] = self._saved_video
+        config.reset_cache()
+        for tmp in self._tmps:
+            shutil.rmtree(tmp, ignore_errors=True)
+
+    def _tmp_data_root(self) -> str:
+        tmp = tempfile.mkdtemp()
+        self._tmps.append(tmp)
+        return tmp
+
+    def test_add_step_emits_activity_when_disabled(self) -> None:
+        os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "video-act-1"
+        activity._reset_for_tests()
+        config.reset_cache()
+
+        session = RpaVideoSession(
+            skill_slug="publish-toutiao",
+            skill_data_dir=tmp,
+            batch_id="batch1",
+        )
+        self.assertFalse(session.enabled)
+        session.add_step("打开浏览器")
+
+        journal = Path(get_runs_dir()) / "video-act-1.jsonl"
+        self.assertTrue(journal.is_file())
+        content = journal.read_text(encoding="utf-8")
+        self.assertIn("打开浏览器", content)
+        self.assertIn("rpa.video", content)
+        self.assertIn("publish-toutiao", content)
+
+    def test_emit_activity_false_skips_journal(self) -> None:
+        os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "video-act-off"
+        activity._reset_for_tests()
+
+        session = RpaVideoSession(
+            skill_slug="publish-toutiao",
+            skill_data_dir=tmp,
+            batch_id="batch2",
+            emit_activity=False,
+        )
+        session.add_step("不应写入")
+        journal = Path(get_runs_dir()) / "video-act-off.jsonl"
+        self.assertFalse(journal.exists())
+
+    def test_add_step_writes_both_journal_and_subtitle_steps(self) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "video-act-both"
+        activity._reset_for_tests()
+        config.reset_cache()
+
+        session = RpaVideoSession(
+            skill_slug="receive-order",
+            skill_data_dir=tmp,
+            batch_id="batch3",
+        )
+        session.add_step("录制步骤")
+        self.assertEqual(len(session._steps), 1)
+
+        journal = Path(get_runs_dir()) / "video-act-both.jsonl"
+        lines = [json.loads(ln) for ln in journal.read_text(encoding="utf-8").splitlines()]
+        activity_events = [e for e in lines if e.get("type") == "activity"]
+        self.assertEqual(len(activity_events), 1)
+        self.assertEqual(activity_events[0]["text"], "录制步骤")
+
+
+if __name__ == "__main__":
+    unittest.main()