fix: unified logging probe before file handler attach

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

src/jiangchang_skill_core/rpa/__init__.py

@@ -9,6 +9,11 @@ try:
 except ImportError:
     launch_persistent_browser = None  # type: ignore[misc, assignment]
 
+try:
+    from .video_session import RpaVideoSession
+except ImportError:
+    RpaVideoSession = None  # type: ignore[misc, assignment]
+
 __all__ = [
     "anti_detect",
     "artifacts",
@@ -16,6 +21,7 @@ __all__ = [
     "human_login",
     "stealth",
     "launch_persistent_browser",
+    "RpaVideoSession",
     # re-export common anti_detect helpers
     "random_delay",
     "human_delay_short",

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

@@ -0,0 +1,461 @@
+"""RPA 运行录屏会话：ffmpeg 桌面录制 → 字幕 + 背景音乐 → 最终 MP4。"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import random
+import shutil
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, IO, List, Optional
+
+from .. import config
+
+_JIANGCHANG_SUBTITLE_STYLE = (
+    "FontName=Arial,FontSize=14,"
+    "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
+    "Outline=2,Shadow=1,Alignment=2"
+)
+
+_DEFAULT_MUSIC_ROOT = os.environ.get(
+    "MEDIA_ASSETS_ROOT",
+    r"D:\OpenClaw\client-commons\media-assets",
+)
+
+_CAPTURE_FRAMERATE = 15
+
+
+def _record_video_enabled() -> bool:
+    return config.get_bool("OPENCLAW_RECORD_VIDEO", default=False)
+
+
+def _resolve_music_file() -> Optional[Path]:
+    music_dir = Path(_DEFAULT_MUSIC_ROOT) / "music"
+    if not music_dir.is_dir():
+        return None
+    files = list(music_dir.rglob("*.mp3"))
+    if not files:
+        return None
+    return random.choice(files)
+
+
+@dataclass
+class _StepEntry:
+    start: float
+    text: str
+    duration: float = 4.0
+
+
+def _generate_srt(entries: List[_StepEntry], output_path: str) -> None:
+    MIN_DURATION = 1.0
+    GAP = 0.1
+    steps = sorted(entries, key=lambda e: e.start)
+    if not steps:
+        steps = [_StepEntry(0.0, "任务执行中", 4.0)]
+
+    for i in range(len(steps) - 1):
+        cur, nxt = steps[i], steps[i + 1]
+        max_end = nxt.start - GAP
+        new_dur = max_end - cur.start
+        if new_dur < MIN_DURATION:
+            cur.duration = MIN_DURATION
+            nxt.start = cur.start + MIN_DURATION + GAP
+        else:
+            cur.duration = min(cur.duration, new_dur)
+
+    def _fmt(s: float) -> str:
+        h = int(s // 3600)
+        m = int((s % 3600) // 60)
+        sec = int(s % 60)
+        ms = int((s % 1) * 1000)
+        return f"{h:02d}:{m:02d}:{sec:02d},{ms:03d}"
+
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w", encoding="utf-8") as f:
+        for i, e in enumerate(steps, 1):
+            f.write(f"{i}\n")
+            f.write(f"{_fmt(e.start)} --> {_fmt(e.start + e.duration)}\n")
+            f.write(f"{e.text}\n\n")
+
+
+def _escape_srt_path(path: str) -> str:
+    return path.replace("\\", "/").replace(":", "\\:")
+
+
+def _run_ffmpeg(cmd: List[str]) -> tuple[bool, str]:
+    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 _build_desktop_record_cmd(capture_path: str) -> Optional[List[str]]:
+    """Windows gdigrab 全桌面录制；非 Windows 返回 None。"""
+    if sys.platform != "win32":
+        return None
+    return [
+        "ffmpeg",
+        "-y",
+        "-f",
+        "gdigrab",
+        "-framerate",
+        str(_CAPTURE_FRAMERATE),
+        "-i",
+        "desktop",
+        "-c:v",
+        "libx264",
+        "-preset",
+        "ultrafast",
+        "-pix_fmt",
+        "yuv420p",
+        capture_path,
+    ]
+
+
+def _compose_capture_mp4(
+    capture_path: Path,
+    srt_path: Path,
+    output_path: Path,
+    *,
+    music_file: Optional[Path] = None,
+    music_volume: float = 0.15,
+) -> tuple[bool, str]:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    srt_esc = _escape_srt_path(str(srt_path.resolve()))
+    subtitle_filter = (
+        f"subtitles='{srt_esc}'"
+        f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
+    )
+
+    if music_file and music_file.is_file():
+        filter_complex = (
+            f"[0:v]{subtitle_filter}[vout];"
+            f"[1:a]volume={music_volume},apad[aout]"
+        )
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(capture_path),
+            "-i",
+            str(music_file),
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            "[vout]",
+            "-map",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-shortest",
+            "-pix_fmt",
+            "yuv420p",
+            str(output_path),
+        ]
+    else:
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(capture_path),
+            "-vf",
+            subtitle_filter,
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-pix_fmt",
+            "yuv420p",
+            str(output_path),
+        ]
+    return _run_ffmpeg(cmd)
+
+
+class RpaVideoSession:
+    """Service/RPA 层统一 ffmpeg 桌面录屏；OPENCLAW_RECORD_VIDEO=1 时启用成片流程。"""
+
+    def __init__(
+        self,
+        *,
+        skill_slug: str,
+        skill_data_dir: str,
+        batch_id: str,
+        title: str = "",
+    ) -> None:
+        self.skill_slug = skill_slug
+        self.skill_data_dir = os.path.abspath(skill_data_dir)
+        self.batch_id = batch_id
+        self.title = title
+        self.enabled = _record_video_enabled()
+        self.warnings: List[str] = []
+        self.output_video_path: Optional[str] = None
+        self.capture_path: Optional[str] = None
+        self.artifacts: Dict[str, Any] = {}
+
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        safe_batch = "".join(c if c.isalnum() or c in "-_" else "_" for c in batch_id)
+        self._video_stem = f"{skill_slug}_{ts}_{safe_batch}"
+
+        self._videos_dir = os.path.join(self.skill_data_dir, "videos")
+        self._artifact_root = os.path.join(
+            self.skill_data_dir, "rpa-artifacts", batch_id
+        )
+        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._subtitles_dir, exist_ok=True)
+            os.makedirs(self._logs_dir, exist_ok=True)
+            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:
+            self._t0 = time.monotonic()
+        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
+        if not shutil.which("ffmpeg"):
+            self.warnings.append("ffmpeg_not_found")
+            return
+
+        cmd = _build_desktop_record_cmd(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.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:
+        """停止录制后合成最终 MP4；失败只记 warning，不抛异常。"""
+        if not self.enabled:
+            self.artifacts = self.summary()
+            return
+
+        if not self._steps and self.title:
+            self.add_step(self.title)
+
+        try:
+            capture = Path(self._capture_path)
+            if not capture.is_file() or capture.stat().st_size <= 0:
+                self._append_warning("ffmpeg_capture_missing")
+                self.artifacts = self.summary()
+                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()
+            if music is None:
+                self.warnings.append("background_music_missing")
+
+            srt_file = Path(self._srt_path)
+            use_srt = srt_file.is_file() and srt_file.stat().st_size > 0
+
+            ok, err = _compose_capture_mp4(
+                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_capture_mp4(
+                    capture,
+                    srt_file if use_srt else Path(self._srt_path),
+                    Path(self._planned_output),
+                    music_file=None,
+                )
+            if not ok and not use_srt:
+                ok, err = _run_ffmpeg(
+                    [
+                        "ffmpeg",
+                        "-y",
+                        "-i",
+                        str(capture),
+                        "-c:v",
+                        "libx264",
+                        "-preset",
+                        "fast",
+                        "-crf",
+                        "23",
+                        "-pix_fmt",
+                        "yuv420p",
+                        str(self._planned_output),
+                    ]
+                )
+            if not ok:
+                self._append_warning("ffmpeg_compose_failed", err)
+                self.artifacts = self.summary()
+                return
+
+            self.output_video_path = str(Path(self._planned_output).resolve())
+        except Exception as exc:
+            self.warnings.append(f"video_finalize_error: {exc}")
+
+        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),
+        }
+
+    def to_artifact_dict(self) -> Dict[str, Any]:
+        data = self.summary()
+        data["batch_id"] = self.batch_id
+        return data

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")
-    os.makedirs(path, exist_ok=True)
+    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:
-            os.makedirs(parent, exist_ok=True)
+            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:
-        return
+    _clear_logger_handlers(log)
     log.setLevel(_log_level_from_env())
-    path = get_skill_log_file_path()
-    fh = TimedRotatingFileHandler(
-        path,
-        when="midnight",
-        interval=1,
-        backupCount=_backup_count(),
-        encoding="utf-8",
-        delay=True,
-    )
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
-    fh.setFormatter(fmt)
-    fh.addFilter(_SkillContextFilter())
-    log.addHandler(fh)
+    filt = _SkillContextFilter()
+
+    log_path = get_skill_log_file_path()
+    ok, _err = _can_open_log_file(log_path)
+    if ok:
+        try:
+            fh = TimedRotatingFileHandler(
+                log_path,
+                when="midnight",
+                interval=1,
+                backupCount=_backup_count(),
+                encoding="utf-8",
+                delay=False,
+            )
+            fh.setFormatter(fmt)
+            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",
     ):
-        sh = logging.StreamHandler(sys.stderr)
-        sh.setLevel(logging.WARNING)
-        sh.setFormatter(fmt)
-        sh.addFilter(_SkillContextFilter())
-        log.addHandler(sh)
+        try:
+            sh = logging.StreamHandler(sys.stderr)
+            sh.setLevel(logging.WARNING)
+            sh.setFormatter(fmt)
+            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)
-    fh.setFormatter(fmt)
-    fh.addFilter(_SkillContextFilter())
-    lg.addHandler(fh)
+    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(filt)
+            lg.addHandler(fh)
+        except Exception:
+            _attach_fallback_handler(lg)
+    else:
+        _attach_fallback_handler(lg)
     lg.propagate = False
     return lg

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

@@ -0,0 +1,147 @@
+# -*- coding: utf-8 -*-
+"""RpaVideoSession：ffmpeg 桌面录屏。"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from jiangchang_skill_core import config
+from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+
+
+class TestRpaVideoSession(unittest.TestCase):
+    def setUp(self) -> None:
+        config.reset_cache()
+
+    def test_disabled_is_noop(self) -> None:
+        old = os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+        try:
+            with tempfile.TemporaryDirectory() as tmp:
+                config.reset_cache()
+                session = RpaVideoSession(
+                    skill_slug="receive-order",
+                    skill_data_dir=tmp,
+                    batch_id="batch_test",
+                )
+                self.assertFalse(session.enabled)
+                summ = session.summary()
+                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_use_ffmpeg_capture(self) -> 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="exmail_20260602_120000",
+                    title="接收订单",
+                )
+                self.assertTrue(session.enabled)
+                self.assertTrue(
+                    session._capture_path.replace("\\", "/").endswith(
+                        "/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(
+                    session._planned_output.replace("\\", "/").startswith(
+                        tmp.replace("\\", "/") + "/videos/"
+                    )
+                )
+        finally:
+            os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
+            config.reset_cache()
+
+    @unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture")
+    @patch("jiangchang_skill_core.rpa.video_session.shutil.which", return_value="ffmpeg")
+    @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen")
+    def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, _which: MagicMock) -> None:
+        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.assertIn("ffmpeg", cmd)
+                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.shutil.which", return_value=None)
+    def test_ffmpeg_not_found_warning(self, _which: 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:
+                config.reset_cache()
+                session = RpaVideoSession(
+                    skill_slug="receive-order",
+                    skill_data_dir=tmp,
+                    batch_id="batch_warn",
+                )
+                asyncio.run(session.finalize())
+                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)
+            config.reset_cache()
+
+
+if __name__ == "__main__":
+    unittest.main()