chore: update skill release readme source

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

.github/workflows/publish.yml

@@ -4,9 +4,6 @@ on:
   push:
     branches:
       - main
-    tags:
-      - "v*"
-  workflow_dispatch:
 
 jobs:
   publish:
@@ -25,13 +22,6 @@ jobs:
     steps:
       - uses: https://git.jc2009.com/admin/actions-checkout@v4
 
-      - name: Resolve and patch package version
-        env:
-          GITHUB_REF: ${{ github.ref }}
-          GITHUB_RUN_NUMBER: ${{ github.run_number }}
-          GITHUB_RUN_ID: ${{ github.run_id }}
-        run: python3.12 tools/ci_set_package_version.py
-
       - name: Install build tools
         run: python3.12 -m pip install --upgrade build twine --break-system-packages
 

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

@@ -108,6 +108,13 @@ jobs:
             '        shutil.copytree(src, dst, ignore=ignore_fn, dirs_exist_ok=True)' \
             '' \
             "shutil.copy2('SKILL.md', os.path.join(PACKAGE, 'SKILL.md'))" \
+            "readme_src = os.path.abspath('README.md')" \
+            "readme_dst = os.path.join(PACKAGE, 'README.md')" \
+            'if os.path.isfile(readme_src):' \
+            "    shutil.copy2(readme_src, readme_dst)" \
+            "    print('Copied README.md')" \
+            'if os.path.isfile(readme_src) and not os.path.isfile(readme_dst):' \
+            "    raise RuntimeError('README.md exists in skill root but was not packaged')" \
             "copy_dir('references', os.path.join(PACKAGE, 'references'), references_ignore)" \
             "copy_dir('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \
             "copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \
@@ -144,7 +151,7 @@ jobs:
           skill_readme_md = (post.content or '').strip()
           skill_description = metadata.get('description')
           metadata['readme_md'] = skill_readme_md
-          readme_path = os.path.join('references', 'README.md')
+          readme_path = 'README.md'
           if os.path.isfile(readme_path):
               try:
                   readme_post = frontmatter.load(readme_path)

README.md

@@ -86,29 +86,74 @@ send_prompt_via_composer(page, "第一行\n第二行")
 
 读取背景音乐、字体、水印和 ffmpeg 工具。
 
-如果目录不存在，会优先通过 ZIP 下载公开仓库：
+如果目录不存在，会优先下载稳定 release bundle（不依赖用户电脑安装 Git）：
 
-`https://git.jc2009.com/client-commons/media-assets/archive/main.zip`
+`https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip`
 
-这不依赖用户电脑安装 Git。
+bundle 内含背景音乐、字体与水印；**不包含** ffmpeg 二进制。ffmpeg / ffprobe 仍由 bundle 内 `manifest.json` 的 `tools.ffmpeg` 下载源单独拉取。
 
 可通过环境变量覆盖：
 
-`MEDIA_ASSETS_ROOT=D:\OpenClaw\client-commons\media-assets`
+- `MEDIA_ASSETS_ROOT=/path/to/custom/media-assets` — 本地资源目录
+- `MEDIA_ASSETS_BUNDLE_URL=https://example.com/media-assets.zip` — bundle 下载地址
+
+### Runtime diagnostics（统一 health 诊断）
+
+`jiangchang_skill_core.runtime_diagnostics` 提供各 Skill 共用的 Runtime / media-assets / ffmpeg 探测，避免每个技能各自复制检查逻辑。Skill 的 `health` 命令可调用：
+
+```python
+from jiangchang_skill_core import (
+    collect_runtime_diagnostics,
+    format_runtime_health_lines,
+    runtime_diagnostics_dict,
+)
+
+diag = collect_runtime_diagnostics(
+    skill_slug="my-skill",
+    platform_kit_min_version="1.0.10",  # 可选；省略则只报告当前版本
+    skill_root="/path/to/my-skill",     # 可选；用于检测 vendored jiangchang_skill_core
+)
+for line in format_runtime_health_lines(diag):
+    print(line)
+print(runtime_diagnostics_dict(diag))  # 可 JSON 序列化
+```
+
+诊断只读探测本地状态，不会触发 media-assets 下载。
+
+### 录屏合成（`screencast`）
+
+`run_screencast()` / `compose_video()` 通过 `jiangchang_skill_core.media_assets` 解析 ffmpeg 与背景音乐，**不使用系统 PATH 中的 `ffmpeg`**。
+
+- **默认**：不传 `media_assets_root` 时，使用 `{JIANGCHANG_DATA_ROOT}/shared/media-assets`（或进程环境中已有的 `MEDIA_ASSETS_ROOT`）。
+- **覆盖**：传入 `media_assets_root="/path/to/media-assets"`，或在 `compose_video(..., media_assets_env={"MEDIA_ASSETS_ROOT": "..."})` 中设置；参数 `media_assets_root` 优先于 `media_assets_env` 里已有的 `MEDIA_ASSETS_ROOT`。
+- `run_screencast(media_assets_root=...)` 会把该路径原样传给 `compose_video()`；`music_subdir` 已废弃（会触发 `DeprecationWarning`），音乐选择统一由 `pick_background_music()` 完成。
+
+```python
+from screencast import run_screencast
+
+run_screencast(
+    skill_slug="demo-skill",
+    subtitle_script=[("PASSED", "测试通过")],
+    pytest_args=["-q", "tests/test_demo.py"],
+    output_dir="./out",
+    media_assets_root="/path/to/media-assets",  # 可选
+)
+```
 
 ## 5. 发版流程（维护者）
 
-### main 分支自动预发布
+### 发布规则
 
-每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **post 预发布版本**（仅供 CI 流水线内部使用，**不是**正式版本线）。
-
-正式版本线使用 **1.0.x** semver，通过 git tag 发布。
+- jiangchang-platform-kit **只发布正式版本**。
+- 每次需要发布公共能力时，先 bump `pyproject.toml` 的 `version`（例如 `1.0.10` → `1.0.11`）。
+- 推送 `main` 后，`.github/workflows/publish.yml` 会构建并上传该正式版本到 Gitea PyPI。
+- 发布包版本必须等于 `pyproject.toml` 中的 `version`；若该版本已存在，上传会失败，需继续 bump 版本号后再推送。
 
 ```bash
 git push origin main
 ```
 
-测试人员升级安装（始终安装当前正式 tag 对应的版本，例如 `1.0.9`）：
+升级安装：
 
 ```bash
 python -m pip install --upgrade \
@@ -117,15 +162,6 @@ python -m pip install --upgrade \
   jiangchang-platform-kit
 ```
 
-### 正式版本（tag）
-
-打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v1.0.9` → PyPI 包 `1.0.9`）：
-
-```bash
-git tag v1.0.9
-git push origin v1.0.9
-```
-
 ### 安装后验证
 
 ```bash

pyproject.toml

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

src/jiangchang_desktop_sdk/__init__.py

@@ -23,7 +23,7 @@ __all__ = [
     "LaunchError",
     "GatewayDownError",
 ]
-__version__ = "1.0.9"
+__version__ = "1.0.13"
 
 
 def __getattr__(name: str):

src/jiangchang_skill_core/__init__.py

@@ -23,13 +23,28 @@ from .unified_logging import (
 )
 from .media_assets import (
     MediaAssetsStatus,
+    background_music_issue,
     ensure_media_assets,
+    is_git_lfs_pointer,
+    is_usable_audio_file,
     pick_background_music,
+    probe_background_music,
+    probe_ffmpeg,
+    probe_media_assets,
     resolve_ffmpeg,
     resolve_ffprobe,
     resolve_media_assets_root,
     resolve_music_dir,
 )
+from .runtime_diagnostics import (
+    RuntimeDiagnostics,
+    RuntimeIssue,
+    collect_runtime_diagnostics,
+    format_runtime_health_lines,
+    is_jiangchang_skill_core_from_skill_tree,
+    runtime_diagnostics_dict,
+    version_ge,
+)
 
 try:
     from . import rpa
@@ -43,12 +58,25 @@ __all__ = [
     "EntitlementResult",
     "EntitlementServiceError",
     "MediaAssetsStatus",
+    "RuntimeDiagnostics",
+    "RuntimeIssue",
+    "background_music_issue",
+    "collect_runtime_diagnostics",
     "ensure_media_assets",
+    "format_runtime_health_lines",
+    "is_git_lfs_pointer",
+    "is_jiangchang_skill_core_from_skill_tree",
+    "is_usable_audio_file",
+    "probe_background_music",
+    "probe_ffmpeg",
+    "probe_media_assets",
     "pick_background_music",
     "resolve_ffmpeg",
     "resolve_ffprobe",
     "resolve_media_assets_root",
     "resolve_music_dir",
+    "runtime_diagnostics_dict",
+    "version_ge",
     "apply_cli_local_defaults",
     "attach_unified_file_handler",
     "ensure_env_file",

src/jiangchang_skill_core/config.py

@@ -69,6 +69,69 @@ def reset_cache() -> None:
     _example_defaults_cache = None
 
 
+def get_env_file_path() -> str | None:
+    return _env_file_path
+
+
+def get_example_path() -> str | None:
+    return _example_path
+
+
+def _iter_env_assignments(path: str) -> list[tuple[str, str]]:
+    """按文件顺序返回 (key, 完整赋值行)。"""
+    result: list[tuple[str, str]] = []
+    if not path or not os.path.isfile(path):
+        return result
+    with open(path, encoding="utf-8") as f:
+        for raw_line in f:
+            line = raw_line.rstrip("\n\r")
+            stripped = line.strip()
+            if not stripped or stripped.startswith("#"):
+                continue
+            if "=" not in stripped:
+                continue
+            key, _, _ = stripped.partition("=")
+            key = key.strip()
+            if key:
+                result.append((key, stripped))
+    return result
+
+
+def merge_missing_env_keys(
+    example_path: str,
+    dest_path: str,
+    *,
+    comment_skill: str | None = None,
+) -> list[str]:
+    """将 example 中有而用户 .env 没有的 key 追加到 dest 末尾，不修改已有项。"""
+    if not example_path or not os.path.isfile(example_path):
+        return []
+    if not dest_path or not os.path.isfile(dest_path):
+        return []
+
+    dest_keys = set(_parse_env_file(dest_path).keys())
+    to_append: list[tuple[str, str]] = []
+    for key, assignment in _iter_env_assignments(example_path):
+        if key not in dest_keys:
+            to_append.append((key, assignment))
+
+    if not to_append:
+        return []
+
+    header = (
+        f"# Added by {comment_skill} from .env.example"
+        if comment_skill
+        else "# Added from .env.example"
+    )
+    with open(dest_path, "a", encoding="utf-8") as f:
+        f.write("\n\n" + header + "\n")
+        for _, assignment in to_append:
+            f.write(assignment + "\n")
+
+    reset_cache()
+    return [key for key, _ in to_append]
+
+
 def ensure_env_file(skill_slug: str, example_path: str) -> str:
     """首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env（已存在不覆盖），返回路径。"""
     global _skill_slug, _example_path, _env_file_path

src/jiangchang_skill_core/media_assets.py

@@ -16,17 +16,86 @@ import zipfile
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Optional
+from typing import Any, Mapping, Optional
 
-MEDIA_ASSETS_ZIP_URL = (
-    "https://git.jc2009.com/client-commons/media-assets/archive/main.zip"
+MEDIA_ASSETS_BUNDLE_URL = (
+    "https://git.jc2009.com/client-commons/media-assets/releases/download/vlatest/media-assets.zip"
 )
+MEDIA_ASSETS_ZIP_URL = MEDIA_ASSETS_BUNDLE_URL
 MEDIA_ASSETS_GIT_URL = "https://git.jc2009.com/client-commons/media-assets"
 
 _WIN_DEFAULT_DATA_ROOT = Path(r"D:\jiangchang-data")
 _NON_WIN_DEFAULT_DATA_ROOT = Path.home() / ".jiangchang"
 
 _AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac"}
+_GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1"
+_MIN_AUDIO_BYTES = 1024
+
+
+def is_git_lfs_pointer(path: str | Path) -> bool:
+    try:
+        head = Path(path).read_bytes()[:256]
+    except OSError:
+        return False
+    return head.startswith(_GIT_LFS_POINTER_PREFIX)
+
+
+_is_git_lfs_pointer = is_git_lfs_pointer
+
+
+def _enumerate_audio_like(music_dir: Path) -> list[Path]:
+    return [
+        path
+        for path in music_dir.rglob("*")
+        if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS
+    ]
+
+
+def _music_content_issue(music_dir: Optional[Path]) -> Optional[str]:
+    if music_dir is None or not music_dir.is_dir():
+        return "music_dir_missing"
+    files = _enumerate_audio_like(music_dir)
+    if not files:
+        return "background_music_dir_empty"
+    if all(_is_git_lfs_pointer(path) for path in files):
+        return "background_music_lfs_pointer_only"
+    if not any(_is_usable_audio_file(path) for path in files):
+        return "background_music_invalid"
+    return None
+
+
+def _music_is_ready(music_dir: Optional[Path]) -> bool:
+    return _music_content_issue(music_dir) is None
+
+
+def _needs_music_content_repair(status: MediaAssetsStatus) -> bool:
+    if status.music_dir is None:
+        return "music_dir_missing" in status.warnings
+    issue = _music_content_issue(status.music_dir)
+    return issue in (
+        "background_music_dir_empty",
+        "background_music_lfs_pointer_only",
+        "background_music_invalid",
+    )
+
+
+def is_usable_audio_file(path: str | Path, *, min_bytes: int = _MIN_AUDIO_BYTES) -> bool:
+    p = Path(path)
+    if not p.is_file():
+        return False
+    if p.suffix.lower() not in _AUDIO_EXTENSIONS:
+        return False
+    if is_git_lfs_pointer(p):
+        return False
+    try:
+        if p.stat().st_size < min_bytes:
+            return False
+    except OSError:
+        return False
+    return True
+
+
+_is_usable_audio_file = is_usable_audio_file
 
 
 @dataclass
@@ -39,6 +108,7 @@ class MediaAssetsStatus:
     ffmpeg_path: Optional[Path] = None
     ffprobe_path: Optional[Path] = None
     music_dir: Optional[Path] = None
+    music_ready: bool = False
     fonts_dir: Optional[Path] = None
     watermark_dir: Optional[Path] = None
 
@@ -47,6 +117,12 @@ def _env_dict(env: dict[str, str] | None) -> dict[str, str]:
     return env if env is not None else dict(os.environ)
 
 
+def _media_assets_bundle_url(env: dict[str, str] | None = None) -> str:
+    env_map = _env_dict(env)
+    override = (env_map.get("MEDIA_ASSETS_BUNDLE_URL") or "").strip()
+    return override or MEDIA_ASSETS_BUNDLE_URL
+
+
 def _resolve_jiangchang_data_root(
     data_root: str | Path | None = None,
     env: dict[str, str] | None = None,
@@ -275,6 +351,13 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
     if not ffprobe_exists:
         warnings.append("ffprobe_missing")
 
+    music_ready = False
+    if music_exists:
+        music_issue = _music_content_issue(music_dir)
+        if music_issue and music_issue != "music_dir_missing":
+            warnings.append(music_issue)
+        music_ready = music_issue is None
+
     ready = (
         exists
         and music_exists
@@ -282,6 +365,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
         and watermark_exists
         and ffmpeg_exists
         and ffprobe_exists
+        and music_ready
     )
 
     return MediaAssetsStatus(
@@ -293,6 +377,7 @@ def _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
         ffmpeg_path=ffmpeg_path if ffmpeg_exists else None,
         ffprobe_path=ffprobe_path if ffprobe_exists else None,
         music_dir=music_dir if music_exists else None,
+        music_ready=music_ready,
         fonts_dir=fonts_dir if fonts_exists else None,
         watermark_dir=watermark_dir if watermark_exists else None,
     )
@@ -403,12 +488,16 @@ def _try_download_via_zip(
     shared: Path,
     target: Path,
     warnings: list[str],
+    *,
+    env: dict[str, str] | None = None,
+    url: str | None = None,
 ) -> bool:
     download_root = _download_dir(shared)
     download_root.mkdir(parents=True, exist_ok=True)
     zip_path = download_root / "media-assets.zip"
     extracted = download_root / "extracted"
     staging = download_root / "staging"
+    bundle_url = url or _media_assets_bundle_url(env)
 
     try:
         if staging.exists():
@@ -416,7 +505,7 @@ def _try_download_via_zip(
         if extracted.exists():
             shutil.rmtree(extracted, ignore_errors=True)
 
-        _download_zip(MEDIA_ASSETS_ZIP_URL, zip_path)
+        _download_zip(bundle_url, zip_path)
         _extract_zip(zip_path, extracted)
 
         found = _find_media_assets_root_in_tree(extracted)
@@ -487,10 +576,48 @@ def _needs_media_repo_content(status: MediaAssetsStatus) -> bool:
             "music_dir_missing",
             "fonts_dir_missing",
             "watermark_dir_missing",
+            "background_music_dir_empty",
+            "background_music_lfs_pointer_only",
+            "background_music_invalid",
         )
     )
 
 
+def _maybe_repair_music_content(
+    root: Path,
+    status: MediaAssetsStatus,
+    *,
+    env: dict[str, str] | None = None,
+) -> MediaAssetsStatus:
+    if not _needs_music_content_repair(status):
+        return status
+    if _media_assets_root_overridden(env):
+        return status
+
+    warnings: list[str] = []
+    if _is_git_repo(root) and root.exists():
+        if _try_download_via_git(root, warnings, clone=False):
+            refreshed = _inspect_media_assets(root, source="git")
+            for warning in warnings:
+                if warning not in refreshed.warnings:
+                    refreshed.warnings.append(warning)
+            return _maybe_ensure_ffmpeg_tools(root, refreshed)
+
+    shared = _shared_parent(root)
+    if _try_download_via_zip(shared, root, warnings, env=env):
+        refreshed = _inspect_media_assets(root, source="zip")
+        for warning in warnings:
+            if warning not in refreshed.warnings:
+                refreshed.warnings.append(warning)
+        return _maybe_ensure_ffmpeg_tools(root, refreshed)
+
+    refreshed = _inspect_media_assets(root, source=status.source)
+    for warning in warnings:
+        if warning not in refreshed.warnings:
+            refreshed.warnings.append(warning)
+    return refreshed
+
+
 def ensure_media_assets(
     data_root: str | Path | None = None,
     env: dict[str, str] | None = None,
@@ -527,19 +654,27 @@ def ensure_media_assets(
     shared.mkdir(parents=True, exist_ok=True)
     warnings: list[str] = []
 
-    if _try_download_via_zip(shared, root, warnings):
+    if _try_download_via_zip(shared, root, warnings, env=env_map):
         return _maybe_ensure_ffmpeg_tools(
             root,
-            _inspect_media_assets(root, source="zip"),
+            _maybe_repair_music_content(
+                root,
+                _inspect_media_assets(root, source="zip"),
+                env=env_map,
+            ),
         )
 
     clone_needed = not root.exists() or not status.ready or update
     if _try_download_via_git(root, warnings, clone=clone_needed):
         return _maybe_ensure_ffmpeg_tools(
             root,
-            _inspect_media_assets(
+            _maybe_repair_music_content(
                 root,
-                source="git" if _is_git_repo(root) else "local",
+                _inspect_media_assets(
+                    root,
+                    source="git" if _is_git_repo(root) else "local",
+                ),
+                env=env_map,
             ),
         )
 
@@ -547,7 +682,29 @@ def ensure_media_assets(
     for warning in warnings:
         if warning not in final.warnings:
             final.warnings.append(warning)
-    return _maybe_ensure_ffmpeg_tools(root, final)
+    repaired = _maybe_repair_music_content(root, final, env=env_map)
+    for warning in warnings:
+        if warning not in repaired.warnings:
+            repaired.warnings.append(warning)
+    return _maybe_ensure_ffmpeg_tools(root, repaired)
+
+
+def probe_media_assets(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> MediaAssetsStatus:
+    """只检查本地 media-assets 目录，不触发下载。"""
+    env_map = _env_dict(env)
+    root = resolve_media_assets_root(data_root, env_map)
+    return _inspect_media_assets(root, source="local")
+
+
+def probe_ffmpeg(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Optional[Path]:
+    status = probe_media_assets(data_root, env)
+    return status.ffmpeg_path
 
 
 def resolve_ffmpeg(
@@ -574,6 +731,58 @@ def resolve_music_dir(
     return status.music_dir
 
 
+def probe_background_music(
+    root: str | Path | None = None,
+    *,
+    env: Mapping[str, str] | None = None,
+) -> dict[str, Any]:
+    """只读探测背景音乐目录，不触发 media-assets 下载。"""
+    env_map = _env_dict(dict(env) if env is not None else None)
+    if root is not None:
+        music_dir = Path(root) / "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()
+        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
+
+    music_root = str(music_dir) if music_dir is not None else None
+    if music_dir is None or not music_dir.is_dir():
+        return {
+            "music_root": music_root,
+            "mp3_count": 0,
+            "usable_count": 0,
+            "sample_path": None,
+            "issue": "music_dir_missing",
+        }
+
+    mp3_files = [path for path in music_dir.rglob("*.mp3") if path.is_file()]
+    usable = [path for path in music_dir.rglob("*") if is_usable_audio_file(path)]
+    issue = _music_content_issue(music_dir)
+    sample = str(sorted(usable, key=lambda p: str(p).lower())[0]) if usable else None
+    return {
+        "music_root": music_root,
+        "mp3_count": len(mp3_files),
+        "usable_count": len(usable),
+        "sample_path": sample,
+        "issue": issue,
+    }
+
+
+def background_music_issue(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Optional[str]:
+    """若 music 目录存在但无可用音频，返回标准 warning code。"""
+    if pick_background_music(data_root, env) is not None:
+        return None
+
+    status = ensure_media_assets(data_root, env)
+    return _music_content_issue(status.music_dir)
+
+
 def pick_background_music(
     data_root: str | Path | None = None,
     env: dict[str, str] | None = None,
@@ -583,10 +792,11 @@ def pick_background_music(
     if music_dir is None or not music_dir.is_dir():
         return None
 
-    candidates: list[Path] = []
-    for path in music_dir.rglob("*"):
-        if path.is_file() and path.suffix.lower() in _AUDIO_EXTENSIONS:
-            candidates.append(path)
+    candidates = [
+        path
+        for path in music_dir.rglob("*")
+        if _is_usable_audio_file(path)
+    ]
 
     if not candidates:
         return None

src/jiangchang_skill_core/rpa/video_session.py

@@ -1,11 +1,14 @@
-"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
+"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 旁白 + 背景音乐 → 最终 MP4。
+
+背景音乐默认循环铺满录屏时长并在结尾淡出；旁白使用 Windows 本地 SAPI 合成，
+失败时静默降级，不影响视频生成。Skill 无需额外配置。
+"""
 
 from __future__ import annotations
 
 import asyncio
 import os
-import random
-import shutil
+import re
 import subprocess
 import sys
 import time
@@ -15,6 +18,13 @@ from pathlib import Path
 from typing import Any, Dict, IO, List, Optional
 
 from .. import config
+from ..media_assets import (
+    background_music_issue,
+    ensure_media_assets,
+    pick_background_music,
+    resolve_ffmpeg,
+    resolve_ffprobe,
+)
 
 _JIANGCHANG_SUBTITLE_STYLE = (
     "FontName=Arial,FontSize=14,"
@@ -22,26 +32,140 @@ _JIANGCHANG_SUBTITLE_STYLE = (
     "Outline=2,Shadow=1,Alignment=2"
 )
 
-_DEFAULT_MUSIC_ROOT = os.environ.get(
-    "MEDIA_ASSETS_ROOT",
-    r"D:\OpenClaw\client-commons\media-assets",
+_CAPTURE_FRAMERATE = 15
+
+# 背景音乐循环 + 结尾淡出（默认策略，非 env 配置）
+_MUSIC_VOLUME = 0.12
+_VOICE_VOLUME = 1.0
+_MUSIC_FADE_OUT_SECONDS = 3.0
+_MIN_MUSIC_FADE_OUT_SECONDS = 1.0
+_SHORT_VIDEO_FADE_RATIO = 0.25
+
+_VOICEOVER_MAX_CHARS = 60
+_VOICEOVER_DEDUP_SECONDS = 10.0
+_VOICEOVER_MIN_GAP_SECONDS = 2.0
+
+_INTRO_BUFFER_SECONDS = 5.0
+_INTRO_STABILIZE_SECONDS = 1.0
+_OUTRO_BUFFER_SECONDS = 5.0
+
+_VOICEOVER_SKIP_SUBSTRINGS = (
+    "等待 1-5s",
+    "跳过重复",
+    "写入联系人库",
+    "探针模式",
+    "异常",
+    "失败",
 )
 
-_CAPTURE_FRAMERATE = 15
+_VOICEOVER_IMPORTANT_SUBSTRINGS = (
+    "开始",
+    "打开",
+    "启动",
+    "定位",
+    "输入",
+    "点击",
+    "搜索",
+    "等待搜索结果",
+    "检查登录状态",
+    "处理验证码",
+    "等待人工验证",
+    "解析",
+    "提取",
+    "采集完成",
+    "部分完成",
+    "完成",
+)
+
+# 明显技术日志：时间戳前缀、纯数字进度等
+_TECH_LOG_RE = re.compile(
+    r"^\d{4}-\d{2}-\d{2}|^\[\w+\]|^DEBUG:|^INFO:|^WARN:|^ERROR:"
+)
 
 
 def _record_video_enabled() -> bool:
     return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
 
 
-def _resolve_music_file() -> Optional[Path]:
-    music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music"
-    if not music_dir.is_dir():
+def _resolve_ffmpeg_exe() -> Optional[Path]:
+    ffmpeg = resolve_ffmpeg()
+    if ffmpeg is None:
         return None
-    files = list(music_dir.rglob("*.mp3"))
-    if not files:
+    path = Path(ffmpeg)
+    return path if path.is_file() else None
+
+
+def _resolve_ffprobe_exe() -> Optional[Path]:
+    """定位 ffprobe；优先 media_assets bundle，否则尝试 ffmpeg 同目录。"""
+    ffprobe = resolve_ffprobe()
+    if ffprobe is not None:
+        path = Path(ffprobe)
+        if path.is_file():
+            return path
+    ffmpeg = _resolve_ffmpeg_exe()
+    if ffmpeg is not None:
+        sibling = ffmpeg.parent / ("ffprobe.exe" if sys.platform == "win32" else "ffprobe")
+        if sibling.is_file():
+            return sibling
+    return None
+
+
+def _probe_media_duration_seconds(
+    ffprobe_exe: Optional[Path], media_path: Path
+) -> Optional[float]:
+    if ffprobe_exe is None or not media_path.is_file():
         return None
-    return random.choice(files)
+    try:
+        result = subprocess.run(
+            [
+                str(ffprobe_exe),
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=nk=1:nw=1",
+                str(media_path),
+            ],
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=30,
+        )
+        if result.returncode != 0:
+            return None
+        raw = (result.stdout or "").strip()
+        if not raw:
+            return None
+        value = float(raw)
+        return value if value > 0 else None
+    except Exception:
+        return None
+
+
+def _music_fade_params(
+    duration_seconds: Optional[float],
+) -> tuple[Optional[float], Optional[float]]:
+    if duration_seconds is None or duration_seconds <= 0:
+        return (None, None)
+    fade = min(
+        _MUSIC_FADE_OUT_SECONDS,
+        max(_MIN_MUSIC_FADE_OUT_SECONDS, duration_seconds * _SHORT_VIDEO_FADE_RATIO),
+    )
+    fade_start = max(0.0, duration_seconds - fade)
+    return (fade_start, fade)
+
+
+def _resolve_music_file(warnings: List[str]) -> Optional[Path]:
+    music = pick_background_music()
+    if music is not None:
+        return music
+
+    issue = background_music_issue()
+    if issue:
+        warnings.append(issue)
+    return None
 
 
 @dataclass
@@ -51,6 +175,183 @@ class _StepEntry:
     duration: float = 4.0
 
 
+@dataclass
+class _VoiceClip:
+    path: Path
+    delay_ms: int
+
+
+def _voiceover_text_for_step(text: str) -> Optional[str]:
+    """过滤适合朗读的步骤文本；噪音/技术日志返回 None。"""
+    cleaned = (text or "").strip()
+    if not cleaned:
+        return None
+    if _TECH_LOG_RE.search(cleaned):
+        return None
+    for skip in _VOICEOVER_SKIP_SUBSTRINGS:
+        if skip in cleaned:
+            return None
+    if len(cleaned) > _VOICEOVER_MAX_CHARS:
+        cleaned = cleaned[:_VOICEOVER_MAX_CHARS]
+    return cleaned
+
+
+def _is_important_voiceover(text: str) -> bool:
+    for important in _VOICEOVER_IMPORTANT_SUBSTRINGS:
+        if important in text:
+            return True
+    return False
+
+
+def _select_voiceover_clips(
+    entries: List[_StepEntry],
+) -> List[tuple[_StepEntry, str]]:
+    """按时间排序，去重并保证旁白间隔；关键动作可绕过最小间隔。"""
+    result: List[tuple[_StepEntry, str]] = []
+    last_text_at: Dict[str, float] = {}
+    last_spoken_at = -_VOICEOVER_MIN_GAP_SECONDS
+
+    for entry in sorted(entries, key=lambda e: e.start):
+        spoken = _voiceover_text_for_step(entry.text)
+        if not spoken:
+            continue
+        prev = last_text_at.get(spoken)
+        if prev is not None and entry.start - prev < _VOICEOVER_DEDUP_SECONDS:
+            continue
+        if (
+            not _is_important_voiceover(spoken)
+            and entry.start - last_spoken_at < _VOICEOVER_MIN_GAP_SECONDS
+        ):
+            continue
+        last_text_at[spoken] = entry.start
+        last_spoken_at = entry.start
+        result.append((entry, spoken))
+    return result
+
+
+def _tts_available() -> bool:
+    return sys.platform == "win32"
+
+
+def _sapi_synthesize_wav(text: str, output_wav: Path) -> bool:
+    if not _tts_available():
+        return False
+    output_wav.parent.mkdir(parents=True, exist_ok=True)
+    out_str = str(output_wav.resolve()).replace("'", "''")
+    text_esc = text.replace("'", "''")
+    ps_script = (
+        "Add-Type -AssemblyName System.Speech; "
+        "$s = New-Object System.Speech.Synthesis.SpeechSynthesizer; "
+        f"$s.SetOutputToWaveFile('{out_str}'); "
+        f"$s.Speak('{text_esc}'); "
+        "$s.Dispose()"
+    )
+    try:
+        result = subprocess.run(
+            ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_script],
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=120,
+        )
+        return (
+            result.returncode == 0
+            and output_wav.is_file()
+            and output_wav.stat().st_size > 0
+        )
+    except Exception:
+        return False
+
+
+def _mix_voice_clips(
+    ffmpeg_exe: Path, clips: List[_VoiceClip], output_wav: Path
+) -> bool:
+    if not clips:
+        return False
+    output_wav.parent.mkdir(parents=True, exist_ok=True)
+
+    filter_parts: List[str] = []
+    mix_labels: List[str] = []
+    cmd_inputs: List[str] = []
+
+    for i, clip in enumerate(clips):
+        cmd_inputs.extend(["-i", str(clip.path)])
+        label = f"v{i}"
+        filter_parts.append(
+            f"[{i}:a]adelay={clip.delay_ms}|{clip.delay_ms},"
+            f"volume={_VOICE_VOLUME}[{label}]"
+        )
+        mix_labels.append(f"[{label}]")
+
+    if len(clips) == 1:
+        filter_complex = filter_parts[0].replace(f"[v0]", "[out]")
+        # single clip: relabel output
+        filter_complex = (
+            f"[0:a]adelay={clips[0].delay_ms}|{clips[0].delay_ms},"
+            f"volume={_VOICE_VOLUME}[out]"
+        )
+    else:
+        filter_complex = (
+            ";".join(filter_parts)
+            + ";"
+            + "".join(mix_labels)
+            + f"amix=inputs={len(clips)}:duration=longest:dropout_transition=0:normalize=0[out]"
+        )
+
+    cmd_tail = [
+        "-y",
+        *cmd_inputs,
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[out]",
+        str(output_wav),
+    ]
+    ok, _ = _run_ffmpeg(ffmpeg_exe, cmd_tail)
+    return ok and output_wav.is_file()
+
+
+def _synthesize_step_voiceover(
+    entries: List[_StepEntry],
+    output_wav: Path,
+    warnings: List[str],
+    *,
+    ffmpeg_exe: Optional[Path] = None,
+) -> Optional[Path]:
+    """本地 Windows SAPI 逐条合成旁白并按步骤 start 时间对齐；失败返回 None。"""
+    if not _tts_available():
+        warnings.append("tts_unsupported_platform")
+        return None
+
+    clips_to_speak = _select_voiceover_clips(entries)
+    if not clips_to_speak:
+        return None
+
+    voice_dir = output_wav.parent
+    voice_dir.mkdir(parents=True, exist_ok=True)
+    temp_clips: List[_VoiceClip] = []
+
+    for i, (entry, spoken) in enumerate(clips_to_speak):
+        clip_path = voice_dir / f"voice_{i:03d}.wav"
+        if not _sapi_synthesize_wav(spoken, clip_path):
+            warnings.append("tts_synthesis_failed")
+            return None
+        delay_ms = max(0, int(entry.start * 1000))
+        temp_clips.append(_VoiceClip(path=clip_path, delay_ms=delay_ms))
+
+    if ffmpeg_exe is None:
+        ffmpeg_exe = _resolve_ffmpeg_exe()
+    if ffmpeg_exe is None:
+        warnings.append("tts_mix_ffmpeg_missing")
+        return None
+
+    if not _mix_voice_clips(ffmpeg_exe, temp_clips, output_wav):
+        warnings.append("tts_mix_failed")
+        return None
+    return output_wav
+
+
 def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
     MIN_DURATION = 1.0
     GAP = 0.1
@@ -87,7 +388,12 @@ def _escape_srt_path(path: str) -> str:
     return path.replace("\\", "/").replace(":", "\\:")
 
 
-def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
+def _ffmpeg_cmd(ffmpeg_exe: Path, *args: str) -> List[str]:
+    return [str(ffmpeg_exe), *args]
+
+
+def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: List[str]) -> tuple[bool, str]:
+    cmd = _ffmpeg_cmd(ffmpeg_exe, *cmd_tail)
     try:
         result = subprocess.run(
             cmd,
@@ -104,36 +410,27 @@ def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
     return True, ""
 
 
-def _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]:
-    """Windows gdigrab 全桌面录制；非 Windows 返回 None。"""
-    if sys.platform != "win32":
-        return None
-    return [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "gdigrab",
-        "-framerate",
-        str(_CAPTURE_FRAMERATE),
-        "-i",
-        "desktop",
-        "-c:v",
-        "libx264",
-        "-preset",
-        "ultrafast",
-        "-pix_fmt",
-        "yuv420p",
-        capture_path,
-    ]
+def _build_music_filter(
+    audio_input_idx: int,
+    fade_start: Optional[float],
+    fade_duration: Optional[float],
+    output_label: str = "music",
+) -> str:
+    filt = f"[{audio_input_idx}:a]volume={_MUSIC_VOLUME}"
+    if fade_start is not None and fade_duration is not None:
+        filt += f",afade=t=out:st={fade_start}:d={fade_duration}"
+    return filt + f"[{output_label}]"
 
 
 def _compose_capture_mp4(
+    ffmpeg_exe: Path,
     capture_path: Path,
     srt_path: Path,
     output_path: Path,
     *,
     music_file: Optional[Path] = None,
-    music_volume: float = 0.15,
+    voiceover_file: Optional[Path] = None,
+    warnings: Optional[List[str]] = None,
 ) -> tuple[bool, str]:
     output_path.parent.mkdir(parents=True, exist_ok=True)
     srt_esc = _escape_srt_path(str(srt_path.resolve()))
@@ -142,42 +439,79 @@ def _compose_capture_mp4(
         f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
     )
 
-    if music_file and music_file.is_file():
-        filter_complex = (
-            f"[0:v]{subtitle_filter}[vout];"
-            f"[1:a]volume={music_volume},apad[aout]"
+    has_music = music_file is not None and music_file.is_file()
+    has_voice = voiceover_file is not None and voiceover_file.is_file()
+
+    fade_start: Optional[float] = None
+    fade_duration: Optional[float] = None
+    if has_music:
+        ffprobe = _resolve_ffprobe_exe()
+        duration = _probe_media_duration_seconds(ffprobe, capture_path)
+        fade_start, fade_duration = _music_fade_params(duration)
+        if fade_start is None and warnings is not None:
+            warnings.append("video_duration_probe_failed_for_music_fade")
+
+    if has_music or has_voice:
+        cmd_tail: List[str] = ["-y", "-i", str(capture_path)]
+        input_idx = 1
+        filter_parts: List[str] = [f"[0:v]{subtitle_filter}[vout]"]
+        audio_labels: List[str] = []
+
+        if has_music:
+            cmd_tail.extend(["-stream_loop", "-1", "-i", str(music_file)])
+            music_label = "aout" if not has_voice else "music"
+            filter_parts.append(
+                _build_music_filter(input_idx, fade_start, fade_duration, music_label)
+            )
+            if has_voice:
+                audio_labels.append(f"[{music_label}]")
+            input_idx += 1
+
+        if has_voice:
+            cmd_tail.extend(["-i", str(voiceover_file)])
+            if has_music:
+                filter_parts.append(
+                    f"[{input_idx}:a]volume={_VOICE_VOLUME}[voice]"
+                )
+                audio_labels.append("[voice]")
+            else:
+                # voice-only：apad 补静音，避免 -shortest 按旁白长度截短视频
+                filter_parts.append(
+                    f"[{input_idx}:a]volume={_VOICE_VOLUME},apad[aout]"
+                )
+            input_idx += 1
+
+        if len(audio_labels) == 2:
+            filter_parts.append(
+                "".join(audio_labels) + "amix=inputs=2:normalize=0[aout]"
+            )
+
+        cmd_tail.extend(
+            [
+                "-filter_complex",
+                ";".join(filter_parts),
+                "-map",
+                "[vout]",
+                "-map",
+                "[aout]",
+                "-c:v",
+                "libx264",
+                "-preset",
+                "fast",
+                "-crf",
+                "23",
+                "-c:a",
+                "aac",
+                "-b:a",
+                "128k",
+                "-shortest",
+                "-pix_fmt",
+                "yuv420p",
+                str(output_path),
+            ]
         )
-        cmd = [
-            "ffmpeg",
-            "-y",
-            "-i",
-            str(capture_path),
-            "-i",
-            str(music_file),
-            "-filter_complex",
-            filter_complex,
-            "-map",
-            "[vout]",
-            "-map",
-            "[aout]",
-            "-c:v",
-            "libx264",
-            "-preset",
-            "fast",
-            "-crf",
-            "23",
-            "-c:a",
-            "aac",
-            "-b:a",
-            "128k",
-            "-shortest",
-            "-pix_fmt",
-            "yuv420p",
-            str(output_path),
-        ]
     else:
-        cmd = [
-            "ffmpeg",
+        cmd_tail = [
             "-y",
             "-i",
             str(capture_path),
@@ -197,7 +531,30 @@ def _compose_capture_mp4(
             "yuv420p",
             str(output_path),
         ]
-    return _run_ffmpeg(cmd)
+    return _run_ffmpeg(ffmpeg_exe, cmd_tail)
+
+
+def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]:
+    """Windows gdigrab 全桌面录制；非 Windows 返回 None。"""
+    if sys.platform != "win32":
+        return None
+    return _ffmpeg_cmd(
+        ffmpeg_exe,
+        "-y",
+        "-f",
+        "gdigrab",
+        "-framerate",
+        str(_CAPTURE_FRAMERATE),
+        "-i",
+        "desktop",
+        "-c:v",
+        "libx264",
+        "-preset",
+        "ultrafast",
+        "-pix_fmt",
+        "yuv420p",
+        capture_path,
+    )
 
 
 class RpaVideoSession:
@@ -210,11 +567,13 @@ class RpaVideoSession:
         skill_data_dir: str,
         batch_id: str,
         title: str = "",
+        closing_title: str = "",
     ) -> None:
         self.skill_slug = skill_slug
         self.skill_data_dir = os.path.abspath(skill_data_dir)
         self.batch_id = batch_id
         self.title = title
+        self.closing_title = closing_title
         self.enabled = _record_video_enabled()
         self.warnings: List[str] = []
         self.output_video_path: Optional[str] = None
@@ -230,12 +589,17 @@ class RpaVideoSession:
             self.skill_data_dir, "rpa-artifacts", batch_id
         )
         self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
+        self._voiceover_dir = os.path.join(self._artifact_root, "voiceover")
         self._logs_dir = os.path.join(self._artifact_root, "logs")
         self._ffmpeg_record_log_path = os.path.join(self._logs_dir, "ffmpeg-record.log")
         self._capture_path = os.path.join(self._artifact_root, "capture.mp4")
         self._srt_path = os.path.join(self._subtitles_dir, f"{self._video_stem}.srt")
+        self._voiceover_path = os.path.join(self._voiceover_dir, "voiceover.wav")
         self._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
 
+        self._music_path: Optional[str] = None
+        self._voiceover_output: Optional[str] = None
+
         self._steps: List[_StepEntry] = []
         self._t0: Optional[float] = None
         self._ffmpeg_proc: Optional[subprocess.Popen] = None
@@ -244,6 +608,7 @@ class RpaVideoSession:
         if self.enabled:
             os.makedirs(self._videos_dir, exist_ok=True)
             os.makedirs(self._subtitles_dir, exist_ok=True)
+            os.makedirs(self._voiceover_dir, exist_ok=True)
             os.makedirs(self._logs_dir, exist_ok=True)
             os.makedirs(self._artifact_root, exist_ok=True)
 
@@ -287,11 +652,13 @@ class RpaVideoSession:
     def _start_ffmpeg_capture(self) -> None:
         if not self.enabled:
             return
-        if not shutil.which("ffmpeg"):
+
+        ffmpeg_exe = _resolve_ffmpeg_exe()
+        if ffmpeg_exe is None:
             self.warnings.append("ffmpeg_not_found")
             return
 
-        cmd = _build_desktop_record_cmd(self._capture_path)
+        cmd = _build_desktop_record_cmd(ffmpeg_exe, self._capture_path)
         if cmd is None:
             self._append_warning(
                 "ffmpeg_record_start_failed",
@@ -352,19 +719,80 @@ class RpaVideoSession:
         self._close_ffmpeg_record_log()
         time.sleep(0.3)
 
+    async def _safe_add_step(self, text: str, *, duration: float = 4.0) -> None:
+        try:
+            self.add_step(text, duration=duration)
+        except Exception:
+            pass
+
     async def __aenter__(self) -> "RpaVideoSession":
         if self.enabled:
             self._t0 = time.monotonic()
-            if self.title:
-                self.add_step(self.title, duration=3.0)
             self._start_ffmpeg_capture()
+            await asyncio.sleep(_INTRO_STABILIZE_SECONDS)
+            if self.title:
+                await self._safe_add_step(self.title, duration=4.0)
+            intro_wait = _INTRO_BUFFER_SECONDS - _INTRO_STABILIZE_SECONDS
+            if intro_wait > 0:
+                await asyncio.sleep(intro_wait)
         return self
 
     async def __aexit__(self, exc_type, exc, tb) -> None:
         if self.enabled:
+            if self.closing_title:
+                await self._safe_add_step(self.closing_title, duration=4.0)
+            await asyncio.sleep(_OUTRO_BUFFER_SECONDS)
             self._stop_ffmpeg_capture()
             await self.finalize()
 
+    def _try_compose_variants(
+        self,
+        ffmpeg_exe: Path,
+        capture: Path,
+        srt_file: Path,
+        output: Path,
+        *,
+        music: Optional[Path],
+        voiceover: Optional[Path],
+        use_srt: bool,
+    ) -> tuple[bool, str]:
+        """按 music/voice 组合尝试合成，失败时逐级降级。"""
+        srt = srt_file if use_srt else Path(self._srt_path)
+        variants: List[tuple[Optional[Path], Optional[Path]]] = [
+            (music, voiceover),
+        ]
+        if music and voiceover:
+            variants.append((None, voiceover))
+        if music:
+            variants.append((music, None))
+        variants.append((None, None))
+
+        seen: set[tuple[Optional[str], Optional[str]]] = set()
+        last_err = ""
+        for m, v in variants:
+            key = (str(m) if m else None, str(v) if v else None)
+            if key in seen:
+                continue
+            seen.add(key)
+            ok, err = _compose_capture_mp4(
+                ffmpeg_exe,
+                capture,
+                srt,
+                output,
+                music_file=m,
+                voiceover_file=v,
+                warnings=self.warnings,
+            )
+            if ok:
+                return True, ""
+            last_err = err
+            if m is not None and v is not None:
+                self.warnings.append(f"ffmpeg_with_music_and_voice_failed: {err}")
+            elif m is not None:
+                self.warnings.append(f"ffmpeg_with_music_failed: {err}")
+
+        return False, last_err
+
     async def finalize(self) -> None:
         """停止录制后合成最终 MP4；失败只记 warning，不抛异常。"""
         if not self.enabled:
@@ -391,31 +819,51 @@ class RpaVideoSession:
             except Exception as exc:
                 self.warnings.append(f"subtitle_write_failed: {exc}")
 
-            music = _resolve_music_file()
-            if music is None:
+            music = _resolve_music_file(self.warnings)
+            if music is not None:
+                self._music_path = str(music.resolve())
+            elif (
+                "background_music_lfs_pointer_only" not in self.warnings
+                and "background_music_invalid" not in self.warnings
+            ):
                 self.warnings.append("background_music_missing")
 
+            ffmpeg_exe = _resolve_ffmpeg_exe()
+            if ffmpeg_exe is None:
+                self.warnings.append("ffmpeg_not_found")
+                self.artifacts = self.summary()
+                return
+
+            voiceover: Optional[Path] = None
+            try:
+                vo = _synthesize_step_voiceover(
+                    entries,
+                    Path(self._voiceover_path),
+                    self.warnings,
+                    ffmpeg_exe=ffmpeg_exe,
+                )
+                if vo is not None:
+                    voiceover = vo
+                    self._voiceover_output = str(vo.resolve())
+            except Exception as exc:
+                self.warnings.append(f"tts_unexpected_error: {exc}")
+
             srt_file = Path(self._srt_path)
             use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
 
-            ok, err = _compose_capture_mp4(
+            ok, err = self._try_compose_variants(
+                ffmpeg_exe,
                 capture,
-                srt_file if use_srt else Path(self._srt_path),
+                srt_file,
                 Path(self._planned_output),
-                music_file=music,
+                music=music,
+                voiceover=voiceover,
+                use_srt=use_srt,
             )
-            if not ok and music is not None:
-                self.warnings.append(f"ffmpeg_with_music_failed: {err}")
-                ok, err = _compose_capture_mp4(
-                    capture,
-                    srt_file if use_srt else Path(self._srt_path),
-                    Path(self._planned_output),
-                    music_file=None,
-                )
             if not ok and not use_srt:
                 ok, err = _run_ffmpeg(
+                    ffmpeg_exe,
                     [
-                        "ffmpeg",
                         "-y",
                         "-i",
                         str(capture),
@@ -428,7 +876,7 @@ class RpaVideoSession:
                         "-pix_fmt",
                         "yuv420p",
                         str(self._planned_output),
-                    ]
+                    ],
                 )
             if not ok:
                 self._append_warning("ffmpeg_compose_failed", err)
@@ -447,12 +895,28 @@ class RpaVideoSession:
             if self.enabled and self._ffmpeg_record_log_path
             else None
         )
+        audio_warnings = [
+            w
+            for w in self.warnings
+            if w.startswith(
+                (
+                    "tts_",
+                    "video_duration_probe_failed",
+                    "ffmpeg_with_music",
+                    "background_music_",
+                )
+            )
+            or w == "background_music_missing"
+        ]
         return {
             "enabled": self.enabled,
             "path": self.output_video_path,
             "capture_path": self.capture_path,
             "record_log_path": record_log,
             "warnings": list(self.warnings),
+            "voiceover_path": self._voiceover_output,
+            "music_path": self._music_path,
+            "audio_warnings": audio_warnings,
         }
 
     def to_artifact_dict(self) -> Dict[str, Any]:

src/jiangchang_skill_core/runtime_diagnostics.py

@@ -0,0 +1,339 @@
+"""共享 Runtime 诊断：health 输出与 media-assets / ffmpeg 探测。"""
+
+from __future__ import annotations
+
+import importlib.metadata
+import logging
+import os
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping
+
+import jiangchang_skill_core
+from .media_assets import probe_background_music, probe_media_assets, resolve_media_assets_root
+from .runtime_env import platform_default_data_root
+
+_SHARED_CORE_HINT = (
+    "jiangchang_skill_core is loaded from the skill tree; expected the shared "
+    "jiangchang-platform-kit installed in the host Python runtime."
+)
+
+
+@dataclass(frozen=True)
+class RuntimeIssue:
+    code: str
+    message: str
+    severity: str = "warning"
+
+
+@dataclass(frozen=True)
+class RuntimeDiagnostics:
+    skill_slug: str
+    python_executable: str
+    platform_kit_version: str | None
+    platform_kit_min_version: str | None
+    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
+
+    media_assets_root: str | None
+    ffmpeg_available: bool
+    ffmpeg_path: str | None
+
+    background_music_mp3_count: int
+    background_music_usable_count: int
+    background_music_issue: str | None
+    background_music_sample_path: str | None
+
+    record_video_enabled: bool
+    issues: tuple[RuntimeIssue, ...]
+
+    @property
+    def has_fatal_issues(self) -> bool:
+        return any(issue.severity == "error" for issue in self.issues)
+
+    def issue_codes(self) -> list[str]:
+        return [issue.code for issue in self.issues]
+
+
+def _env_dict(env: Mapping[str, str] | None) -> dict[str, str]:
+    return dict(env) if env is not None else dict(os.environ)
+
+
+def _resolve_data_root(env: Mapping[str, str]) -> str:
+    root = (env.get("CLAW_DATA_ROOT") or env.get("JIANGCHANG_DATA_ROOT") or "").strip()
+    if root:
+        return root
+    return platform_default_data_root()
+
+
+def _parse_version(version: str) -> tuple[int, int, int, int]:
+    """Parse major.minor.patch[.postN] into (major, minor, patch, post)."""
+    text = version.strip()
+    post = 0
+    if ".post" in text:
+        base, _, post_part = text.partition(".post")
+        post_digits = ""
+        for ch in post_part:
+            if ch.isdigit():
+                post_digits += ch
+            else:
+                break
+        post = int(post_digits) if post_digits else 0
+        text = base
+
+    parts: list[int] = []
+    for piece in text.split("."):
+        digits = ""
+        for ch in piece:
+            if ch.isdigit():
+                digits += ch
+            else:
+                break
+        if digits:
+            parts.append(int(digits))
+
+    while len(parts) < 3:
+        parts.append(0)
+
+    return parts[0], parts[1], parts[2], post
+
+
+def version_ge(installed: str, required: str) -> bool:
+    """Compare semver-like versions without external deps (supports legacy .postN)."""
+    return _parse_version(installed) >= _parse_version(required)
+
+
+def _path_under(parent: str | Path, child: str | Path) -> bool:
+    try:
+        parent_path = Path(parent).resolve()
+        child_path = Path(child).resolve()
+        if sys.platform == "win32":
+            parent_norm = os.path.normcase(str(parent_path))
+            child_norm = os.path.normcase(str(child_path))
+            common = os.path.commonpath([parent_norm, child_norm])
+            return common == parent_norm
+        return os.path.commonpath([str(parent_path), str(child_path)]) == str(parent_path)
+    except (OSError, ValueError):
+        return False
+
+
+def is_jiangchang_skill_core_from_skill_tree(
+    *,
+    skill_root: str | Path,
+    core_file: str | Path | None = None,
+) -> bool:
+    """Return True when jiangchang_skill_core is loaded from inside skill_root."""
+    try:
+        root = Path(skill_root)
+        if not str(root).strip():
+            return False
+        core = Path(core_file) if core_file is not None else Path(jiangchang_skill_core.__file__)
+        return _path_under(root, core)
+    except (OSError, TypeError, ValueError):
+        return False
+
+
+def _platform_kit_version() -> str | None:
+    try:
+        return importlib.metadata.version("jiangchang-platform-kit")
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+
+def _bool_from_env(env: Mapping[str, str], key: str, default: bool = False) -> bool:
+    val = env.get(key)
+    if val is None or val == "":
+        return default
+    return str(val).strip().lower() in ("1", "true", "yes", "on")
+
+
+def collect_runtime_diagnostics(
+    *,
+    skill_slug: str,
+    platform_kit_min_version: str | None = None,
+    skill_root: str | Path | None = None,
+    record_video: bool | None = None,
+    env: Mapping[str, str] | None = None,
+) -> RuntimeDiagnostics:
+    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)
+
+    media_root_path = resolve_media_assets_root(resolved_data_root, env_map)
+    media_status = probe_media_assets(resolved_data_root, env_map)
+    ffmpeg_path_obj = media_status.ffmpeg_path
+    ffmpeg_ok = ffmpeg_path_obj is not None and ffmpeg_path_obj.is_file()
+
+    music_probe = probe_background_music(media_root_path, env=env_map)
+    mp3_count = int(music_probe["mp3_count"])
+    usable_count = int(music_probe["usable_count"])
+    music_issue = music_probe.get("issue")
+    music_sample = music_probe.get("sample_path")
+
+    if music_issue is None and not media_status.music_ready and media_status.warnings:
+        for warning in media_status.warnings:
+            if warning.startswith("background_music") or warning == "music_dir_missing":
+                music_issue = warning
+                break
+
+    platform_version = _platform_kit_version()
+    platform_ok: bool | None
+    if platform_kit_min_version is None:
+        platform_ok = None
+    elif platform_version is None:
+        platform_ok = False
+    else:
+        platform_ok = version_ge(platform_version, platform_kit_min_version)
+
+    if platform_version is None:
+        issues.append(
+            RuntimeIssue(
+                code="platform_kit_not_installed",
+                message="jiangchang-platform-kit package metadata not found",
+                severity="error",
+            )
+        )
+    elif platform_kit_min_version is not None and not platform_ok:
+        issues.append(
+            RuntimeIssue(
+                code="platform_kit_version_low",
+                message=(
+                    f"jiangchang-platform-kit version {platform_version} "
+                    f"is below required {platform_kit_min_version}"
+                ),
+                severity="error",
+            )
+        )
+
+    core_file: str | None
+    try:
+        core_file = os.path.abspath(jiangchang_skill_core.__file__)
+    except (OSError, TypeError, ValueError):
+        core_file = None
+
+    if skill_root is not None and core_file is not None:
+        if is_jiangchang_skill_core_from_skill_tree(skill_root=skill_root, core_file=core_file):
+            issues.append(
+                RuntimeIssue(
+                    code="jiangchang_skill_core_loaded_from_skill_tree",
+                    message=_SHARED_CORE_HINT,
+                    severity="warning",
+                )
+            )
+
+    if record_video is None:
+        record_video_enabled = _bool_from_env(env_map, "OPENCLAW_RECORD_VIDEO", False)
+    else:
+        record_video_enabled = record_video
+
+    if not ffmpeg_ok:
+        severity = "warning"
+        message = "ffmpeg not available"
+        if record_video_enabled:
+            message = "OPENCLAW_RECORD_VIDEO enabled but ffmpeg is not available"
+        issues.append(
+            RuntimeIssue(
+                code="ffmpeg_unavailable",
+                message=message,
+                severity=severity,
+            )
+        )
+
+    if record_video_enabled and music_issue:
+        issues.append(
+            RuntimeIssue(
+                code="background_music_unavailable",
+                message=f"background music unavailable: {music_issue}",
+                severity="warning",
+            )
+        )
+
+    return RuntimeDiagnostics(
+        skill_slug=skill_slug,
+        python_executable=sys.executable,
+        platform_kit_version=platform_version,
+        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),
+        ffmpeg_available=ffmpeg_ok,
+        ffmpeg_path=str(ffmpeg_path_obj) if ffmpeg_path_obj else None,
+        background_music_mp3_count=mp3_count,
+        background_music_usable_count=usable_count,
+        background_music_issue=str(music_issue) if music_issue else None,
+        background_music_sample_path=str(music_sample) if music_sample else None,
+        record_video_enabled=record_video_enabled,
+        issues=tuple(issues),
+    )
+
+
+def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]:
+    lines = [
+        f"skill_slug: {diagnostics.skill_slug}",
+        f"python_executable: {diagnostics.python_executable}",
+        f"platform_kit_version: {diagnostics.platform_kit_version or ''}",
+        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 ''}",
+        f"ffmpeg_available: {diagnostics.ffmpeg_available}",
+        f"ffmpeg_path: {diagnostics.ffmpeg_path or ''}",
+        f"background_music_mp3_count: {diagnostics.background_music_mp3_count}",
+        f"background_music_usable_count: {diagnostics.background_music_usable_count}",
+        f"background_music_issue: {diagnostics.background_music_issue or ''}",
+        f"background_music_sample_path: {diagnostics.background_music_sample_path or ''}",
+        f"record_video_enabled: {diagnostics.record_video_enabled}",
+    ]
+    for issue in diagnostics.issues:
+        lines.append(f"runtime_issue[{issue.severity}]: {issue.code} - {issue.message}")
+    return lines
+
+
+def runtime_diagnostics_dict(diagnostics: RuntimeDiagnostics) -> dict[str, Any]:
+    return {
+        "skill_slug": diagnostics.skill_slug,
+        "python_executable": diagnostics.python_executable,
+        "platform_kit_version": diagnostics.platform_kit_version,
+        "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,
+        "ffmpeg_available": diagnostics.ffmpeg_available,
+        "ffmpeg_path": diagnostics.ffmpeg_path,
+        "background_music_mp3_count": diagnostics.background_music_mp3_count,
+        "background_music_usable_count": diagnostics.background_music_usable_count,
+        "background_music_issue": diagnostics.background_music_issue,
+        "background_music_sample_path": diagnostics.background_music_sample_path,
+        "record_video_enabled": diagnostics.record_video_enabled,
+        "runtime_issues": [
+            {"code": issue.code, "message": issue.message, "severity": issue.severity}
+            for issue in diagnostics.issues
+        ],
+    }
+
+
+def log_runtime_diagnostics(
+    diagnostics: RuntimeDiagnostics,
+    logger: logging.Logger | None = None,
+    level: int = logging.INFO,
+) -> None:
+    log = logger or logging.getLogger("jiangchang_skill_core.runtime_diagnostics")
+    for line in format_runtime_health_lines(diagnostics):
+        log.log(level, line)

src/jiangchang_skill_core/runtime_env.py

@@ -34,14 +34,22 @@ def platform_default_data_root() -> str:
 
 
 def get_data_root() -> str:
-    env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
+    env = (
+        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 (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon"
+    return (
+        os.getenv("CLAW_USER_ID")
+        or os.getenv("JIANGCHANG_USER_ID")
+        or ""
+    ).strip() or "_anon"
 
 
 def _looks_like_skills_root(path: str) -> bool:
@@ -55,6 +63,7 @@ def _looks_like_skills_root(path: str) -> bool:
         "toutiao-publisher",
         "logistics-tracker",
         "api-key-vault",
+        "receive-order",
     ):
         if os.path.isdir(os.path.join(path, marker)):
             return True
@@ -130,10 +139,21 @@ def apply_cli_local_defaults() -> None:
         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():
+    if not (
+        os.getenv("CLAW_DATA_ROOT")
+        or os.getenv("JIANGCHANG_DATA_ROOT")
+        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/screencast/composer.py

@@ -1,16 +1,18 @@
 """FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
 from __future__ import annotations
 
-import random
 import subprocess
 import sys
+from dataclasses import dataclass, field
 from pathlib import Path
 
+from jiangchang_skill_core.media_assets import (
+    background_music_issue,
+    pick_background_music,
+    resolve_ffmpeg,
+)
 
 # 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性
-# FontSize 14：1080p 屏幕约占 1.3% 高度，演示视频留足画面呼吸感
-# Outline 2：14 号字配 Outline 3 会显胖，2 已足够在浅色背景上保持可读
-# 修改请评估对所有 skill 录屏的影响。
 _JIANGCHANG_SUBTITLE_STYLE = (
     "FontName=Arial,FontSize=14,"
     "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
@@ -18,63 +20,155 @@ _JIANGCHANG_SUBTITLE_STYLE = (
 )
 
 
+@dataclass
+class ComposeVideoResult:
+    output_path: str
+    warnings: list[str] = field(default_factory=list)
+
+
+def _escape_srt_path(path: str) -> str:
+    return path.replace("\\", "/").replace(":", "\\:")
+
+
+def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
+    cmd = [str(ffmpeg_exe), *cmd_tail]
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+        )
+    except FileNotFoundError:
+        return False, "ffmpeg not found"
+    if result.returncode != 0:
+        err = (result.stderr or result.stdout or "").strip()
+        return False, err[:500] if err else f"exit {result.returncode}"
+    return True, ""
+
+
+def _media_assets_env(
+    media_assets_root: str | Path | None,
+    media_assets_env: dict[str, str] | None,
+) -> dict[str, str] | None:
+    if media_assets_root is None and media_assets_env is None:
+        return None
+    env = dict(media_assets_env or {})
+    if media_assets_root is not None:
+        env["MEDIA_ASSETS_ROOT"] = str(media_assets_root)
+    return env
+
+
 def compose_video(
     frames_dir: str,
     fps: int,
     subtitle_path: str,
-    music_dir: str,
     output_path: str,
+    *,
     music_volume: float = 0.15,
-) -> str:
-    frames_dir = Path(frames_dir)
-    subtitle_path = Path(subtitle_path)
-    output_path = Path(output_path)
-    output_path.parent.mkdir(parents=True, exist_ok=True)
+    allow_no_music: bool = True,
+    media_assets_root: str | Path | None = None,
+    media_assets_env: dict[str, str] | None = None,
+) -> ComposeVideoResult:
+    """合成 MP4；ffmpeg 与背景音乐均走 jiangchang_skill_core.media_assets。"""
+    frames_dir_path = Path(frames_dir)
+    subtitle_path_obj = Path(subtitle_path)
+    output_path_obj = Path(output_path)
+    output_path_obj.parent.mkdir(parents=True, exist_ok=True)
+    warnings: list[str] = []
 
-    # 随机选一首背景音乐
-    music_files = list(Path(music_dir).rglob("*.mp3"))
-    if not music_files:
-        raise FileNotFoundError(f"music_dir 下没有找到 MP3 文件: {music_dir}")
-    music_file = random.choice(music_files)
-    print(f"[screencast] 背景音乐: {music_file.name}")
+    assets_env = _media_assets_env(media_assets_root, media_assets_env)
 
-    # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法）
-    srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:")
+    ffmpeg_exe = resolve_ffmpeg(env=assets_env)
+    if ffmpeg_exe is None or not Path(ffmpeg_exe).is_file():
+        raise RuntimeError("ffmpeg_not_found")
 
+    ffmpeg_path = Path(ffmpeg_exe)
+    music_file = pick_background_music(env=assets_env)
+    if music_file is None:
+        issue = background_music_issue(env=assets_env)
+        if issue:
+            warnings.append(issue)
+        else:
+            warnings.append("background_music_missing")
+        if not allow_no_music:
+            raise FileNotFoundError(warnings[-1])
+
+    srt_esc = _escape_srt_path(str(subtitle_path_obj.resolve()))
     subtitle_filter = (
-        f"subtitles='{srt_path_str}'"
+        f"subtitles='{srt_esc}'"
         f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
     )
 
-    # 视频和音频都放进 filter_complex，避免 -vf 与 -filter_complex 混用报错
-    filter_complex = (
-        f"[0:v]{subtitle_filter}[vout];"
-        f"[1:a]volume={music_volume},apad[aout]"
-    )
+    if music_file and music_file.is_file():
+        filter_complex = (
+            f"[0:v]{subtitle_filter}[vout];"
+            f"[1:a]volume={music_volume},apad[aout]"
+        )
+        cmd_tail = [
+            "-y",
+            "-framerate",
+            str(fps),
+            "-i",
+            str(frames_dir_path / "frame_%06d.png"),
+            "-i",
+            str(music_file),
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            "[vout]",
+            "-map",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-shortest",
+            "-pix_fmt",
+            "yuv420p",
+            str(output_path_obj),
+        ]
+        ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail)
+        if not ok:
+            warnings.append(f"ffmpeg_with_music_failed: {err}")
+            if not allow_no_music:
+                raise RuntimeError(f"ffmpeg_compose_failed: {err}")
+            music_file = None
+        else:
+            print(f"[screencast] 背景音乐: {music_file.name}")
+            print(f"[screencast] 输出: {output_path_obj}")
+            return ComposeVideoResult(str(output_path_obj.resolve()), warnings)
 
-    cmd = [
-        "ffmpeg", "-y",
-        "-framerate", str(fps),
-        "-i", str(frames_dir / "frame_%06d.png"),
-        "-i", str(music_file),
-        "-filter_complex", filter_complex,
-        "-map", "[vout]",
-        "-map", "[aout]",
-        "-c:v", "libx264",
-        "-preset", "fast",
-        "-crf", "23",
-        "-c:a", "aac",
-        "-b:a", "128k",
-        "-shortest",
-        "-pix_fmt", "yuv420p",
-        str(output_path),
+    cmd_tail = [
+        "-y",
+        "-framerate",
+        str(fps),
+        "-i",
+        str(frames_dir_path / "frame_%06d.png"),
+        "-vf",
+        subtitle_filter,
+        "-c:v",
+        "libx264",
+        "-preset",
+        "fast",
+        "-crf",
+        "23",
+        "-pix_fmt",
+        "yuv420p",
+        str(output_path_obj),
     ]
+    print("[screencast] 运行 FFmpeg 合成（无背景音乐）...")
+    ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail)
+    if not ok:
+        print(err, file=sys.stderr)
+        raise RuntimeError(f"ffmpeg_compose_failed: {err}")
 
-    print(f"[screencast] 运行 FFmpeg 合成...")
-    result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
-    if result.returncode != 0:
-        print(result.stderr, file=sys.stderr)
-        raise RuntimeError(f"FFmpeg 合成失败，退出码 {result.returncode}")
-
-    print(f"[screencast] 输出: {output_path}")
-    return str(output_path)
+    print(f"[screencast] 输出: {output_path_obj}")
+    return ComposeVideoResult(str(output_path_obj.resolve()), warnings)

src/screencast/runner.py

@@ -30,6 +30,8 @@ def _ensure_sdk_on_path() -> None:
 _ensure_sdk_on_path()
 from jiangchang_desktop_sdk.window_win32 import activate_and_maximize_jiangchang_window  # noqa: E402
 
+import warnings
+
 from .composer import compose_video
 from .recorder import ScreenRecorder
 from .subtitle import SubtitleEngine
@@ -54,10 +56,11 @@ def run_screencast(
         subtitle_script:  [(关键词, 字幕文案), ...] 按顺序匹配 pytest stdout
         pytest_args:      传递给 pytest 的参数列表
         output_dir:       最终 MP4 输出目录
-        media_assets_root: media-assets 仓库本地路径；
-                           不传时读 MEDIA_ASSETS_ROOT 环境变量，
-                           再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
-        music_subdir:     music_assets_root 下音乐子目录，默认 "music"
+        media_assets_root: media-assets 根目录；不传时使用
+                           ``resolve_media_assets_root()`` 默认路径
+                           （JIANGCHANG_DATA_ROOT/shared/media-assets 或 MEDIA_ASSETS_ROOT）
+        music_subdir:     **已废弃**，保留仅为兼容旧调用；背景音乐由
+                          ``media_assets.pick_background_music()`` 统一解析
         fps:               截帧帧率，默认 10
         activate_window:   录屏开始前是否自动激活+最大化匠厂窗口（默认 True；
                           Windows 走 ctypes，非 Windows 走 jiangchang:// 协议）
@@ -67,13 +70,14 @@ def run_screencast(
     Returns:
         输出 MP4 的绝对路径
     """
-    if media_assets_root is None:
-        media_assets_root = os.environ.get(
-            "MEDIA_ASSETS_ROOT",
-            r"D:\OpenClaw\client-commons\media-assets",
+    if music_subdir != "music":
+        warnings.warn(
+            "music_subdir is deprecated and ignored; "
+            "background music is resolved via media_assets.pick_background_music()",
+            DeprecationWarning,
+            stacklevel=2,
         )
 
-    music_dir = str(Path(media_assets_root) / music_subdir)
     output_dir_path = Path(output_dir)
     output_dir_path.mkdir(parents=True, exist_ok=True)
 
@@ -135,8 +139,8 @@ def run_screencast(
             frames_dir=frames_dir,
             fps=fps,
             subtitle_path=str(subtitle_file),
-            music_dir=music_dir,
             output_path=str(output_mp4),
+            media_assets_root=media_assets_root,
         )
 
     return str(output_mp4)

tests/test_config.py

@@ -81,6 +81,27 @@ class TestConfig(unittest.TestCase):
             self.assertEqual(config.get_float("BAD", 9.9), 9.9)
             self.assertEqual(config.get_int("BAD", 7), 7)
 
+    def test_merge_missing_env_keys_appends_only_new(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            example = os.path.join(tmp, ".env.example")
+            dest = os.path.join(tmp, ".env")
+            with open(example, "w", encoding="utf-8") as f:
+                f.write("EXISTING=1\nNEW_KEY=abc\n")
+            with open(dest, "w", encoding="utf-8") as f:
+                f.write("EXISTING=keep\n")
+
+            added = config.merge_missing_env_keys(
+                example,
+                dest,
+                comment_skill="demo-skill",
+            )
+            self.assertEqual(added, ["NEW_KEY"])
+            with open(dest, encoding="utf-8") as f:
+                content = f.read()
+            self.assertIn("EXISTING=keep", content)
+            self.assertIn("NEW_KEY=abc", content)
+            self.assertIn("Added by demo-skill", content)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_media_assets.py

@@ -15,6 +15,7 @@ from jiangchang_skill_core import media_assets as ma
 
 def _make_complete_assets(root: Path, *, with_ffmpeg: bool = True, with_ffprobe: bool = True) -> None:
     (root / "music" / "calm").mkdir(parents=True)
+    (root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 1024)
     (root / "fonts").mkdir(parents=True)
     (root / "watermark").mkdir(parents=True)
     bin_key = ma._platform_bin_key()
@@ -57,6 +58,56 @@ def _install_tools_from_manifest(root: Path, warnings: list[str]) -> bool:
     return True
 
 
+def _build_archive_zip(dest: Path, inner_name: str) -> None:
+    manifest = _sample_manifest()
+    with zipfile.ZipFile(dest, "w") as zf:
+        zf.writestr(f"{inner_name}/README.md", "# media-assets\n")
+        zf.writestr(f"{inner_name}/manifest.json", json.dumps(manifest, ensure_ascii=False))
+        zf.writestr(f"{inner_name}/music/calm/track.mp3", b"x" * 1024)
+        zf.writestr(f"{inner_name}/fonts/.keep", "")
+        zf.writestr(f"{inner_name}/watermark/.keep", "")
+
+
+def test_default_bundle_url_is_release() -> None:
+    url = ma._media_assets_bundle_url()
+    assert url == ma.MEDIA_ASSETS_BUNDLE_URL
+    assert url.endswith("/releases/download/vlatest/media-assets.zip")
+    assert "/archive/main.zip" not in url
+
+
+def test_bundle_url_env_override() -> None:
+    override = "https://example.com/media-assets.zip"
+    assert ma._media_assets_bundle_url(env={"MEDIA_ASSETS_BUNDLE_URL": override}) == override
+
+
+def test_zip_download_uses_bundle_url_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    data_root = tmp_path / "data"
+    shared = data_root / "shared"
+    target = shared / "media-assets"
+    archive = tmp_path / "media-assets.zip"
+    override = "https://example.com/media-assets.zip"
+
+    _build_archive_zip(archive, "media-assets")
+    captured_urls: list[str] = []
+
+    def _fake_download(url: str, dest: Path) -> None:
+        captured_urls.append(url)
+        dest.write_bytes(archive.read_bytes())
+
+    monkeypatch.setattr(ma, "_download_zip", _fake_download)
+    monkeypatch.setattr(ma, "_find_git_executable", lambda: None)
+    monkeypatch.setattr(ma, "_try_download_ffmpeg_tools", _install_tools_from_manifest)
+
+    env = {
+        "JIANGCHANG_DATA_ROOT": str(data_root),
+        "MEDIA_ASSETS_BUNDLE_URL": override,
+    }
+    status = ma.ensure_media_assets(env=env)
+    assert captured_urls == [override]
+    assert target.is_dir()
+    assert status.ready is True
+
+
 def test_media_assets_root_env_priority(tmp_path: Path) -> None:
     custom = tmp_path / "custom-media"
     custom.mkdir()
@@ -117,9 +168,9 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.
     root = tmp_path / "data" / "shared" / "media-assets"
     _make_complete_assets(root)
     music = root / "music"
-    (music / "b-upbeat.mp3").write_bytes(b"mp3")
-    (music / "calm" / "a-calm.mp3").write_bytes(b"mp3")
-    (music / "z-last.wav").write_bytes(b"wav")
+    (music / "b-upbeat.mp3").write_bytes(b"x" * 1024)
+    (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 1024)
+    (music / "z-last.wav").write_bytes(b"x" * 1024)
 
     monkeypatch.setattr(
         ma,
@@ -131,23 +182,18 @@ def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.
     assert picked == music / "b-upbeat.mp3"
 
 
-def _build_archive_zip(dest: Path, inner_name: str) -> None:
-    manifest = _sample_manifest()
-    with zipfile.ZipFile(dest, "w") as zf:
-        zf.writestr(f"{inner_name}/README.md", "# media-assets\n")
-        zf.writestr(f"{inner_name}/manifest.json", json.dumps(manifest, ensure_ascii=False))
-        zf.writestr(f"{inner_name}/music/.keep", "")
-        zf.writestr(f"{inner_name}/fonts/.keep", "")
-        zf.writestr(f"{inner_name}/watermark/.keep", "")
-
-
-def test_zip_wrapper_layout_detected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.mark.parametrize("inner_name", ["media-assets", "media-assets-main"])
+def test_zip_wrapper_layout_detected(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    inner_name: str,
+) -> None:
     data_root = tmp_path / "data"
     shared = data_root / "shared"
     target = shared / "media-assets"
     archive = tmp_path / "media-assets.zip"
 
-    _build_archive_zip(archive, "media-assets-main")
+    _build_archive_zip(archive, inner_name)
 
     def _fake_download(url: str, dest: Path) -> None:
         dest.write_bytes(archive.read_bytes())
@@ -213,9 +259,138 @@ def test_windows_default_data_root() -> None:
     assert root == Path(r"D:\jiangchang-data") / "shared" / "media-assets"
 
 
+def test_probe_media_assets_does_not_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    def _fail_download(*args, **kwargs):
+        raise AssertionError("download should not be called")
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", _fail_download)
+    monkeypatch.setattr(ma, "_try_download_via_git", _fail_download)
+
+    status = ma.probe_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.exists is False
+    assert status.ffmpeg_path is None
+
+
+def test_lfs_pointer_warning_in_probe_and_ensure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root)
+    music = root / "music"
+    (music / "calm" / "track.mp3").unlink()
+    (music / "lfs-only.mp3").write_text(
+        "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
+    )
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
+
+    probed = ma.probe_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert "background_music_lfs_pointer_only" in probed.warnings
+    assert probed.music_ready is False
+
+    ensured = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert "background_music_lfs_pointer_only" in ensured.warnings
+    assert ensured.ready is False
+
+
+def test_pick_background_music_skips_lfs_pointer(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root, with_ffmpeg=True, with_ffprobe=True)
+    music = root / "music"
+    (music / "calm" / "track.mp3").unlink()
+    (music / "real.mp3").write_bytes(b"x" * 1024)
+    lfs = music / "lfs-only.mp3"
+    lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n")
+
+    monkeypatch.setattr(
+        ma,
+        "ensure_media_assets",
+        lambda *a, **k: ma._inspect_media_assets(root, source="local"),
+    )
+
+    picked = ma.pick_background_music(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert picked == music / "real.mp3"
+
+
+def test_pick_background_music_no_valid_files(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root)
+    music = root / "music"
+    (music / "calm" / "track.mp3").unlink()
+    (music / "pointer.mp3").write_text(
+        "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
+    )
+
+    monkeypatch.setattr(
+        ma,
+        "ensure_media_assets",
+        lambda *a, **k: ma._inspect_media_assets(root, source="local"),
+    )
+
+    picked = ma.pick_background_music(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert picked is None
+
+
+def test_video_finalize_without_music_fallback(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    import asyncio
+    import os
+    import tempfile
+    from pathlib import Path
+
+    from jiangchang_skill_core import config
+    from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+
+    ffmpeg_exe = tmp_path / "ffmpeg.exe"
+    ffmpeg_exe.write_bytes(b"ffmpeg")
+
+    def _fake_resolve_ffmpeg(*args, **kwargs):
+        return ffmpeg_exe
+
+    def _fake_pick_music(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(
+        "jiangchang_skill_core.rpa.video_session.resolve_ffmpeg",
+        _fake_resolve_ffmpeg,
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.rpa.video_session.pick_background_music",
+        _fake_pick_music,
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.rpa.video_session.background_music_issue",
+        lambda *a, **k: None,
+    )
+
+    os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+    try:
+        config.reset_cache()
+        with tempfile.TemporaryDirectory() as skill_data:
+            session = RpaVideoSession(
+                skill_slug="receive-order",
+                skill_data_dir=skill_data,
+                batch_id="batch_no_music",
+            )
+            capture = Path(session._capture_path)
+            capture.parent.mkdir(parents=True, exist_ok=True)
+            capture.write_bytes(b"\x00" * 64)
+
+            monkeypatch.setattr(
+                "jiangchang_skill_core.rpa.video_session._compose_capture_mp4",
+                lambda *a, **k: (True, ""),
+            )
+
+            asyncio.run(session.finalize())
+            assert "background_music_missing" in session.warnings
+            assert session.output_video_path is not None
+    finally:
+        os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        config.reset_cache()
+
+
 def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     root = tmp_path / "data" / "shared" / "media-assets"
-    (root / "music").mkdir(parents=True)
+    (root / "music" / "calm").mkdir(parents=True)
+    (root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 1024)
     (root / "fonts").mkdir(parents=True)
     (root / "watermark").mkdir(parents=True)
     (root / "manifest.json").write_text(
@@ -231,3 +406,32 @@ def test_manifest_ffmpeg_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatc
     assert status.ready is True
     assert status.ffmpeg_path is not None
     assert status.ffprobe_path is not None
+
+
+def test_is_usable_audio_file_rejects_small_and_lfs(tmp_path: Path) -> None:
+    small = tmp_path / "small.mp3"
+    small.write_bytes(b"x" * 512)
+    assert ma.is_usable_audio_file(small) is False
+
+    ok = tmp_path / "ok.mp3"
+    ok.write_bytes(b"x" * 1024)
+    assert ma.is_usable_audio_file(ok) is True
+
+    lfs = tmp_path / "lfs.mp3"
+    lfs.write_text("version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n")
+    assert ma.is_usable_audio_file(lfs) is False
+    assert ma.is_git_lfs_pointer(lfs) is True
+
+
+def test_probe_background_music_structure(tmp_path: Path) -> None:
+    root = tmp_path / "media-assets"
+    music = root / "music" / "calm"
+    music.mkdir(parents=True)
+    (music / "track.mp3").write_bytes(b"x" * 1024)
+
+    probe = ma.probe_background_music(root)
+    assert probe["music_root"] == str(root / "music")
+    assert probe["mp3_count"] == 1
+    assert probe["usable_count"] == 1
+    assert probe["issue"] is None
+    assert probe["sample_path"] == str(music / "track.mp3")

tests/test_release_workflow.py

@@ -1,4 +1,4 @@
-"""Text-level checks that reusable-release-skill packages .env.example at package root."""
+"""Text-level checks for release-related GitHub Actions workflows."""
 
 from __future__ import annotations
 
@@ -8,17 +8,52 @@ import re
 import pytest
 
 _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-_WORKFLOW_PATH = os.path.join(
+_SKILL_WORKFLOW_PATH = os.path.join(
     _REPO_ROOT, ".github", "workflows", "reusable-release-skill.yaml"
 )
+_PUBLISH_WORKFLOW_PATH = os.path.join(
+    _REPO_ROOT, ".github", "workflows", "publish.yml"
+)
 
 
 @pytest.fixture(scope="module")
 def workflow_text() -> str:
-    with open(_WORKFLOW_PATH, encoding="utf-8") as f:
+    with open(_SKILL_WORKFLOW_PATH, encoding="utf-8") as f:
         return f.read()
 
 
+@pytest.fixture(scope="module")
+def publish_workflow_text() -> str:
+    with open(_PUBLISH_WORKFLOW_PATH, encoding="utf-8") as f:
+        return f.read()
+
+
+def test_workflow_packages_readme_at_root(workflow_text: str) -> None:
+    assert "readme_src = os.path.abspath('README.md')" in workflow_text
+    assert "readme_dst = os.path.join(PACKAGE, 'README.md')" in workflow_text
+    assert "Copied README.md" in workflow_text
+    assert (
+        "raise RuntimeError('README.md exists in skill root but was not packaged')"
+        in workflow_text
+    )
+
+
+def test_workflow_metadata_reads_root_readme(workflow_text: str) -> None:
+    assert "readme_path = 'README.md'" in workflow_text
+    assert "frontmatter.load(readme_path)" in workflow_text
+    assert "metadata['readme_md'] = body" in workflow_text
+
+
+def test_workflow_does_not_use_references_readme_for_marketplace(
+    workflow_text: str,
+) -> None:
+    assert "os.path.join('references', 'README.md')" not in workflow_text
+    assert not re.search(
+        r"frontmatter\.load\([^)]*references[^)]*README",
+        workflow_text,
+    )
+
+
 def test_workflow_packages_env_example_at_root(workflow_text: str) -> None:
     assert ".env.example" in workflow_text
     assert "Copied .env.example" in workflow_text
@@ -40,3 +75,18 @@ def test_workflow_does_not_package_dot_env(workflow_text: str) -> None:
         r"os\.path\.join\(PACKAGE,\s*['\"]\.env['\"]\)",
         workflow_text,
     )
+
+
+def test_publish_workflow_triggers_on_main_only(publish_workflow_text: str) -> None:
+    assert "branches:" in publish_workflow_text
+    assert "- main" in publish_workflow_text
+    assert "tags:" not in publish_workflow_text
+
+
+def test_publish_workflow_uses_pyproject_version_only(publish_workflow_text: str) -> None:
+    assert "ci_set_package_version" not in publish_workflow_text
+    assert "GITHUB_RUN_NUMBER" not in publish_workflow_text
+    assert "GITHUB_RUN_ID" not in publish_workflow_text
+    assert ".post" not in publish_workflow_text
+    assert "python3.12 -m build" in publish_workflow_text
+    assert "twine upload" in publish_workflow_text

tests/test_runtime_diagnostics.py

@@ -0,0 +1,347 @@
+# -*- coding: utf-8 -*-
+"""runtime_diagnostics 公共 Runtime / media-assets 诊断测试。"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+import jiangchang_skill_core
+from jiangchang_skill_core import media_assets as ma
+from jiangchang_skill_core.runtime_diagnostics import (
+    RuntimeDiagnostics,
+    RuntimeIssue,
+    collect_runtime_diagnostics,
+    format_runtime_health_lines,
+    is_jiangchang_skill_core_from_skill_tree,
+    runtime_diagnostics_dict,
+    version_ge,
+)
+
+
+def test_version_ge_post_release() -> None:
+    assert version_ge("1.0.10.post23", "1.0.10") is True
+    assert version_ge("1.0.10", "1.0.10") is True
+    assert version_ge("1.0.9.post19", "1.0.10") is False
+    assert version_ge("1.0.10.post23", "1.0.10.post21") is True
+    assert version_ge("1.0.10.post21", "1.0.10.post23") is False
+    assert version_ge("1.0.10", "1.0.10.post1") is False
+    assert version_ge("1.0.11", "1.0.10.post99") is True
+    assert version_ge("1.0.11", "1.0.10") is True
+
+
+def test_is_jiangchang_skill_core_from_skill_tree(tmp_path: Path) -> None:
+    skill_root = tmp_path / "my-skill"
+    inside = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
+    inside.parent.mkdir(parents=True)
+    inside.write_text("# vendored\n", encoding="utf-8")
+    outside = tmp_path / "site-packages" / "jiangchang_skill_core" / "__init__.py"
+    outside.parent.mkdir(parents=True)
+    outside.write_text("# shared\n", encoding="utf-8")
+
+    assert is_jiangchang_skill_core_from_skill_tree(
+        skill_root=skill_root,
+        core_file=inside,
+    ) is True
+    assert is_jiangchang_skill_core_from_skill_tree(
+        skill_root=skill_root,
+        core_file=outside,
+    ) is False
+
+
+def test_collect_runtime_diagnostics_json_serializable(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    data_root = tmp_path / "data"
+    env = {
+        "CLAW_DATA_ROOT": str(data_root),
+        "OPENCLAW_RECORD_VIDEO": "0",
+    }
+    media_root = data_root / "shared" / "media-assets"
+    (media_root / "music").mkdir(parents=True)
+
+    def _fake_probe_media_assets(data_root_arg=None, env_arg=None):
+        return ma.MediaAssetsStatus(
+            root=media_root,
+            exists=True,
+            ready=False,
+            source="local",
+            warnings=["ffmpeg_missing"],
+            ffmpeg_path=None,
+            music_dir=media_root / "music",
+            music_ready=False,
+        )
+
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
+        _fake_probe_media_assets,
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
+        lambda *a, **k: {
+            "music_root": str(media_root / "music"),
+            "mp3_count": 0,
+            "usable_count": 0,
+            "sample_path": None,
+            "issue": "background_music_dir_empty",
+        },
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
+        lambda: "1.0.10",
+    )
+
+    diag = collect_runtime_diagnostics(
+        skill_slug="my-skill",
+        platform_kit_min_version="1.0.9",
+        env=env,
+    )
+    payload = runtime_diagnostics_dict(diag)
+    json.dumps(payload)
+    assert payload["skill_slug"] == "my-skill"
+    assert payload["resolved_data_root"] == str(data_root)
+    assert payload["platform_kit_version_ok"] is True
+
+
+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")}
+    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": "music_dir_missing",
+        },
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
+        lambda: "1.0.8",
+    )
+
+    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
+    assert diag.platform_kit_version_ok is None
+    assert "platform_kit_version_low" not in diag.issue_codes()
+
+
+def test_platform_kit_missing_records_issue(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    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: None,
+    )
+
+    diag = collect_runtime_diagnostics(
+        skill_slug="my-skill",
+        platform_kit_min_version="1.0.10",
+        env=env,
+    )
+    assert "platform_kit_not_installed" in diag.issue_codes()
+
+
+def test_record_video_ffmpeg_and_music_warnings_not_fatal(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    media_root = tmp_path / "shared" / "media-assets"
+    env = {"CLAW_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"}
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
+        lambda *a, **k: ma.MediaAssetsStatus(
+            root=media_root,
+            exists=True,
+            ready=False,
+            source="local",
+            warnings=["ffmpeg_missing", "music_dir_missing"],
+            ffmpeg_path=None,
+            music_dir=None,
+        ),
+    )
+    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": "music_dir_missing",
+        },
+    )
+    monkeypatch.setattr(
+        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
+        lambda: "1.0.10",
+    )
+
+    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
+    codes = diag.issue_codes()
+    assert "ffmpeg_unavailable" in codes
+    assert "background_music_unavailable" in codes
+    assert diag.has_fatal_issues is False
+    ffmpeg_issue = next(i for i in diag.issues if i.code == "ffmpeg_unavailable")
+    assert ffmpeg_issue.severity == "warning"
+
+
+def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None:
+    skill_root = tmp_path / "my-skill"
+    fake_core = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
+    fake_core.parent.mkdir(parents=True)
+    fake_core.write_text("# vendored\n", encoding="utf-8")
+
+    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    with patch.object(jiangchang_skill_core, "__file__", str(fake_core)):
+        diag = collect_runtime_diagnostics(
+            skill_slug="my-skill",
+            skill_root=skill_root,
+            record_video=False,
+            env=env,
+        )
+    assert "jiangchang_skill_core_loaded_from_skill_tree" in diag.issue_codes()
+
+
+def test_format_runtime_health_lines_core_fields(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
+    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.10",
+    )
+
+    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
+    text = "\n".join(format_runtime_health_lines(diag))
+    for marker in (
+        "python_executable:",
+        "platform_kit_version:",
+        "resolved_data_root:",
+        "media_assets_root:",
+        "ffmpeg_available:",
+        "background_music_mp3_count:",
+    ):
+        assert marker in text
+
+
+def test_background_music_lfs_and_size_rules(tmp_path: Path) -> None:
+    music = tmp_path / "music"
+    music.mkdir()
+    (music / "tiny.mp3").write_bytes(b"x" * 512)
+    (music / "lfs.mp3").write_text(
+        "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
+    )
+    (music / "ok.mp3").write_bytes(b"x" * 1024)
+
+    probe = ma.probe_background_music(tmp_path)
+    assert probe["mp3_count"] == 3
+    assert probe["usable_count"] == 1
+    assert probe["sample_path"] == str(music / "ok.mp3")
+
+
+def test_runtime_diagnostics_frozen_dataclass() -> None:
+    diag = RuntimeDiagnostics(
+        skill_slug="x",
+        python_executable="python",
+        platform_kit_version="1.0.10",
+        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",
+        ffmpeg_available=False,
+        ffmpeg_path=None,
+        background_music_mp3_count=0,
+        background_music_usable_count=0,
+        background_music_issue=None,
+        background_music_sample_path=None,
+        record_video_enabled=False,
+        issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
+    )
+    assert diag.has_fatal_issues is False
+    assert diag.issue_codes() == ["ffmpeg_unavailable"]
+
+
+def test_format_runtime_health_lines_issue_separator_ascii() -> None:
+    diag = RuntimeDiagnostics(
+        skill_slug="x",
+        python_executable="python",
+        platform_kit_version="1.0.11",
+        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",
+        ffmpeg_available=False,
+        ffmpeg_path=None,
+        background_music_mp3_count=0,
+        background_music_usable_count=0,
+        background_music_issue=None,
+        background_music_sample_path=None,
+        record_video_enabled=False,
+        issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
+    )
+    lines = format_runtime_health_lines(diag)
+    issue_lines = [line for line in lines if line.startswith("runtime_issue[")]
+    assert len(issue_lines) == 1
+    assert " - " in issue_lines[0]
+    assert "\u2014" not in issue_lines[0]
+    assert "\u2013" not in issue_lines[0]

tests/test_screencast_composer.py

@@ -0,0 +1,252 @@
+# -*- coding: utf-8 -*-
+"""screencast composer 媒体资源集成测试。"""
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from screencast.composer import ComposeVideoResult, compose_video
+
+
+def _write_frame(frames_dir: Path) -> None:
+    frames_dir.mkdir(parents=True, exist_ok=True)
+    (frames_dir / "frame_000001.png").write_bytes(b"png")
+
+
+def _write_srt(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("1\n00:00:00,000 --> 00:00:02,000\nhello\n", encoding="utf-8")
+
+
+def test_compose_video_uses_absolute_ffmpeg(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    ffmpeg = tmp_path / "ffmpeg.exe"
+    ffmpeg.write_bytes(b"ffmpeg")
+    _write_frame(frames)
+    _write_srt(srt)
+
+    captured: list[list[str]] = []
+
+    def _fake_run(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
+        captured.append([str(ffmpeg_exe), *cmd_tail])
+        output.write_bytes(b"mp4")
+        return True, ""
+
+    with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch(
+        "screencast.composer.pick_background_music",
+        return_value=None,
+    ), patch("screencast.composer.background_music_issue", return_value="background_music_missing"), patch(
+        "screencast.composer._run_ffmpeg",
+        side_effect=_fake_run,
+    ):
+        result = compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+        )
+
+    assert isinstance(result, ComposeVideoResult)
+    assert captured
+    assert captured[0][0] == str(ffmpeg)
+    assert "background_music_missing" in result.warnings
+
+
+def test_compose_video_with_music(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    ffmpeg = tmp_path / "ffmpeg.exe"
+    music = tmp_path / "music.mp3"
+    ffmpeg.write_bytes(b"ffmpeg")
+    music.write_bytes(b"x" * 256)
+    _write_frame(frames)
+    _write_srt(srt)
+
+    with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch(
+        "screencast.composer.pick_background_music",
+        return_value=music,
+    ), patch(
+        "screencast.composer._run_ffmpeg",
+        return_value=(True, ""),
+    ):
+        result = compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+        )
+        output.write_bytes(b"mp4")
+        assert result.warnings == []
+
+
+def test_compose_video_music_failure_fallback(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    ffmpeg = tmp_path / "ffmpeg.exe"
+    music = tmp_path / "music.mp3"
+    ffmpeg.write_bytes(b"ffmpeg")
+    music.write_bytes(b"x" * 256)
+    _write_frame(frames)
+    _write_srt(srt)
+
+    calls = {"count": 0}
+
+    def _fake_run(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
+        calls["count"] += 1
+        if calls["count"] == 1:
+            return False, "music mux failed"
+        output.write_bytes(b"mp4")
+        return True, ""
+
+    with patch("screencast.composer.resolve_ffmpeg", return_value=ffmpeg), patch(
+        "screencast.composer.pick_background_music",
+        return_value=music,
+    ), patch(
+        "screencast.composer._run_ffmpeg",
+        side_effect=_fake_run,
+    ):
+        result = compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+        )
+
+    assert any("ffmpeg_with_music_failed" in w for w in result.warnings)
+    assert calls["count"] == 2
+
+
+def test_compose_video_passes_media_assets_root_to_media_apis(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    custom_root = tmp_path / "custom-media-assets"
+    custom_root.mkdir()
+    ffmpeg = custom_root / "ffmpeg.exe"
+    ffmpeg.write_bytes(b"ffmpeg")
+    _write_frame(frames)
+    _write_srt(srt)
+
+    captured_env: list[dict[str, str] | None] = []
+
+    def _capture_env(*args, **kwargs):
+        captured_env.append(kwargs.get("env"))
+        if len(captured_env) == 1:
+            return ffmpeg
+        if len(captured_env) == 2:
+            return None
+        return "background_music_missing"
+
+    with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch(
+        "screencast.composer.pick_background_music",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer.background_music_issue",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer._run_ffmpeg",
+        return_value=(True, ""),
+    ):
+        compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+            media_assets_root=str(custom_root),
+        )
+        output.write_bytes(b"mp4")
+
+    assert len(captured_env) == 3
+    for env in captured_env:
+        assert env is not None
+        assert env["MEDIA_ASSETS_ROOT"] == str(custom_root)
+
+
+def test_compose_video_default_does_not_set_media_assets_root(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    ffmpeg = tmp_path / "ffmpeg.exe"
+    ffmpeg.write_bytes(b"ffmpeg")
+    _write_frame(frames)
+    _write_srt(srt)
+
+    captured_env: list[dict[str, str] | None] = []
+
+    def _capture_env(*args, **kwargs):
+        captured_env.append(kwargs.get("env"))
+        if len(captured_env) == 1:
+            return ffmpeg
+        if len(captured_env) == 2:
+            return None
+        return "background_music_missing"
+
+    with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch(
+        "screencast.composer.pick_background_music",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer.background_music_issue",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer._run_ffmpeg",
+        return_value=(True, ""),
+    ):
+        compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+        )
+
+    assert all(env is None for env in captured_env)
+
+
+def test_compose_video_media_assets_root_overrides_env(tmp_path: Path) -> None:
+    frames = tmp_path / "frames"
+    srt = tmp_path / "out.srt"
+    output = tmp_path / "final.mp4"
+    custom_root = tmp_path / "override-root"
+    custom_root.mkdir()
+    ffmpeg = custom_root / "ffmpeg.exe"
+    ffmpeg.write_bytes(b"ffmpeg")
+    _write_frame(frames)
+    _write_srt(srt)
+
+    original_env = {"MEDIA_ASSETS_ROOT": "/old/path", "OTHER": "keep"}
+    captured_env: list[dict[str, str] | None] = []
+
+    def _capture_env(*args, **kwargs):
+        captured_env.append(kwargs.get("env"))
+        if len(captured_env) == 1:
+            return ffmpeg
+        if len(captured_env) == 2:
+            return None
+        return "background_music_missing"
+
+    with patch("screencast.composer.resolve_ffmpeg", side_effect=_capture_env), patch(
+        "screencast.composer.pick_background_music",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer.background_music_issue",
+        side_effect=_capture_env,
+    ), patch(
+        "screencast.composer._run_ffmpeg",
+        return_value=(True, ""),
+    ):
+        compose_video(
+            frames_dir=str(frames),
+            fps=10,
+            subtitle_path=str(srt),
+            output_path=str(output),
+            media_assets_root=str(custom_root),
+            media_assets_env=original_env,
+        )
+        output.write_bytes(b"mp4")
+
+    assert original_env["MEDIA_ASSETS_ROOT"] == "/old/path"
+    assert captured_env[0]["MEDIA_ASSETS_ROOT"] == str(custom_root)
+    assert captured_env[0]["OTHER"] == "keep"

tests/test_screencast_runner.py

@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+"""screencast runner 媒体资源参数传递测试。"""
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from screencast.runner import run_screencast
+
+
+def test_run_screencast_passes_media_assets_root_to_compose_video(tmp_path: Path) -> None:
+    custom_root = tmp_path / "media-assets"
+    custom_root.mkdir()
+    captured: dict = {}
+
+    mock_recorder = MagicMock()
+    mock_recorder.frame_count = 1
+
+    def _fake_compose_video(**kwargs):
+        captured.update(kwargs)
+        Path(kwargs["output_path"]).write_bytes(b"mp4")
+        from screencast.composer import ComposeVideoResult
+
+        return ComposeVideoResult(kwargs["output_path"])
+
+    with patch("screencast.runner.activate_and_maximize_jiangchang_window", return_value=True), patch(
+        "screencast.runner.ScreenRecorder",
+        return_value=mock_recorder,
+    ), patch("screencast.runner.SubtitleEngine") as subtitle_cls, patch(
+        "screencast.runner.subprocess.Popen",
+    ) as popen_cls, patch(
+        "screencast.runner.compose_video",
+        side_effect=_fake_compose_video,
+    ):
+        subtitle = subtitle_cls.return_value
+        proc = popen_cls.return_value
+        proc.stdout = iter([])
+        proc.wait.return_value = 0
+
+        out = run_screencast(
+            skill_slug="demo-skill",
+            subtitle_script=[("ok", "done")],
+            pytest_args=["-q", "tests/test_demo.py"],
+            output_dir=str(tmp_path / "out"),
+            media_assets_root=str(custom_root),
+        )
+
+    assert Path(out).is_file()
+    assert captured.get("media_assets_root") == str(custom_root)
+
+
+def test_run_screencast_music_subdir_deprecated_warning() -> None:
+    with pytest.warns(DeprecationWarning, match="music_subdir is deprecated"):
+        with patch("screencast.runner.activate_and_maximize_jiangchang_window", return_value=False), patch(
+            "screencast.runner.ScreenRecorder",
+        ) as rec_cls, patch("screencast.runner.SubtitleEngine"), patch(
+            "screencast.runner.subprocess.Popen",
+        ) as popen_cls, patch(
+            "screencast.runner.compose_video",
+        ) as compose_mock:
+            rec = rec_cls.return_value
+            rec.frame_count = 0
+            proc = popen_cls.return_value
+            proc.stdout = iter([])
+            proc.wait.return_value = 0
+            from screencast.composer import ComposeVideoResult
+
+            compose_mock.return_value = ComposeVideoResult("out.mp4")
+
+            run_screencast(
+                skill_slug="demo",
+                subtitle_script=[],
+                pytest_args=["-q"],
+                output_dir=".",
+                music_subdir="calm",
+            )

tests/test_video_session.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""RpaVideoSession：ffmpeg 桌面录屏。"""
+"""RpaVideoSession：ffmpeg 桌面录屏、背景音乐循环淡出、旁白混音。"""
 from __future__ import annotations
 
 import asyncio
@@ -7,10 +7,22 @@ import os
 import sys
 import tempfile
 import unittest
-from unittest.mock import MagicMock, patch
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from jiangchang_skill_core import config
-from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+from jiangchang_skill_core.rpa import video_session as vs
+from jiangchang_skill_core.rpa.video_session import (
+    RpaVideoSession,
+    _StepEntry,
+    _INTRO_BUFFER_SECONDS,
+    _INTRO_STABILIZE_SECONDS,
+    _OUTRO_BUFFER_SECONDS,
+    _compose_capture_mp4,
+    _music_fade_params,
+    _select_voiceover_clips,
+    _voiceover_text_for_step,
+)
 
 
 class TestRpaVideoSession(unittest.TestCase):
@@ -33,6 +45,8 @@ class TestRpaVideoSession(unittest.TestCase):
                 self.assertIsNone(summ["path"])
                 self.assertIsNone(summ["capture_path"])
                 self.assertIsNone(summ["record_log_path"])
+                self.assertIsNone(summ["voiceover_path"])
+                self.assertIsNone(summ["music_path"])
         finally:
             if old is not None:
                 os.environ["OPENCLAW_RECORD_VIDEO"] = old
@@ -60,6 +74,11 @@ class TestRpaVideoSession(unittest.TestCase):
                         "/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log"
                     )
                 )
+                self.assertTrue(
+                    session._voiceover_path.replace("\\", "/").endswith(
+                        "/rpa-artifacts/exmail_20260602_120000/voiceover/voiceover.wav"
+                    )
+                )
                 self.assertTrue(
                     session._planned_output.replace("\\", "/").startswith(
                         tmp.replace("\\", "/") + "/videos/"
@@ -70,9 +89,10 @@ class TestRpaVideoSession(unittest.TestCase):
             config.reset_cache()
 
     @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture")
-    @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value="ffmpeg")
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
     @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen")
-    def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, _which: MagicMock) -> None:
+    def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, mock_resolve: MagicMock) -> None:
+        mock_resolve.return_value = Path(r"C:\media\bin\win32-x64\ffmpeg.exe")
         proc = MagicMock()
         proc.poll.return_value = None
         proc.stdin = MagicMock()
@@ -96,7 +116,7 @@ class TestRpaVideoSession(unittest.TestCase):
 
                 self.assertTrue(mock_popen.called)
                 cmd = mock_popen.call_args[0][0]
-                self.assertIn("ffmpeg", cmd)
+                self.assertTrue(str(cmd[0]).endswith("ffmpeg.exe"))
                 self.assertIn("gdigrab", cmd)
                 self.assertIn("desktop", cmd)
                 kwargs = mock_popen.call_args[1]
@@ -105,8 +125,8 @@ class TestRpaVideoSession(unittest.TestCase):
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
             config.reset_cache()
 
-    @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value=None)
-    def test_ffmpeg_not_found_warning(self, _which: MagicMock) -> None:
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe", return_value=None)
+    def test_ffmpeg_not_found_warning(self, _resolve: MagicMock) -> None:
         os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
         try:
             with tempfile.TemporaryDirectory() as tmp:
@@ -143,5 +163,413 @@ class TestRpaVideoSession(unittest.TestCase):
             config.reset_cache()
 
 
+class TestMusicFadeParams(unittest.TestCase):
+    def test_normal_duration(self) -> None:
+        start, dur = _music_fade_params(20.0)
+        self.assertAlmostEqual(start, 17.0)
+        self.assertAlmostEqual(dur, 3.0)
+
+    def test_short_video(self) -> None:
+        start, dur = _music_fade_params(4.0)
+        self.assertAlmostEqual(dur, 1.0)
+        self.assertAlmostEqual(start, 3.0)
+
+    def test_invalid_duration(self) -> None:
+        self.assertEqual(_music_fade_params(None), (None, None))
+        self.assertEqual(_music_fade_params(0), (None, None))
+        self.assertEqual(_music_fade_params(-1), (None, None))
+
+
+class TestVoiceoverTextFilter(unittest.TestCase):
+    def test_keeps_meaningful_steps(self) -> None:
+        for text in (
+            "开始采集",
+            "获取1688账号",
+            "启动浏览器",
+            "检查登录状态",
+            "搜索关键词：螺丝",
+            "解析第 1 页店铺列表",
+            "打开店铺联系方式页",
+            "提取联系人信息",
+            "采集完成",
+        ):
+            self.assertIsNotNone(_voiceover_text_for_step(text), msg=text)
+
+    def test_skips_noise(self) -> None:
+        for text in (
+            "等待 1-5s",
+            "跳过重复店铺",
+            "写入联系人库",
+            "探针模式",
+            "操作异常",
+            "登录失败",
+            "2024-01-01 12:00:00 INFO: debug",
+            "",
+            "   ",
+        ):
+            self.assertIsNone(_voiceover_text_for_step(text), msg=text)
+
+    def test_truncates_long_text(self) -> None:
+        long_text = "开" * 80
+        result = _voiceover_text_for_step(long_text)
+        self.assertIsNotNone(result)
+        assert result is not None
+        self.assertEqual(len(result), 60)
+
+    def test_dedup_within_ten_seconds(self) -> None:
+        entries = [
+            _StepEntry(0.0, "启动浏览器"),
+            _StepEntry(5.0, "启动浏览器"),
+            _StepEntry(12.0, "启动浏览器"),
+        ]
+        clips = _select_voiceover_clips(entries)
+        texts = [t for _, t in clips]
+        self.assertEqual(texts.count("启动浏览器"), 2)
+
+    def test_min_gap_two_seconds(self) -> None:
+        entries = [
+            _StepEntry(0.0, "获取1688账号"),
+            _StepEntry(1.0, "连接CDP"),
+            _StepEntry(3.0, "检查登录状态"),
+        ]
+        clips = _select_voiceover_clips(entries)
+        self.assertEqual(len(clips), 2)
+        self.assertEqual(clips[0][1], "获取1688账号")
+        self.assertEqual(clips[1][1], "检查登录状态")
+
+    def test_important_bypasses_min_gap(self) -> None:
+        entries = [
+            _StepEntry(0.0, "定位搜索框"),
+            _StepEntry(0.5, "输入关键词：宁德电池"),
+            _StepEntry(1.0, "点击搜索"),
+            _StepEntry(1.5, "等待搜索结果"),
+        ]
+        clips = _select_voiceover_clips(entries)
+        texts = [t for _, t in clips]
+        self.assertEqual(len(clips), 4)
+        self.assertIn("输入关键词：宁德电池", texts)
+        self.assertIn("点击搜索", texts)
+
+    def test_noise_still_skipped_with_important_nearby(self) -> None:
+        entries = [
+            _StepEntry(0.0, "点击搜索"),
+            _StepEntry(0.5, "跳过重复店铺"),
+            _StepEntry(1.0, "写入联系人库"),
+        ]
+        clips = _select_voiceover_clips(entries)
+        texts = [t for _, t in clips]
+        self.assertEqual(texts, ["点击搜索"])
+
+
+class TestComposeCaptureMp4(unittest.TestCase):
+    def setUp(self) -> None:
+        self.ffmpeg = Path(r"C:\ffmpeg\ffmpeg.exe")
+        self.capture = Path(r"C:\tmp\capture.mp4")
+        self.srt = Path(r"C:\tmp\sub.srt")
+        self.output = Path(r"C:\tmp\out.mp4")
+        self.music = Path(r"C:\tmp\music.mp3")
+        self.voice = Path(r"C:\tmp\voiceover.wav")
+
+    def _run_compose(self, **kwargs) -> list[str]:
+        captured: list[str] = []
+
+        def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
+            captured.extend(cmd_tail)
+            return True, ""
+
+        with (
+            patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
+            patch.object(vs, "_resolve_ffprobe_exe", return_value=Path(r"C:\ffmpeg\ffprobe.exe")),
+            patch.object(vs, "_probe_media_duration_seconds", return_value=kwargs.pop("duration", 20.0)),
+            patch.object(Path, "is_file", return_value=True),
+        ):
+            _compose_capture_mp4(
+                self.ffmpeg,
+                self.capture,
+                self.srt,
+                self.output,
+                music_file=kwargs.get("music_file"),
+                voiceover_file=kwargs.get("voiceover_file"),
+                warnings=kwargs.get("warnings"),
+            )
+        return captured
+
+    def test_music_loop_params(self) -> None:
+        cmd = self._run_compose(music_file=self.music)
+        cmd_str = " ".join(cmd)
+        loop_idx = cmd.index("-stream_loop")
+        minus_one_idx = cmd.index("-1", loop_idx)
+        music_i_idx = cmd.index("-i", minus_one_idx)
+        self.assertEqual(cmd[music_i_idx + 1], str(self.music))
+        self.assertIn("-shortest", cmd)
+        filter_idx = cmd.index("-filter_complex")
+        filt = cmd[filter_idx + 1]
+        self.assertNotIn("apad", filt)
+
+    def test_music_fade_filter_long_video(self) -> None:
+        cmd = self._run_compose(music_file=self.music, duration=20.0)
+        filt = cmd[cmd.index("-filter_complex") + 1]
+        self.assertIn("afade=t=out:st=17", filt)
+        self.assertIn("d=3", filt)
+        self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
+
+    def test_music_fade_filter_short_video(self) -> None:
+        cmd = self._run_compose(music_file=self.music, duration=4.0)
+        filt = cmd[cmd.index("-filter_complex") + 1]
+        self.assertIn("afade=t=out:st=3", filt)
+        self.assertIn("d=1", filt)
+
+    def test_duration_probe_failed_no_afade(self) -> None:
+        warnings: list[str] = []
+        captured: list[str] = []
+
+        def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
+            captured.extend(cmd_tail)
+            return True, ""
+
+        with (
+            patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
+            patch.object(vs, "_resolve_ffprobe_exe", return_value=None),
+            patch.object(vs, "_probe_media_duration_seconds", return_value=None),
+            patch.object(Path, "is_file", return_value=True),
+        ):
+            ok, _ = _compose_capture_mp4(
+                self.ffmpeg,
+                self.capture,
+                self.srt,
+                self.output,
+                music_file=self.music,
+                warnings=warnings,
+            )
+        self.assertTrue(ok)
+        self.assertIn("video_duration_probe_failed_for_music_fade", warnings)
+        filt = captured[captured.index("-filter_complex") + 1]
+        self.assertNotIn("afade", filt)
+        self.assertIn("-stream_loop", captured)
+
+    def test_music_and_voice_amix(self) -> None:
+        cmd = self._run_compose(music_file=self.music, voiceover_file=self.voice)
+        filt = cmd[cmd.index("-filter_complex") + 1]
+        self.assertIn("amix", filt)
+        self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
+        self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
+        self.assertLess(vs._MUSIC_VOLUME, vs._VOICE_VOLUME)
+        self.assertIn("afade", filt)
+        self.assertNotIn("apad", filt)
+        self.assertIn("-stream_loop", cmd)
+        self.assertIn("-shortest", cmd)
+
+    def test_voice_only_no_amix(self) -> None:
+        cmd = self._run_compose(voiceover_file=self.voice)
+        filt = cmd[cmd.index("-filter_complex") + 1]
+        self.assertNotIn("amix", filt)
+        self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
+        self.assertIn("apad", filt)
+        self.assertIn("-shortest", cmd)
+        self.assertNotIn("-stream_loop", cmd)
+
+
+class TestVideoSessionLifecycle(unittest.TestCase):
+    def setUp(self) -> None:
+        config.reset_cache()
+
+    def tearDown(self) -> None:
+        os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        config.reset_cache()
+
+    @patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
+    @patch.object(RpaVideoSession, "_start_ffmpeg_capture")
+    @patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
+    @patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
+    def test_enter_starts_capture_before_title(
+        self,
+        mock_finalize: AsyncMock,
+        mock_stop: MagicMock,
+        mock_start: MagicMock,
+        mock_sleep: AsyncMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        call_order: list[str] = []
+        mock_start.side_effect = lambda: call_order.append("start_capture")
+
+        with tempfile.TemporaryDirectory() as tmp:
+            config.reset_cache()
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch_intro",
+                title="开始执行测试",
+            )
+            original_add = session.add_step
+
+            def _track_add(text: str, *, duration: float = 4.0) -> None:
+                call_order.append(f"add_step:{text}")
+                original_add(text, duration=duration)
+
+            with patch.object(session, "add_step", side_effect=_track_add):
+                asyncio.run(session.__aenter__())
+
+        self.assertEqual(call_order[0], "start_capture")
+        self.assertIn("add_step:开始执行测试", call_order)
+        mock_sleep.assert_awaited()
+        total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
+        self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
+
+    @patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
+    @patch.object(RpaVideoSession, "_start_ffmpeg_capture")
+    def test_enter_no_title_still_buffers(
+        self,
+        mock_start: MagicMock,
+        mock_sleep: AsyncMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        with tempfile.TemporaryDirectory() as tmp:
+            config.reset_cache()
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch_no_title",
+            )
+            with patch.object(session, "add_step") as mock_add:
+                asyncio.run(session.__aenter__())
+            mock_start.assert_called_once()
+            mock_add.assert_not_called()
+            total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
+            self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
+
+    @patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
+    @patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
+    @patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
+    def test_exit_closing_title_before_stop(
+        self,
+        mock_finalize: AsyncMock,
+        mock_stop: MagicMock,
+        mock_sleep: AsyncMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        call_order: list[str] = []
+        mock_stop.side_effect = lambda: call_order.append("stop_capture")
+
+        with tempfile.TemporaryDirectory() as tmp:
+            config.reset_cache()
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch_outro",
+                closing_title="任务已完成",
+            )
+            original_add = session.add_step
+
+            def _track_add(text: str, *, duration: float = 4.0) -> None:
+                call_order.append(f"add_step:{text}")
+                original_add(text, duration=duration)
+
+            with patch.object(session, "add_step", side_effect=_track_add):
+                asyncio.run(session.__aexit__(None, None, None))
+
+        self.assertIn("add_step:任务已完成", call_order)
+        self.assertEqual(call_order[-1], "stop_capture")
+        total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
+        self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
+
+    @patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
+    @patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
+    @patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
+    def test_exit_no_closing_title_still_outro_buffer(
+        self,
+        mock_finalize: AsyncMock,
+        mock_stop: MagicMock,
+        mock_sleep: AsyncMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        with tempfile.TemporaryDirectory() as tmp:
+            config.reset_cache()
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch_outro_empty",
+            )
+            with patch.object(session, "add_step") as mock_add:
+                asyncio.run(session.__aexit__(None, None, None))
+            mock_add.assert_not_called()
+            mock_stop.assert_called_once()
+            total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
+            self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
+
+
+class TestFinalizeDegradation(unittest.TestCase):
+    def setUp(self) -> None:
+        config.reset_cache()
+
+    def tearDown(self) -> None:
+        os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        config.reset_cache()
+
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
+    @patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
+    @patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
+    def test_tts_failure_still_composes(
+        self,
+        mock_compose: MagicMock,
+        mock_tts: MagicMock,
+        mock_music: MagicMock,
+        mock_ffmpeg: MagicMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
+        mock_music.return_value = Path(r"C:\tmp\music.mp3")
+        mock_tts.return_value = None
+        mock_compose.return_value = (True, "")
+
+        with tempfile.TemporaryDirectory() as tmp:
+            capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
+            capture.parent.mkdir(parents=True)
+            capture.write_bytes(b"fake")
+
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch",
+            )
+            session.add_step("启动浏览器")
+            asyncio.run(session.finalize())
+
+        self.assertTrue(mock_compose.called)
+        self.assertIsNotNone(session.output_video_path)
+
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
+    @patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
+    @patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
+    @patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
+    def test_compose_fallback_without_music(
+        self,
+        mock_compose: MagicMock,
+        mock_tts: MagicMock,
+        mock_music: MagicMock,
+        mock_ffmpeg: MagicMock,
+    ) -> None:
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
+        mock_music.return_value = Path(r"C:\tmp\music.mp3")
+        mock_tts.return_value = None
+        mock_compose.side_effect = [(False, "music err"), (True, "")]
+
+        with tempfile.TemporaryDirectory() as tmp:
+            capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
+            capture.parent.mkdir(parents=True)
+            capture.write_bytes(b"fake")
+
+            session = RpaVideoSession(
+                skill_slug="test",
+                skill_data_dir=tmp,
+                batch_id="batch",
+            )
+            asyncio.run(session.finalize())
+
+        self.assertEqual(mock_compose.call_count, 2)
+        self.assertTrue(any("ffmpeg_with_music_failed" in w for w in session.warnings))
+
+
 if __name__ == "__main__":
     unittest.main()

tools/ci_set_package_version.py

@@ -1,79 +0,0 @@
-"""Patch pyproject.toml and __init__.py version for CI builds (not committed)."""
-from __future__ import annotations
-
-import os
-import re
-import sys
-from pathlib import Path
-
-ROOT = Path(__file__).resolve().parent.parent
-
-
-def _read_base_version() -> str:
-    text = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
-    match = re.search(r'^version\s*=\s*"([^"]+)"', text, re.MULTILINE)
-    if not match:
-        raise SystemExit("ERROR: version not found in pyproject.toml")
-    base = match.group(1)
-    return re.sub(r"\.post\d+$", "", base)
-
-
-def resolve_version() -> str:
-    ref = os.environ.get("GITHUB_REF", "")
-    if ref.startswith("refs/tags/v"):
-        return ref.removeprefix("refs/tags/v")
-    if ref.startswith("refs/tags/"):
-        return ref.removeprefix("refs/tags/")
-
-    run_number = (
-        os.environ.get("GITHUB_RUN_NUMBER")
-        or os.environ.get("GITHUB_RUN_ID")
-        or "0"
-    )
-    base = _read_base_version()
-    return f"{base}.post{run_number}"
-
-
-def patch_version(version: str) -> None:
-    pyproject = ROOT / "pyproject.toml"
-    py_text = pyproject.read_text(encoding="utf-8")
-    py_text, n = re.subn(
-        r'^version\s*=\s*"[^"]*"',
-        f'version = "{version}"',
-        py_text,
-        count=1,
-        flags=re.MULTILINE,
-    )
-    if n != 1:
-        raise SystemExit("ERROR: failed to patch version in pyproject.toml")
-    pyproject.write_text(py_text, encoding="utf-8")
-
-    init_py = ROOT / "src" / "jiangchang_desktop_sdk" / "__init__.py"
-    init_text = init_py.read_text(encoding="utf-8")
-    init_text, n = re.subn(
-        r'^__version__\s*=\s*"[^"]*"',
-        f'__version__ = "{version}"',
-        init_text,
-        count=1,
-        flags=re.MULTILINE,
-    )
-    if n != 1:
-        raise SystemExit("ERROR: failed to patch __version__ in __init__.py")
-    init_py.write_text(init_text, encoding="utf-8")
-
-
-def main() -> None:
-    version = resolve_version()
-    print(f"Resolved package version: {version}")
-    patch_version(version)
-    if os.environ.get("GITHUB_OUTPUT"):
-        with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as fh:
-            fh.write(f"package_version={version}\n")
-
-
-if __name__ == "__main__":
-    try:
-        main()
-    except SystemExit as exc:
-        print(exc, file=sys.stderr)
-        raise

tools/release.ps1

@@ -28,6 +28,7 @@
   加密与 ZIP 内容由 CI 工作流 reusable-release-skill.yaml 的「Encrypt Source Code」步骤执行。CI 会：
   - 加密 scripts/（递归 PyArmor -r，输出到包内 scripts/，与源码目录树一致）
   - 复制 SKILL.md
+  - 若根目录存在 README.md，则复制到包根（明文，用于技能市场用户说明）
   - 复制 references/（排除 REQUIREMENTS.md）
   - 复制 assets/
   - 复制 tests/