Release v1.0.10: shared media_assets and screencast media_assets_root override

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

README.md

@@ -12,7 +12,7 @@ pip install jiangchang-platform-kit --index-url https://git.jc2009.com/api/packa
 
 ## 3. 包含的模块
 
-- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析等，供 Skill CLI / 服务进程使用。
+- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。
 - **jiangchang_desktop_sdk**：匠厂桌面应用自动化客户端（基于 CDP + Playwright 同步 API）。**本包不声明 Playwright 依赖**；请在运行环境中自行安装 Playwright，以便 `import playwright` 可用。
 
 ## 4. 快速开始
@@ -78,17 +78,57 @@ assert_skill_env_file(ctx, ["API_KEY"])
 send_prompt_via_composer(page, "第一行\n第二行")
 ```
 
+## 共享媒体资源
+
+`jiangchang_skill_core.media_assets` 提供共享媒体资源解析能力。默认会从：
+
+`{JIANGCHANG_DATA_ROOT}/shared/media-assets`
+
+读取背景音乐、字体、水印和 ffmpeg 工具。
+
+如果目录不存在，会优先通过 ZIP 下载公开仓库：
+
+`https://git.jc2009.com/client-commons/media-assets/archive/main.zip`
+
+这不依赖用户电脑安装 Git。
+
+可通过环境变量覆盖：
+
+`MEDIA_ASSETS_ROOT=/path/to/custom/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 版本（基于 `pyproject.toml` 中的基础版本，例如 `0.1.0.post42`）。版本号不会写回仓库，仅在 CI 构建前临时改写 `pyproject.toml` 与 `jiangchang_desktop_sdk.__version__`。
+每次推送到 `main` 会触发 `.github/workflows/publish.yml`，自动构建并发布 **post 预发布版本**（仅供 CI 流水线内部使用，**不是**正式版本线）。
+
+正式版本线使用 **1.0.x** semver，通过 git tag 发布。
 
 ```bash
 git push origin main
 ```
 
-测试人员升级安装：
+测试人员升级安装（始终安装当前正式 tag 对应的版本，例如 `1.0.9`）：
 
 ```bash
 python -m pip install --upgrade \
@@ -99,11 +139,11 @@ python -m pip install --upgrade \
 
 ### 正式版本（tag）
 
-打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v0.1.1` → PyPI 包 `0.1.1`）：
+打 tag 并推送后，发布 **去掉 `v` 前缀** 的正式版本号（例如 `v1.0.9` → PyPI 包 `1.0.9`）：
 
 ```bash
-git tag v0.1.1
+git tag v1.0.9
-git push origin v0.1.1
+git push origin v1.0.9
 ```
 
 ### 安装后验证

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jiangchang-platform-kit"
-version = "0.1.0"
+version = "1.0.10"
 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__ = "0.1.0"
+__version__ = "1.0.9"
 
 
 def __getattr__(name: str):

src/jiangchang_skill_core/__init__.py

@@ -21,6 +21,18 @@ from .unified_logging import (
     setup_skill_logging,
     subprocess_env_with_trace,
 )
+from .media_assets import (
+    MediaAssetsStatus,
+    background_music_issue,
+    ensure_media_assets,
+    pick_background_music,
+    probe_ffmpeg,
+    probe_media_assets,
+    resolve_ffmpeg,
+    resolve_ffprobe,
+    resolve_media_assets_root,
+    resolve_music_dir,
+)
 
 try:
     from . import rpa
@@ -33,6 +45,16 @@ __all__ = [
     "EntitlementError",
     "EntitlementResult",
     "EntitlementServiceError",
+    "MediaAssetsStatus",
+    "background_music_issue",
+    "ensure_media_assets",
+    "probe_ffmpeg",
+    "probe_media_assets",
+    "pick_background_music",
+    "resolve_ffmpeg",
+    "resolve_ffprobe",
+    "resolve_media_assets_root",
+    "resolve_music_dir",
     "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

@@ -0,0 +1,740 @@
+"""共享媒体资源解析：背景音乐、字体、水印与 ffmpeg 工具。"""
+
+from __future__ import annotations
+
+import fnmatch
+import json
+import os
+import platform
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.error
+import urllib.request
+import zipfile
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+MEDIA_ASSETS_ZIP_URL = (
+    "https://git.jc2009.com/client-commons/media-assets/archive/main.zip"
+)
+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 = 128
+
+
+def _is_git_lfs_pointer(path: Path) -> bool:
+    try:
+        head = path.read_bytes()[:256]
+    except OSError:
+        return False
+    return head.startswith(_GIT_LFS_POINTER_PREFIX)
+
+
+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: Path) -> bool:
+    if not path.is_file():
+        return False
+    if path.suffix.lower() not in _AUDIO_EXTENSIONS:
+        return False
+    if _is_git_lfs_pointer(path):
+        return False
+    try:
+        if path.stat().st_size < _MIN_AUDIO_BYTES:
+            return False
+    except OSError:
+        return False
+    return True
+
+
+@dataclass
+class MediaAssetsStatus:
+    root: Path
+    exists: bool
+    ready: bool
+    source: str
+    warnings: list[str] = field(default_factory=list)
+    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
+
+
+def _env_dict(env: dict[str, str] | None) -> dict[str, str]:
+    return env if env is not None else dict(os.environ)
+
+
+def _resolve_jiangchang_data_root(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Path:
+    env_map = _env_dict(env)
+    from_env = (env_map.get("JIANGCHANG_DATA_ROOT") or "").strip()
+    if from_env:
+        return Path(from_env)
+    if data_root is not None:
+        return Path(data_root)
+    if sys.platform == "win32":
+        return _WIN_DEFAULT_DATA_ROOT
+    return _NON_WIN_DEFAULT_DATA_ROOT
+
+
+def resolve_media_assets_root(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Path:
+    env_map = _env_dict(env)
+    media_root = (env_map.get("MEDIA_ASSETS_ROOT") or "").strip()
+    if media_root:
+        return Path(media_root)
+    data = _resolve_jiangchang_data_root(data_root, env)
+    return data / "shared" / "media-assets"
+
+
+def _platform_bin_key() -> str:
+    system = platform.system().lower()
+    machine = platform.machine().lower()
+
+    if system == "windows":
+        if machine in {"amd64", "x86_64"}:
+            return "win32-x64"
+        if machine in {"arm64", "aarch64"}:
+            return "win32-arm64"
+    elif system == "darwin":
+        if machine in {"arm64", "aarch64"}:
+            return "darwin-arm64"
+        if machine in {"x86_64", "amd64"}:
+            return "darwin-x64"
+    elif system == "linux":
+        if machine in {"x86_64", "amd64"}:
+            return "linux-x64"
+        if machine in {"arm64", "aarch64"}:
+            return "linux-arm64"
+
+    return f"{system}-{machine}"
+
+
+def _tool_filename(name: str) -> str:
+    if platform.system().lower() == "windows":
+        return f"{name}.exe"
+    return name
+
+
+def _platform_tool_paths(root: Path) -> tuple[Path, Path]:
+    bin_key = _platform_bin_key()
+    bin_dir = root / "bin" / bin_key
+    return (
+        bin_dir / _tool_filename("ffmpeg"),
+        bin_dir / _tool_filename("ffprobe"),
+    )
+
+
+def _read_manifest(root: Path) -> dict | None:
+    manifest_path = root / "manifest.json"
+    if not manifest_path.is_file():
+        return None
+    try:
+        return json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def _ffmpeg_platform_spec(manifest: dict) -> dict | None:
+    tools = manifest.get("tools")
+    if not isinstance(tools, dict):
+        return None
+    ffmpeg = tools.get("ffmpeg")
+    if not isinstance(ffmpeg, dict):
+        return None
+    spec = ffmpeg.get(_platform_bin_key())
+    if isinstance(spec, dict):
+        return spec
+    return None
+
+
+def _manifest_tool_urls(spec: dict) -> list[tuple[str, str]]:
+    urls: list[tuple[str, str]] = []
+    raw_urls = spec.get("urls")
+    if isinstance(raw_urls, list):
+        for item in raw_urls:
+            if isinstance(item, dict):
+                name = str(item.get("name") or "source")
+                url = str(item.get("url") or "").strip()
+                if url:
+                    urls.append((name, url))
+    legacy = str(spec.get("url") or "").strip()
+    if legacy:
+        urls.append(("legacy", legacy))
+    return urls
+
+
+def _match_archive_member(member_name: str, pattern: str) -> bool:
+    normalized = member_name.replace("\\", "/").lstrip("/")
+    return fnmatch.fnmatch(normalized, pattern)
+
+
+def _extract_tool_members(
+    zip_path: Path,
+    archive_paths: dict[str, str],
+) -> dict[str, bytes]:
+    found: dict[str, bytes] = {}
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        for tool_name, pattern in archive_paths.items():
+            if not isinstance(pattern, str) or not pattern:
+                continue
+            for info in zf.infolist():
+                if info.is_dir():
+                    continue
+                if _match_archive_member(info.filename, pattern):
+                    found[tool_name] = zf.read(info)
+                    break
+    return found
+
+
+def _try_download_ffmpeg_tools(root: Path, warnings: list[str]) -> bool:
+    manifest = _read_manifest(root)
+    if manifest is None:
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    spec = _ffmpeg_platform_spec(manifest)
+    if spec is None:
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    archive_paths = spec.get("archive_paths")
+    if not isinstance(archive_paths, dict):
+        warnings.append("ffmpeg_manifest_missing")
+        return False
+
+    urls = _manifest_tool_urls(spec)
+    if not urls:
+        warnings.append("ffmpeg_tool_download_failed")
+        return False
+
+    ffmpeg_path, ffprobe_path = _platform_tool_paths(root)
+    bin_dir = ffmpeg_path.parent
+    download_root = _download_dir(_shared_parent(root)) / "ffmpeg-tools"
+    download_root.mkdir(parents=True, exist_ok=True)
+
+    for source_name, url in urls:
+        zip_path = download_root / f"{source_name}.zip"
+        try:
+            if zip_path.exists():
+                zip_path.unlink()
+            _download_zip(url, zip_path)
+            members = _extract_tool_members(zip_path, archive_paths)
+            ffmpeg_bytes = members.get("ffmpeg")
+            ffprobe_bytes = members.get("ffprobe")
+            if not ffmpeg_bytes or not ffprobe_bytes:
+                continue
+            bin_dir.mkdir(parents=True, exist_ok=True)
+            ffmpeg_path.write_bytes(ffmpeg_bytes)
+            ffprobe_path.write_bytes(ffprobe_bytes)
+            shutil.rmtree(download_root, ignore_errors=True)
+            return True
+        except urllib.error.URLError:
+            continue
+        except (zipfile.BadZipFile, OSError, KeyError):
+            continue
+        except Exception:
+            continue
+
+    warnings.append("ffmpeg_tool_download_failed")
+    return False
+
+
+def _maybe_ensure_ffmpeg_tools(root: Path, status: MediaAssetsStatus) -> MediaAssetsStatus:
+    if status.ready:
+        return status
+    if not status.exists:
+        return status
+    if "ffmpeg_missing" not in status.warnings and "ffprobe_missing" not in status.warnings:
+        return status
+
+    warnings = list(status.warnings)
+    if _try_download_ffmpeg_tools(root, warnings):
+        refreshed = _inspect_media_assets(root, source=status.source)
+        for warning in warnings:
+            if warning not in refreshed.warnings:
+                refreshed.warnings.append(warning)
+        return 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 _inspect_media_assets(root: Path, *, source: str) -> MediaAssetsStatus:
+    warnings: list[str] = []
+    exists = root.exists()
+    music_dir = root / "music"
+    fonts_dir = root / "fonts"
+    watermark_dir = root / "watermark"
+    ffmpeg_path, ffprobe_path = _platform_tool_paths(root)
+
+    music_exists = music_dir.is_dir()
+    fonts_exists = fonts_dir.is_dir()
+    watermark_exists = watermark_dir.is_dir()
+    ffmpeg_exists = ffmpeg_path.is_file()
+    ffprobe_exists = ffprobe_path.is_file()
+
+    if not music_exists:
+        warnings.append("music_dir_missing")
+    if not fonts_exists:
+        warnings.append("fonts_dir_missing")
+    if not watermark_exists:
+        warnings.append("watermark_dir_missing")
+    if not ffmpeg_exists:
+        warnings.append("ffmpeg_missing")
+    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
+        and fonts_exists
+        and watermark_exists
+        and ffmpeg_exists
+        and ffprobe_exists
+        and music_ready
+    )
+
+    return MediaAssetsStatus(
+        root=root,
+        exists=exists,
+        ready=ready,
+        source=source,
+        warnings=warnings,
+        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,
+    )
+
+
+def _shared_parent(root: Path) -> Path:
+    return root.parent
+
+
+def _download_dir(shared: Path) -> Path:
+    return shared / ".media-assets-download"
+
+
+def _find_media_assets_root_in_tree(extracted: Path) -> Path | None:
+    if not extracted.is_dir():
+        return None
+
+    def _looks_like_assets(path: Path) -> bool:
+        if (path / "music").is_dir():
+            return True
+        if (path / "fonts").is_dir():
+            return True
+        if (path / "watermark").is_dir():
+            return True
+        if (path / "README.md").is_file():
+            return True
+        return False
+
+    if _looks_like_assets(extracted):
+        return extracted
+
+    children = [p for p in extracted.iterdir() if p.is_dir()]
+    for child in children:
+        if _looks_like_assets(child):
+            return child
+
+    for path in extracted.rglob("*"):
+        if path.is_dir() and _looks_like_assets(path):
+            return path
+    return None
+
+
+def _download_zip(url: str, dest: Path) -> None:
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    with urllib.request.urlopen(url, timeout=120) as response:
+        dest.write_bytes(response.read())
+
+
+def _extract_zip(zip_path: Path, extract_to: Path) -> None:
+    extract_to.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        zf.extractall(extract_to)
+
+
+def _find_git_executable() -> Path | None:
+    git = shutil.which("git")
+    return Path(git) if git else None
+
+
+def _git_clone(url: str, target: Path) -> None:
+    target.parent.mkdir(parents=True, exist_ok=True)
+    subprocess.run(
+        ["git", "clone", url, str(target)],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+def _git_pull(target: Path) -> None:
+    subprocess.run(
+        ["git", "pull"],
+        cwd=str(target),
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+def _is_git_repo(path: Path) -> bool:
+    return (path / ".git").exists()
+
+
+def _install_media_assets(source: Path, target: Path) -> None:
+    target.parent.mkdir(parents=True, exist_ok=True)
+    if not target.exists():
+        shutil.move(str(source), str(target))
+        return
+
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    backup = target.parent / f"media-assets.backup-{timestamp}"
+    shutil.move(str(target), str(backup))
+    try:
+        shutil.move(str(source), str(target))
+    except Exception:
+        if backup.exists() and not target.exists():
+            shutil.move(str(backup), str(target))
+        raise
+    else:
+        shutil.rmtree(backup, ignore_errors=True)
+
+
+def _cleanup_download_artifacts(download_root: Path) -> None:
+    shutil.rmtree(download_root, ignore_errors=True)
+
+
+def _try_download_via_zip(
+    shared: Path,
+    target: Path,
+    warnings: list[str],
+) -> 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"
+
+    try:
+        if staging.exists():
+            shutil.rmtree(staging, ignore_errors=True)
+        if extracted.exists():
+            shutil.rmtree(extracted, ignore_errors=True)
+
+        _download_zip(MEDIA_ASSETS_ZIP_URL, zip_path)
+        _extract_zip(zip_path, extracted)
+
+        found = _find_media_assets_root_in_tree(extracted)
+        if found is None:
+            warnings.append("media_assets_zip_invalid_layout")
+            return False
+
+        if staging.exists():
+            shutil.rmtree(staging, ignore_errors=True)
+        shutil.copytree(found, staging)
+        _install_media_assets(staging, target)
+        _cleanup_download_artifacts(download_root)
+        return True
+    except urllib.error.URLError:
+        warnings.append("media_assets_zip_download_failed")
+        return False
+    except zipfile.BadZipFile:
+        warnings.append("media_assets_zip_extract_failed")
+        return False
+    except OSError:
+        warnings.append("media_assets_zip_extract_failed")
+        return False
+    except Exception:
+        warnings.append("media_assets_zip_extract_failed")
+        return False
+
+
+def _try_download_via_git(
+    target: Path,
+    warnings: list[str],
+    *,
+    clone: bool,
+) -> bool:
+    git_exe = _find_git_executable()
+    if git_exe is None:
+        warnings.append("git_not_available")
+        return False
+
+    try:
+        if clone:
+            if target.exists():
+                timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+                partial = target.parent / f"media-assets.git-staging-{timestamp}"
+                if partial.exists():
+                    shutil.rmtree(partial, ignore_errors=True)
+                _git_clone(MEDIA_ASSETS_GIT_URL, partial)
+                _install_media_assets(partial, target)
+            else:
+                _git_clone(MEDIA_ASSETS_GIT_URL, target)
+        else:
+            _git_pull(target)
+        return True
+    except (subprocess.CalledProcessError, OSError):
+        if clone:
+            warnings.append("media_assets_git_clone_failed")
+        return False
+
+
+def _media_assets_root_overridden(env: dict[str, str] | None) -> bool:
+    env_map = _env_dict(env)
+    return bool((env_map.get("MEDIA_ASSETS_ROOT") or "").strip())
+
+
+def _needs_media_repo_content(status: MediaAssetsStatus) -> bool:
+    return any(
+        warning in status.warnings
+        for warning in (
+            "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) -> MediaAssetsStatus:
+    if not _needs_music_content_repair(status):
+        return status
+    if _media_assets_root_overridden(None):
+        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):
+        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,
+    *,
+    update: bool = False,
+) -> MediaAssetsStatus:
+    env_map = _env_dict(env)
+    root = resolve_media_assets_root(data_root, env_map)
+    overridden = _media_assets_root_overridden(env_map)
+
+    status = _inspect_media_assets(root, source="local")
+    if overridden:
+        return _maybe_ensure_ffmpeg_tools(root, status)
+
+    if status.exists and not status.ready and not update and not _needs_media_repo_content(status):
+        return _maybe_ensure_ffmpeg_tools(root, status)
+
+    if status.ready and not update:
+        return status
+
+    if status.ready and update:
+        if _is_git_repo(root):
+            warnings: list[str] = []
+            if _try_download_via_git(root, warnings, clone=False):
+                return _maybe_ensure_ffmpeg_tools(
+                    root,
+                    _inspect_media_assets(root, source="git"),
+                )
+            status = _inspect_media_assets(root, source="local")
+            status.warnings.extend(warnings)
+            return _maybe_ensure_ffmpeg_tools(root, status)
+
+    shared = _shared_parent(root)
+    shared.mkdir(parents=True, exist_ok=True)
+    warnings: list[str] = []
+
+    if _try_download_via_zip(shared, root, warnings):
+        return _maybe_ensure_ffmpeg_tools(
+            root,
+            _maybe_repair_music_content(
+                root,
+                _inspect_media_assets(root, source="zip"),
+            ),
+        )
+
+    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,
+            _maybe_repair_music_content(
+                root,
+                _inspect_media_assets(
+                    root,
+                    source="git" if _is_git_repo(root) else "local",
+                ),
+            ),
+        )
+
+    final = _inspect_media_assets(root, source="local")
+    for warning in warnings:
+        if warning not in final.warnings:
+            final.warnings.append(warning)
+    repaired = _maybe_repair_music_content(root, final)
+    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(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Optional[Path]:
+    status = ensure_media_assets(data_root, env)
+    return status.ffmpeg_path
+
+
+def resolve_ffprobe(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Optional[Path]:
+    status = ensure_media_assets(data_root, env)
+    return status.ffprobe_path
+
+
+def resolve_music_dir(
+    data_root: str | Path | None = None,
+    env: dict[str, str] | None = None,
+) -> Optional[Path]:
+    status = ensure_media_assets(data_root, env)
+    return status.music_dir
+
+
+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,
+) -> Optional[Path]:
+    status = ensure_media_assets(data_root, env)
+    music_dir = status.music_dir
+    if music_dir is None or not music_dir.is_dir():
+        return None
+
+    candidates = [
+        path
+        for path in music_dir.rglob("*")
+        if _is_usable_audio_file(path)
+    ]
+
+    if not candidates:
+        return None
+
+    candidates.sort(key=lambda p: str(p).lower())
+    return candidates[0]

src/jiangchang_skill_core/rpa/browser.py

@@ -1,4 +1,4 @@
-"""统一 persistent context 启动封装。"""
+"""统一 persistent context 启动封装（仅浏览器自动化，不负责录屏）。"""
 
 from __future__ import annotations
 
@@ -24,12 +24,11 @@ async def launch_persistent_browser(
     executable_path: str | None = None,
     headless: bool | None = None,
     extra_args: list[str] | None = None,
-    record_video_dir: str | None = None,
 ) -> "BrowserContext":
     """用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。
 
     headless=None 时读 OPENCLAW_BROWSER_HEADLESS。
-    record_video_dir 非空则开录屏。
+    录屏由 RpaVideoSession（ffmpeg）负责，本函数不向 Playwright 传递任何录屏参数。
     """
     if headless is None:
         v = (os.environ.get("OPENCLAW_BROWSER_HEADLESS") or "0").strip().lower()
@@ -52,8 +51,6 @@ async def launch_persistent_browser(
     )
     if ignore is not None:
         launch_kwargs["ignore_default_args"] = ignore
-    if record_video_dir:
-        launch_kwargs["record_video_dir"] = record_video_dir
 
     context = await playwright.chromium.launch_persistent_context(**launch_kwargs)
     if stealth_enabled():

src/jiangchang_skill_core/rpa/video_session.py

@@ -1,45 +1,55 @@
-"""RPA 运行录屏会话：Playwright raw video → 字幕 + 背景音乐 → 最终 MP4。"""
+"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
 
 from __future__ import annotations
 
 import asyncio
 import os
-import random
 import subprocess
 import sys
 import time
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from 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,
+)
 
-# 与 screencast/composer.py 保持一致的字幕样式
 _JIANGCHANG_SUBTITLE_STYLE = (
     "FontName=Arial,FontSize=14,"
     "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
     "Outline=2,Shadow=1,Alignment=2"
 )
 
-_DEFAULT_MUSIC_ROOT = os.environ.get(
+_CAPTURE_FRAMERATE = 15
-    "MEDIA_ASSETS_ROOT",
-    r"D:\OpenClaw\client-commons\media-assets",
-)
 
 
 def _record_video_enabled() -> bool:
     return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
 
 
-def _resolve_music_file() -> Optional[Path]:
+def _resolve_ffmpeg_exe() -> Optional[Path]:
-    music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music"
+    ffmpeg = resolve_ffmpeg()
-    if not music_dir.is_dir():
+    if ffmpeg is None:
         return None
-    files = list(music_dir.rglob("*.mp3"))
+    path = Path(ffmpeg)
-    if not files:
+    return path if path.is_file() else None
+
+
+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
-    return random.choice(files)
 
 
 @dataclass
@@ -50,7 +60,6 @@ class _StepEntry:
 
 
 def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
-    """根据步骤时间线生成 SRT（防重叠）。"""
     MIN_DURATION = 1.0
     GAP = 0.1
     steps = sorted(entries, key=lambda e: e.start)
@@ -86,17 +95,12 @@ def _escape_srt_path(path: str) -> str:
     return path.replace("\\", "/").replace(":", "\\:")
 
 
-def _find_raw_webm(raw_dir: str) -> Optional[Path]:
+def _ffmpeg_cmd(ffmpeg_exe: Path, *args: str) -> List[str]:
-    root = Path(raw_dir)
+    return [str(ffmpeg_exe), *args]
-    if not root.is_dir():
-        return None
-    webms = list(root.rglob("*.webm"))
-    if not webms:
-        return None
-    return max(webms, key=lambda p: p.stat().st_size)
 
 
-def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
+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,
@@ -113,8 +117,32 @@ def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
     return True, ""
 
 
-def _compose_webm_mp4(
+def _build_desktop_record_cmd(ffmpeg_exe: Path, capture_path: str) -> Optional[List[str]]:
-    webm_path: Path,
+    """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,
+    )
+
+
+def _compose_capture_mp4(
+    ffmpeg_exe: Path,
+    capture_path: Path,
     srt_path: Path,
     output_path: Path,
     *,
@@ -133,40 +161,59 @@ def _compose_webm_mp4(
             f"[0:v]{subtitle_filter}[vout];"
             f"[1:a]volume={music_volume},apad[aout]"
         )
-        cmd = [
+        cmd_tail = [
-            "ffmpeg", "-y",
+            "-y",
-            "-i", str(webm_path),
+            "-i",
-            "-i", str(music_file),
+            str(capture_path),
-            "-filter_complex", filter_complex,
+            "-i",
-            "-map", "[vout]",
+            str(music_file),
-            "-map", "[aout]",
+            "-filter_complex",
-            "-c:v", "libx264",
+            filter_complex,
-            "-preset", "fast",
+            "-map",
-            "-crf", "23",
+            "[vout]",
-            "-c:a", "aac",
+            "-map",
-            "-b:a", "128k",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
             "-shortest",
-            "-pix_fmt", "yuv420p",
+            "-pix_fmt",
+            "yuv420p",
             str(output_path),
         ]
     else:
-        cmd = [
+        cmd_tail = [
-            "ffmpeg", "-y",
+            "-y",
-            "-i", str(webm_path),
+            "-i",
-            "-vf", subtitle_filter,
+            str(capture_path),
-            "-c:v", "libx264",
+            "-vf",
-            "-preset", "fast",
+            subtitle_filter,
-            "-crf", "23",
+            "-c:v",
-            "-c:a", "aac",
+            "libx264",
-            "-b:a", "128k",
+            "-preset",
-            "-pix_fmt", "yuv420p",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-pix_fmt",
+            "yuv420p",
             str(output_path),
         ]
-    return _run_ffmpeg(cmd)
+    return _run_ffmpeg(ffmpeg_exe, cmd_tail)
 
 
 class RpaVideoSession:
-    """Service/RPA 层统一录屏会话；OPENCLAW_RECORD_VIDEO=1 时启用完整成片流程。"""
+    """Service/RPA 层统一 ffmpeg 桌面录屏；OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
 
     def __init__(
         self,
@@ -183,7 +230,7 @@ class RpaVideoSession:
         self.enabled = _record_video_enabled()
         self.warnings: List[str] = []
         self.output_video_path: Optional[str] = None
-        self.record_video_dir: Optional[str] = None
+        self.capture_path: Optional[str] = None
         self.artifacts: Dict[str, Any] = {}
 
         ts = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -194,24 +241,44 @@ class RpaVideoSession:
         self._artifact_root = os.path.join(
             self.skill_data_dir, "rpa-artifacts", batch_id
         )
-        self._raw_video_dir = os.path.join(self._artifact_root, "raw-video")
         self._subtitles_dir = os.path.join(self._artifact_root, "subtitles")
         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._planned_output = os.path.join(self._videos_dir, f"{self._video_stem}.mp4")
 
         self._steps: List[_StepEntry] = []
         self._t0: Optional[float] = None
+        self._ffmpeg_proc: Optional[subprocess.Popen] = None
+        self._ffmpeg_record_log_file: Optional[IO[str]] = None
 
         if self.enabled:
             os.makedirs(self._videos_dir, exist_ok=True)
-            os.makedirs(self._raw_video_dir, exist_ok=True)
             os.makedirs(self._subtitles_dir, exist_ok=True)
             os.makedirs(self._logs_dir, exist_ok=True)
-            self.record_video_dir = self._raw_video_dir
+            os.makedirs(self._artifact_root, exist_ok=True)
+
+    def _record_log_hint(self) -> str:
+        if self.enabled and self._ffmpeg_record_log_path:
+            return f"; record_log={self._ffmpeg_record_log_path}"
+        return ""
+
+    def _append_warning(self, code: str, detail: str = "") -> None:
+        msg = code if not detail else f"{code}: {detail}"
+        if code in (
+            "ffmpeg_record_start_failed",
+            "ffmpeg_capture_missing",
+            "ffmpeg_compose_failed",
+        ):
+            msg += self._record_log_hint()
+        self.warnings.append(msg)
 
     def step(self, text: str, *, duration: float = 4.0) -> None:
-        """记录一步骤字幕（中文说明）。"""
+        self.add_step(text, duration=duration)
+
+    def add_step(self, text: str, *, duration: float = 4.0) -> None:
+        """记录一步骤字幕（相对录屏开始时间）。"""
         if not self.enabled:
             return
         if self._t0 is None:
@@ -219,71 +286,198 @@ class RpaVideoSession:
         elapsed = time.monotonic() - self._t0
         self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration))
 
+    def _close_ffmpeg_record_log(self) -> None:
+        fh = self._ffmpeg_record_log_file
+        self._ffmpeg_record_log_file = None
+        if fh is None:
+            return
+        try:
+            fh.close()
+        except Exception:
+            pass
+
+    def _start_ffmpeg_capture(self) -> None:
+        if not self.enabled:
+            return
+
+        ffmpeg_exe = _resolve_ffmpeg_exe()
+        if ffmpeg_exe is None:
+            self.warnings.append("ffmpeg_not_found")
+            return
+
+        cmd = _build_desktop_record_cmd(ffmpeg_exe, self._capture_path)
+        if cmd is None:
+            self._append_warning(
+                "ffmpeg_record_start_failed",
+                "desktop capture only supported on Windows",
+            )
+            return
+
+        stderr_target: Any = subprocess.DEVNULL
+        try:
+            os.makedirs(self._logs_dir, exist_ok=True)
+            self._ffmpeg_record_log_file = open(
+                self._ffmpeg_record_log_path, "a", encoding="utf-8"
+            )
+            stderr_target = self._ffmpeg_record_log_file
+        except Exception as exc:
+            self.warnings.append(f"ffmpeg_record_log_open_failed: {exc}")
+
+        try:
+            self._ffmpeg_proc = subprocess.Popen(
+                cmd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,
+                stderr=stderr_target,
+            )
+            self.capture_path = self._capture_path
+        except Exception as exc:
+            self._append_warning("ffmpeg_record_start_failed", str(exc))
+            self._ffmpeg_proc = None
+            self._close_ffmpeg_record_log()
+
+    def _stop_ffmpeg_capture(self) -> None:
+        proc = self._ffmpeg_proc
+        if proc is None:
+            self._close_ffmpeg_record_log()
+            return
+        if proc.poll() is None:
+            try:
+                if proc.stdin:
+                    proc.stdin.write(b"q")
+                    proc.stdin.flush()
+            except Exception:
+                try:
+                    proc.terminate()
+                except Exception:
+                    pass
+            try:
+                proc.wait(timeout=20)
+            except subprocess.TimeoutExpired:
+                try:
+                    proc.kill()
+                except Exception:
+                    pass
+                try:
+                    proc.wait(timeout=5)
+                except Exception:
+                    pass
+        self._ffmpeg_proc = None
+        self._close_ffmpeg_record_log()
+        time.sleep(0.3)
+
     async def __aenter__(self) -> "RpaVideoSession":
         if self.enabled:
             self._t0 = time.monotonic()
             if self.title:
-                self.step(self.title, duration=3.0)
+                self.add_step(self.title, duration=3.0)
+            self._start_ffmpeg_capture()
         return self
 
     async def __aexit__(self, exc_type, exc, tb) -> None:
         if self.enabled:
+            self._stop_ffmpeg_capture()
             await self.finalize()
 
     async def finalize(self) -> None:
-        """在 Playwright context 已 close 后合成最终 MP4；失败只记 warning。"""
+        """停止录制后合成最终 MP4；失败只记 warning，不抛异常。"""
         if not self.enabled:
+            self.artifacts = self.summary()
             return
 
         if not self._steps and self.title:
-            self.step(self.title)
+            self.add_step(self.title)
 
         try:
-            await asyncio.sleep(0.5)
+            capture = Path(self._capture_path)
-            webm = _find_raw_webm(self._raw_video_dir)
+            if not capture.is_file() or capture.stat().st_size <= 0:
-            if webm is None:
+                self._append_warning("ffmpeg_capture_missing")
-                self.warnings.append("no_playwright_raw_video")
+                self.artifacts = self.summary()
-                self.artifacts = self.to_artifact_dict()
                 return
 
+            self.capture_path = str(capture.resolve())
+
             entries = list(self._steps)
             if entries:
                 entries[-1].duration = max(entries[-1].duration, 5.0)
+            try:
                 _generate_srt(entries, self._srt_path)
+            except Exception as exc:
+                self.warnings.append(f"subtitle_write_failed: {exc}")
 
-            music = _resolve_music_file()
+            music = _resolve_music_file(self.warnings)
-            if music is None:
+            if music is None and "background_music_lfs_pointer_only" not in self.warnings and "background_music_invalid" not in self.warnings:
                 self.warnings.append("background_music_missing")
 
-            ok, err = _compose_webm_mp4(
+            ffmpeg_exe = _resolve_ffmpeg_exe()
-                webm,
+            if ffmpeg_exe is None:
-                Path(self._srt_path),
+                self.warnings.append("ffmpeg_not_found")
+                self.artifacts = self.summary()
+                return
+
+            srt_file = Path(self._srt_path)
+            use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
+
+            ok, err = _compose_capture_mp4(
+                ffmpeg_exe,
+                capture,
+                srt_file if use_srt else Path(self._srt_path),
                 Path(self._planned_output),
                 music_file=music,
             )
             if not ok and music is not None:
                 self.warnings.append(f"ffmpeg_with_music_failed: {err}")
-                ok, err = _compose_webm_mp4(
+                ok, err = _compose_capture_mp4(
-                    webm,
+                    ffmpeg_exe,
-                    Path(self._srt_path),
+                    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,
+                    [
+                        "-y",
+                        "-i",
+                        str(capture),
+                        "-c:v",
+                        "libx264",
+                        "-preset",
+                        "fast",
+                        "-crf",
+                        "23",
+                        "-pix_fmt",
+                        "yuv420p",
+                        str(self._planned_output),
+                    ],
+                )
             if not ok:
-                self.warnings.append(f"ffmpeg_compose_failed: {err}")
+                self._append_warning("ffmpeg_compose_failed", err)
-                self.artifacts = self.to_artifact_dict()
+                self.artifacts = self.summary()
                 return
 
             self.output_video_path = str(Path(self._planned_output).resolve())
-            self.artifacts = self.to_artifact_dict()
         except Exception as exc:
             self.warnings.append(f"video_finalize_error: {exc}")
-            self.artifacts = self.to_artifact_dict()
 
-    def to_artifact_dict(self) -> Dict[str, Any]:
+        self.artifacts = self.summary()
+
+    def summary(self) -> Dict[str, Any]:
+        record_log = (
+            os.path.abspath(self._ffmpeg_record_log_path)
+            if self.enabled and self._ffmpeg_record_log_path
+            else None
+        )
         return {
             "enabled": self.enabled,
             "path": self.output_video_path,
+            "capture_path": self.capture_path,
+            "record_log_path": record_log,
             "warnings": list(self.warnings),
-            "batch_id": self.batch_id,
         }
+
+    def to_artifact_dict(self) -> Dict[str, Any]:
+        data = self.summary()
+        data["batch_id"] = self.batch_id
+        return data

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():
+    if not (
-        os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
+        os.getenv("CLAW_DATA_ROOT")
-    if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
+        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/jiangchang_skill_core/unified_logging.py

@@ -10,17 +10,23 @@ import os
 import sys
 import uuid
 from logging.handlers import TimedRotatingFileHandler
-from typing import Optional
+from typing import Optional, Tuple
 
 from .runtime_env import get_data_root, get_user_id
 
 _skill_slug: str = ""
 _logger_name: str = ""
 
+# 避免 logging 内部错误再向 stderr 打印 "--- Logging error ---"
+logging.raiseExceptions = False
+
 
 def get_unified_logs_dir() -> str:
     path = os.path.join(get_data_root(), get_user_id(), "logs")
+    try:
         os.makedirs(path, exist_ok=True)
+    except OSError:
+        pass
     return path
 
 
@@ -30,11 +36,27 @@ def get_skill_log_file_path() -> str:
     if override:
         parent = os.path.dirname(os.path.abspath(override))
         if parent:
+            try:
                 os.makedirs(parent, exist_ok=True)
+            except OSError:
+                pass
         return os.path.abspath(override)
     return os.path.join(get_unified_logs_dir(), "jiangchang.log")
 
 
+def _can_open_log_file(log_path: str) -> Tuple[bool, str]:
+    """探测日志文件是否可追加写入（真实 open，非 os.access）。"""
+    try:
+        parent = os.path.dirname(os.path.abspath(log_path))
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        with open(log_path, "a", encoding="utf-8"):
+            pass
+        return True, ""
+    except Exception as exc:
+        return False, str(exc)
+
+
 def ensure_trace_for_process() -> str:
     """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。"""
     existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
@@ -81,44 +103,72 @@ def _backup_count() -> int:
         return 30
 
 
+def _clear_logger_handlers(log: logging.Logger) -> None:
+    for h in list(log.handlers):
+        log.removeHandler(h)
+        try:
+            h.close()
+        except Exception:
+            pass
+
+
+def _attach_fallback_handler(log: logging.Logger) -> None:
+    """文件日志不可用时仅挂 NullHandler，不向 stderr 打印 logging 异常。"""
+    log.addHandler(logging.NullHandler())
+
+
 def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
     """
     幂等：为指定 logger 挂载统一文件日志与可选 stderr。
     须在进程早期调用（如 CLI main 在业务日志之前）。
+    文件日志不可写时降级为 NullHandler，不影响 CLI 返回码。
     """
     global _skill_slug, _logger_name
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     _logger_name = logger_name
 
     log = logging.getLogger(logger_name)
-    if log.handlers:
+    _clear_logger_handlers(log)
-        return
     log.setLevel(_log_level_from_env())
-    path = get_skill_log_file_path()
+    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
+    filt = _SkillContextFilter()
+
+    log_path = get_skill_log_file_path()
+    ok, _err = _can_open_log_file(log_path)
+    if ok:
+        try:
             fh = TimedRotatingFileHandler(
-        path,
+                log_path,
                 when="midnight",
                 interval=1,
                 backupCount=_backup_count(),
                 encoding="utf-8",
-        delay=True,
+                delay=False,
             )
-    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
             fh.setFormatter(fmt)
-    fh.addFilter(_SkillContextFilter())
+            fh.addFilter(filt)
             log.addHandler(fh)
+        except Exception:
+            _attach_fallback_handler(log)
+    else:
+        _attach_fallback_handler(log)
+
     if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
         "1",
         "true",
         "yes",
         "on",
     ):
+        try:
             sh = logging.StreamHandler(sys.stderr)
             sh.setLevel(logging.WARNING)
             sh.setFormatter(fmt)
-        sh.addFilter(_SkillContextFilter())
+            sh.addFilter(filt)
             log.addHandler(sh)
+        except Exception:
+            pass
     log.propagate = False
 
 
@@ -139,18 +189,25 @@ def attach_unified_file_handler(
     独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。
     """
     global _skill_slug
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     lg = logging.getLogger(logger_name)
-    lg.handlers.clear()
+    _clear_logger_handlers(lg)
     lg.setLevel(level)
-    parent = os.path.dirname(os.path.abspath(log_path))
-    if parent:
-        os.makedirs(parent, exist_ok=True)
-    fh = logging.FileHandler(log_path, encoding="utf-8")
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
+    filt = _SkillContextFilter()
+
+    ok, _err = _can_open_log_file(log_path)
+    if ok:
+        try:
+            fh = logging.FileHandler(log_path, encoding="utf-8", delay=False)
             fh.setFormatter(fmt)
-    fh.addFilter(_SkillContextFilter())
+            fh.addFilter(filt)
             lg.addHandler(fh)
+        except Exception:
+            _attach_fallback_handler(lg)
+    else:
+        _attach_fallback_handler(lg)
     lg.propagate = False
     return lg

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:
+    allow_no_music: bool = True,
-    frames_dir = Path(frames_dir)
+    media_assets_root: str | Path | None = None,
-    subtitle_path = Path(subtitle_path)
+    media_assets_env: dict[str, str] | None = None,
-    output_path = Path(output_path)
+) -> ComposeVideoResult:
-    output_path.parent.mkdir(parents=True, exist_ok=True)
+    """合成 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] = []
 
-    # 随机选一首背景音乐
+    assets_env = _media_assets_env(media_assets_root, media_assets_env)
-    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}")
 
-    # FFmpeg 字幕路径在 Windows 下需要转义冒号（subtitles 滤镜内部语法）
+    ffmpeg_exe = resolve_ffmpeg(env=assets_env)
-    srt_path_str = str(subtitle_path).replace("\\", "/").replace(":", "\\:")
+    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 混用报错
+    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 = [
-    cmd = [
+            "-y",
-        "ffmpeg", "-y",
+            "-framerate",
-        "-framerate", str(fps),
+            str(fps),
-        "-i", str(frames_dir / "frame_%06d.png"),
+            "-i",
-        "-i", str(music_file),
+            str(frames_dir_path / "frame_%06d.png"),
-        "-filter_complex", filter_complex,
+            "-i",
-        "-map", "[vout]",
+            str(music_file),
-        "-map", "[aout]",
+            "-filter_complex",
-        "-c:v", "libx264",
+            filter_complex,
-        "-preset", "fast",
+            "-map",
-        "-crf", "23",
+            "[vout]",
-        "-c:a", "aac",
+            "-map",
-        "-b:a", "128k",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
             "-shortest",
-        "-pix_fmt", "yuv420p",
+            "-pix_fmt",
-        str(output_path),
+            "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)
 
-    print(f"[screencast] 运行 FFmpeg 合成...")
+    cmd_tail = [
-    result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
+        "-y",
-    if result.returncode != 0:
+        "-framerate",
-        print(result.stderr, file=sys.stderr)
+        str(fps),
-        raise RuntimeError(f"FFmpeg 合成失败，退出码 {result.returncode}")
+        "-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] 输出: {output_path}")
+    print(f"[screencast] 输出: {output_path_obj}")
-    return str(output_path)
+    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: media-assets 根目录；不传时使用
-                           不传时读 MEDIA_ASSETS_ROOT 环境变量，
+                           ``resolve_media_assets_root()`` 默认路径
-                           再不存在则用默认值 D:\\OpenClaw\\client-commons\\media-assets
+                           （JIANGCHANG_DATA_ROOT/shared/media-assets 或 MEDIA_ASSETS_ROOT）
-        music_subdir:     music_assets_root 下音乐子目录，默认 "music"
+        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:
+    if music_subdir != "music":
-        media_assets_root = os.environ.get(
+        warnings.warn(
-            "MEDIA_ASSETS_ROOT",
+            "music_subdir is deprecated and ignored; "
-            r"D:\OpenClaw\client-commons\media-assets",
+            "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

@@ -0,0 +1,342 @@
+# -*- coding: utf-8 -*-
+"""media_assets 共享媒体资源解析测试。"""
+from __future__ import annotations
+
+import json
+import platform
+import zipfile
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+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" * 256)
+    (root / "fonts").mkdir(parents=True)
+    (root / "watermark").mkdir(parents=True)
+    bin_key = ma._platform_bin_key()
+    bin_dir = root / "bin" / bin_key
+    bin_dir.mkdir(parents=True)
+    if with_ffmpeg:
+        (bin_dir / ma._tool_filename("ffmpeg")).write_bytes(b"ffmpeg")
+    if with_ffprobe:
+        (bin_dir / ma._tool_filename("ffprobe")).write_bytes(b"ffprobe")
+
+
+def _sample_manifest() -> dict:
+    bin_key = ma._platform_bin_key()
+    return {
+        "name": "media-assets",
+        "version": "0.2.0",
+        "directories": {"music": "music", "fonts": "fonts", "watermark": "watermark"},
+        "tools": {
+            "ffmpeg": {
+                bin_key: {
+                    "package": "ffmpeg-release-essentials",
+                    "archive_type": "zip",
+                    "urls": [{"name": "test", "url": "https://example.test/ffmpeg.zip"}],
+                    "archive_paths": {
+                        "ffmpeg": "*/bin/ffmpeg.exe",
+                        "ffprobe": "*/bin/ffprobe.exe",
+                    },
+                }
+            }
+        },
+    }
+
+
+def _install_tools_from_manifest(root: Path, warnings: list[str]) -> bool:
+    bin_key = ma._platform_bin_key()
+    bin_dir = root / "bin" / bin_key
+    bin_dir.mkdir(parents=True, exist_ok=True)
+    (bin_dir / ma._tool_filename("ffmpeg")).write_bytes(b"ffmpeg")
+    (bin_dir / ma._tool_filename("ffprobe")).write_bytes(b"ffprobe")
+    return True
+
+
+def test_media_assets_root_env_priority(tmp_path: Path) -> None:
+    custom = tmp_path / "custom-media"
+    custom.mkdir()
+    env = {
+        "MEDIA_ASSETS_ROOT": str(custom),
+        "JIANGCHANG_DATA_ROOT": str(tmp_path / "data"),
+    }
+    assert ma.resolve_media_assets_root(env=env) == custom
+
+
+def test_default_media_assets_path(tmp_path: Path) -> None:
+    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")}
+    expected = tmp_path / "data" / "shared" / "media-assets"
+    assert ma.resolve_media_assets_root(env=env) == expected
+
+
+def test_complete_assets_skips_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root)
+
+    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.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is True
+    assert status.warnings == []
+    assert status.source == "local"
+
+
+def test_missing_ffmpeg_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root, with_ffmpeg=False)
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is False
+    assert "ffmpeg_missing" in status.warnings
+
+
+def test_missing_ffprobe_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root, with_ffprobe=False)
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is False
+    assert "ffprobe_missing" in status.warnings
+
+
+def test_pick_background_music_stable_first(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    root = tmp_path / "data" / "shared" / "media-assets"
+    _make_complete_assets(root)
+    music = root / "music"
+    (music / "b-upbeat.mp3").write_bytes(b"x" * 256)
+    (music / "calm" / "a-calm.mp3").write_bytes(b"x" * 256)
+    (music / "z-last.wav").write_bytes(b"x" * 256)
+
+    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 / "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/calm/track.mp3", b"x" * 256)
+        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:
+    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")
+
+    def _fake_download(url: str, dest: Path) -> None:
+        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)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(data_root)})
+    assert target.is_dir()
+    assert status.ready is True
+    assert status.source == "zip"
+    assert "media_assets_zip_invalid_layout" not in status.warnings
+
+
+def test_zip_download_failure_returns_warning(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    import urllib.error
+
+    def _raise_url_error(url: str, dest: Path) -> None:
+        raise urllib.error.URLError("network down")
+
+    monkeypatch.setattr(ma, "_download_zip", _raise_url_error)
+    monkeypatch.setattr(ma, "_find_git_executable", lambda: None)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is False
+    assert "media_assets_zip_download_failed" in status.warnings
+    assert "git_not_available" in status.warnings
+
+
+def test_git_not_available_when_zip_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    import urllib.error
+
+    monkeypatch.setattr(
+        ma,
+        "_download_zip",
+        lambda url, dest: (_ for _ in ()).throw(urllib.error.URLError("fail")),
+    )
+    monkeypatch.setattr(ma, "_find_git_executable", lambda: None)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert "git_not_available" in status.warnings
+
+
+def test_media_assets_root_override_no_download(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    custom = tmp_path / "override"
+    custom.mkdir()
+
+    def _fail_download(*args, **kwargs):
+        raise AssertionError("download should not be called when MEDIA_ASSETS_ROOT is set")
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", _fail_download)
+
+    status = ma.ensure_media_assets(env={"MEDIA_ASSETS_ROOT": str(custom)})
+    assert status.root == custom
+    assert status.exists is True
+
+
+@pytest.mark.skipif(platform.system().lower() != "windows", reason="Windows default path check")
+def test_windows_default_data_root() -> None:
+    with patch.dict("os.environ", {}, clear=True):
+        root = ma.resolve_media_assets_root()
+    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_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" * 256)
+    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" / "calm").mkdir(parents=True)
+    (root / "music" / "calm" / "track.mp3").write_bytes(b"x" * 256)
+    (root / "fonts").mkdir(parents=True)
+    (root / "watermark").mkdir(parents=True)
+    (root / "manifest.json").write_text(
+        json.dumps(_sample_manifest(), ensure_ascii=False),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(ma, "_try_download_via_zip", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_via_git", lambda *a, **k: False)
+    monkeypatch.setattr(ma, "_try_download_ffmpeg_tools", _install_tools_from_manifest)
+
+    status = ma.ensure_media_assets(env={"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")})
+    assert status.ready is True
+    assert status.ffmpeg_path is not None
+    assert status.ffprobe_path is not None

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_unified_logging.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""统一日志：文件不可写时降级，首次 emit 不抛 PermissionError。"""
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+import unittest
+from logging.handlers import TimedRotatingFileHandler
+from unittest.mock import patch
+
+from jiangchang_skill_core import unified_logging as ul
+
+
+class TestUnifiedLogging(unittest.TestCase):
+    def setUp(self) -> None:
+        ul._skill_slug = ""
+        ul._logger_name = ""
+
+    def test_can_open_log_file_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "test.log")
+            ok, err = ul._can_open_log_file(path)
+            self.assertTrue(ok)
+            self.assertEqual(err, "")
+            self.assertTrue(os.path.isfile(path))
+
+    def test_setup_skill_logging_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "skill.log")
+            logger_name = "test.logger.writable"
+            with patch.object(ul, "get_skill_log_file_path", return_value=log_path):
+                ul.setup_skill_logging("receive-order", logger_name)
+                ul.get_skill_logger().info("hello")
+                self.assertTrue(os.path.isfile(log_path))
+                ul._clear_logger_handlers(logging.getLogger(logger_name))
+
+    def test_setup_skill_logging_unwritable_no_file_handler(self) -> None:
+        logger_name = "test.logger.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            ul.setup_skill_logging("receive-order", logger_name)
+            lg = logging.getLogger(logger_name)
+            lg.info("hello")
+        handlers = logging.getLogger(logger_name).handlers
+        self.assertFalse(any(isinstance(h, TimedRotatingFileHandler) for h in handlers))
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in handlers))
+
+    def test_attach_unified_file_handler_unwritable(self) -> None:
+        logger_name = "test.logger.attach.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            lg = ul.attach_unified_file_handler(
+                "/no/access/test.log",
+                skill_slug="receive-order",
+                logger_name=logger_name,
+            )
+            lg.info("hello")
+        self.assertFalse(
+            any(isinstance(h, logging.FileHandler) for h in lg.handlers)
+        )
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in lg.handlers))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_video_session.py

@@ -1,11 +1,14 @@
 # -*- coding: utf-8 -*-
-"""RpaVideoSession 路径与 no-op 行为。"""
+"""RpaVideoSession：ffmpeg 桌面录屏。"""
 from __future__ import annotations
 
+import asyncio
 import os
+import sys
 import tempfile
 import unittest
-
+from pathlib import Path
+from unittest.mock import MagicMock, patch
 from jiangchang_skill_core import config
 from jiangchang_skill_core.rpa.video_session import RpaVideoSession
 
@@ -25,14 +28,17 @@ class TestRpaVideoSession(unittest.TestCase):
                     batch_id="batch_test",
                 )
                 self.assertFalse(session.enabled)
-                self.assertIsNone(session.record_video_dir)
+                summ = session.summary()
-                self.assertIsNone(session.output_video_path)
+                self.assertFalse(summ["enabled"])
+                self.assertIsNone(summ["path"])
+                self.assertIsNone(summ["capture_path"])
+                self.assertIsNone(summ["record_log_path"])
         finally:
             if old is not None:
                 os.environ["OPENCLAW_RECORD_VIDEO"] = old
             config.reset_cache()
 
-    def test_enabled_paths(self) -> None:
+    def test_enabled_paths_use_ffmpeg_capture(self) -> None:
         os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
         try:
             with tempfile.TemporaryDirectory() as tmp:
@@ -45,8 +51,13 @@ class TestRpaVideoSession(unittest.TestCase):
                 )
                 self.assertTrue(session.enabled)
                 self.assertTrue(
-                    session.record_video_dir.replace("\\", "/").endswith(
+                    session._capture_path.replace("\\", "/").endswith(
-                        "/rpa-artifacts/exmail_20260602_120000/raw-video"
+                        "/rpa-artifacts/exmail_20260602_120000/capture.mp4"
+                    )
+                )
+                self.assertTrue(
+                    session._ffmpeg_record_log_path.replace("\\", "/").endswith(
+                        "/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log"
                     )
                 )
                 self.assertTrue(
@@ -54,15 +65,65 @@ class TestRpaVideoSession(unittest.TestCase):
                         tmp.replace("\\", "/") + "/videos/"
                     )
                 )
-                self.assertNotIn(
-                    "rpa-artifacts",
-                    session._planned_output.replace("\\", "/").split("/videos/")[-1],
-                )
         finally:
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
             config.reset_cache()
 
-    def test_finalize_missing_webm_only_warning(self) -> None:
+    @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture")
+    @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, 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()
+        mock_popen.return_value = proc
+
+        os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
+        try:
+            with tempfile.TemporaryDirectory() as tmp:
+                config.reset_cache()
+
+                async def _run() -> None:
+                    async with RpaVideoSession(
+                        skill_slug="receive-order",
+                        skill_data_dir=tmp,
+                        batch_id="batch_ff",
+                    ) as video:
+                        video.add_step("测试步骤")
+
+                with patch.object(RpaVideoSession, "finalize", return_value=None):
+                    asyncio.run(_run())
+
+                self.assertTrue(mock_popen.called)
+                cmd = mock_popen.call_args[0][0]
+                self.assertTrue(str(cmd[0]).endswith("ffmpeg.exe"))
+                self.assertIn("gdigrab", cmd)
+                self.assertIn("desktop", cmd)
+                kwargs = mock_popen.call_args[1]
+                self.assertIsNot(kwargs.get("stderr"), os.devnull)
+        finally:
+            os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+            config.reset_cache()
+
+    @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:
+                config.reset_cache()
+                session = RpaVideoSession(
+                    skill_slug="receive-order",
+                    skill_data_dir=tmp,
+                    batch_id="batch_no_ff",
+                )
+                session._start_ffmpeg_capture()
+                self.assertIn("ffmpeg_not_found", session.warnings)
+        finally:
+            os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+            config.reset_cache()
+
+    def test_finalize_missing_capture_warning(self) -> None:
         os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
         try:
             with tempfile.TemporaryDirectory() as tmp:
@@ -72,10 +133,11 @@ class TestRpaVideoSession(unittest.TestCase):
                     skill_data_dir=tmp,
                     batch_id="batch_warn",
                 )
-                import asyncio
-
                 asyncio.run(session.finalize())
-                self.assertIn("no_playwright_raw_video", session.warnings)
+                self.assertTrue(
+                    any("ffmpeg_capture_missing" in w for w in session.warnings)
+                )
+                self.assertIn("record_log=", "".join(session.warnings))
                 self.assertIsNone(session.output_video_path)
         finally:
             os.environ.pop("OPENCLAW_RECORD_VIDEO", None)