feat(unified-render): Phase 2 - Feature Flag + RenderAdapter 适配层

- 新增 RENDER_ENGINE 配置项（legacy/unified），支持运行时切换渲染引擎
- 新建 RenderAdapter 适配层：EditPlan + EditPlanClips → ResolvedClip 转换，素材下载，进度回调
- compose_video 任务改造：根据 RENDER_ENGINE 配置选择旧引擎或新引擎
- 新引擎结果回写字段对齐（engine/width/height/file_size）
- 新增 16 个单元测试，validate_plan/render_plan/download_assets 全覆盖

apps/api/app/config.py

@@ -109,6 +109,9 @@ class Settings(BaseSettings):
     LOG_LEVEL: str = "INFO"
     CORS_ORIGINS_RAW: str = "http://localhost:3000,http://localhost:5173,http://localhost:8000"
 
+    # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
+    RENDER_ENGINE: str = "legacy"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/worker/video_processing/__init__.py

@@ -6,6 +6,7 @@
 from . import dedup_helpers, ffmpeg_utils, oss_helpers
 from .processor import VideoProcessor, VideoResult
 from .unified_render_service import RenderResult, UnifiedRenderService
+from .render_adapter import RenderAdapter, RenderAdapterResult
 
 __all__ = [
     "VideoProcessor",
@@ -15,4 +16,6 @@ __all__ = [
     "dedup_helpers",
     "UnifiedRenderService",
     "RenderResult",
+    "RenderAdapter",
+    "RenderAdapterResult",
 ]

apps/worker/video_processing/render_adapter.py

@@ -0,0 +1,289 @@
+"""统一渲染引擎适配层 — Phase 2.
+
+将 EditPlan + EditPlanClips（来自 DB）适配为 UnifiedRenderService 的输入格式，
+封装素材下载、渲染执行、结果上传的完整流程。
+
+职责：
+  1. 从 DB 读取 EditPlan + EditPlanClips
+  2. 下载素材到本地，构建 asset_path_map
+  3. 调用 UnifiedRenderService 执行渲染
+  4. 上传渲染结果到 OSS
+  5. 支持进度回调（对接 JobService）
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import (
+    SQLAlchemyEditPlanClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+    SQLAlchemyEditPlanRepository,
+)
+from packages.domain.edit_plan import EditPlan, EditPlanStatus
+from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+from video_processing.oss_helpers import download_asset, upload_to_oss
+from video_processing.unified_render_service import (
+    RenderResult,
+    UnifiedRenderService,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ── 数据结构 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class RenderAdapterResult:
+    """渲染适配结果。"""
+
+    success: bool
+    output_url: str = ""
+    output_path: Path | None = None
+    duration: float = 0.0
+    file_size: int = 0
+    width: int = 0
+    height: int = 0
+    clip_count: int = 0
+    error_message: str = ""
+
+
+ProgressCallback = Callable[[float, str], None]
+"""进度回调：(progress_0_100, stage_description) → None"""
+
+
+# ── 适配层主体 ────────────────────────────────────────────────────────────────
+
+
+class RenderAdapter:
+    """统一渲染引擎适配层。
+
+    桥接 EditPlan 领域模型与 UnifiedRenderService 图层模型。
+
+    用法::
+
+        adapter = RenderAdapter(db)
+        result = adapter.render_plan(
+            plan_id=plan_id,
+            job_id=job_id,
+            progress_cb=lambda p, s: job_service.update_progress(job_id, p, s),
+        )
+    """
+
+    def __init__(self, db: Session) -> None:
+        self._db = db
+        self._plan_repo = SQLAlchemyEditPlanRepository(db)
+        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
+
+    # ── 公开方法 ──────────────────────────────────────────────────────────
+
+    def render_plan(
+        self,
+        plan_id: str,
+        *,
+        job_id: str = "",
+        work_dir: Path | None = None,
+        progress_cb: ProgressCallback | None = None,
+    ) -> RenderAdapterResult:
+        """渲染一个 EditPlan。
+
+        完整流程：
+        1. 加载计划与片段
+        2. 下载素材
+        3. 执行统一渲染
+        4. 上传结果
+
+        Args:
+            plan_id: EditPlan ID
+            job_id: 关联的 Job ID（用于结果存储路径）
+            work_dir: 工作目录，不传则使用临时目录
+            progress_cb: 进度回调函数
+
+        Returns:
+            RenderAdapterResult
+        """
+        temp_dir = None
+        try:
+            # 0. 准备工作目录
+            if work_dir is None:
+                temp_dir = tempfile.mkdtemp(prefix="render_")
+                work_dir = Path(temp_dir)
+            work_dir.mkdir(parents=True, exist_ok=True)
+
+            self._report_progress(progress_cb, 5.0, "加载剪辑计划")
+
+            # 1. 加载计划与片段
+            plan = self._plan_repo.get(plan_id)
+            if plan is None:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message=f"剪辑计划不存在: {plan_id}",
+                )
+
+            clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+            ready_clips = [c for c in clips if c.status == EditPlanClipStatus.READY and c.asset_id]
+            ready_clips.sort(key=lambda c: c.order)
+
+            if not ready_clips:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="没有可渲染的就绪片段",
+                    clip_count=0,
+                )
+
+            logger.info(
+                "开始渲染: plan_id=%s job_id=%s ready_clips=%d",
+                plan_id,
+                job_id,
+                len(ready_clips),
+            )
+
+            self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）")
+
+            # 2. 下载素材
+            asset_path_map = self._download_assets(ready_clips, work_dir)
+            if not asset_path_map:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="所有素材下载失败",
+                    clip_count=len(ready_clips),
+                )
+
+            self._report_progress(progress_cb, 40.0, "执行视频渲染")
+
+            # 3. 执行统一渲染
+            render_svc = UnifiedRenderService(
+                plan=plan,
+                clips=ready_clips,
+                asset_path_map=asset_path_map,
+                work_dir=work_dir,
+            )
+            result = render_svc.render()
+
+            self._report_progress(progress_cb, 80.0, "上传渲染结果")
+
+            # 4. 上传结果
+            storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
+            output_url = upload_to_oss(result.output_path, storage_key)
+
+            self._report_progress(progress_cb, 100.0, "渲染完成")
+
+            return RenderAdapterResult(
+                success=True,
+                output_url=output_url or "",
+                output_path=result.output_path,
+                duration=result.duration,
+                file_size=result.file_size,
+                width=result.width,
+                height=result.height,
+                clip_count=len(ready_clips),
+            )
+
+        except Exception as exc:
+            logger.exception("渲染失败: plan_id=%s", plan_id)
+            return RenderAdapterResult(
+                success=False,
+                error_message=str(exc)[:500],
+            )
+        finally:
+            # 清理临时目录
+            if temp_dir:
+                import shutil
+
+                try:
+                    shutil.rmtree(temp_dir, ignore_errors=True)
+                except Exception:
+                    pass
+
+    def validate_plan(self, plan_id: str) -> tuple[bool, list[str], list[str], int, int]:
+        """校验计划是否可渲染（兼容 VideoComposeService.validate_compose 接口）。
+
+        Returns:
+            (valid, errors, warnings, ready_clip_count, total_clip_count)
+        """
+        errors: list[str] = []
+        warnings: list[str] = []
+
+        plan = self._plan_repo.get(plan_id)
+        if plan is None:
+            return False, [f"剪辑计划不存在: {plan_id}"], [], 0, 0
+
+        if plan.status not in (EditPlanStatus.EDITING, EditPlanStatus.RENDERING):
+            errors.append(f"计划状态不正确，需要 editing 或 rendering，当前: {plan.status}")
+
+        clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
+        if not clips:
+            errors.append("计划没有任何片段")
+            return False, errors, warnings, 0, 0
+
+        clips.sort(key=lambda c: c.order)
+
+        ready_count = 0
+        pending_count = 0
+        no_asset_count = 0
+
+        for clip in clips:
+            if clip.status == EditPlanClipStatus.READY:
+                ready_count += 1
+                if not clip.asset_id:
+                    errors.append(f"片段 {clip.id} (order={clip.order}) 没有分配素材")
+                    no_asset_count += 1
+            elif clip.status == EditPlanClipStatus.PENDING:
+                pending_count += 1
+            elif clip.status == EditPlanClipStatus.FAILED:
+                warnings.append(f"片段 {clip.id} (order={clip.order}) 状态为 failed，已跳过")
+
+        if ready_count == 0:
+            errors.append("没有就绪（ready）的片段可以合成")
+
+        if pending_count > 0:
+            warnings.append(f"有 {pending_count} 个片段仍处于 pending 状态")
+
+        return len(errors) == 0, errors, warnings, ready_count, len(clips)
+
+    # ── 内部方法 ──────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _report_progress(progress_cb: ProgressCallback | None, progress: float, stage: str) -> None:
+        """上报进度。"""
+        if progress_cb is not None:
+            try:
+                progress_cb(progress, stage)
+            except Exception:
+                logger.exception("进度回调失败")
+
+    @staticmethod
+    def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> dict[str, Path]:
+        """下载片段素材到本地，返回 asset_id → local_path 映射。
+
+        只保留下载成功的素材。
+        """
+        asset_dir = work_dir / "assets"
+        asset_dir.mkdir(exist_ok=True)
+
+        asset_path_map: dict[str, Path] = {}
+
+        for clip in clips:
+            asset_id = clip.asset_id
+            if not asset_id:
+                continue
+
+            # 生成安全的本地文件名
+            safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4"
+            local_path = asset_dir / safe_name
+
+            if download_asset(asset_id, local_path):
+                asset_path_map[asset_id] = local_path
+                logger.debug("素材下载成功: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
+            else:
+                logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
+
+        return asset_path_map

apps/worker/worker_app/core/config.py

@@ -18,6 +18,9 @@ class WorkerSettings(BaseSettings):
     environment: str = "development"
     auto_create_schema: bool = False
 
+    # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
+    render_engine: str = "legacy"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/worker/worker_app/tasks/compose_video.py

@@ -39,6 +39,10 @@ def _get_job_service():
 def compose_video(self, job_id: str, **kwargs):
     """视频合成任务。
 
+    根据 RENDER_ENGINE 配置选择渲染引擎：
+      - legacy: 旧 VideoComposeService（filter_complex 模式）
+      - unified: 新 UnifiedRenderService（图层架构）
+
     Args:
         job_id: JobService 中的任务 ID
         **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等）
@@ -56,66 +60,16 @@ def compose_video(self, job_id: str, **kwargs):
             job_service.fail_job(job_id, "Missing plan_id in job payload")
             return {"status": "error", "message": "Missing plan_id"}
 
-        # 标记为 running
-        job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
+        # 判断使用哪个渲染引擎
+        from worker_app.core.config import get_settings as get_worker_settings
 
-        # 延迟导入 VideoComposeService
-        from apps.api.app.services.video_compose_service import VideoComposeService
+        worker_settings = get_worker_settings()
+        engine = (worker_settings.render_engine or "legacy").lower()
 
-        compose_svc = VideoComposeService(db)
-
-        # 校验合成条件
-        job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
-        validation = compose_svc.validate_compose(plan_id)
-        if not validation.valid:
-            error_msg = "; ".join(validation.errors)
-            job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
-            return {"status": "error", "message": error_msg}
-
-        # 构建合成命令
-        job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
-        _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-        output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-        compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
-
-        # 执行 FFmpeg
-        job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
-        logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id)
-
-        try:
-            subprocess.run(
-                compose_cmd.command,
-                check=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-                timeout=3600,
-            )
-        except subprocess.CalledProcessError as e:
-            job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}")
-            raise
-
-        # 上传结果
-        job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
-        storage_key = f"rendered/{plan_id}/{job_id}.mp4"
-
-        from worker_app.tasks.edit_plan_generation import _upload_to_oss
-
-        output_url = _upload_to_oss(Path(output_path), storage_key)
-
-        # 更新 Job 状态为完成
-        result_data = {
-            "plan_id": plan_id,
-            "output_path": output_path,
-            "storage_key": storage_key,
-            "output_url": output_url or "",
-            "estimated_duration": compose_cmd.estimated_duration,
-            "clip_count": len(compose_cmd.clip_chains),
-        }
-        job_service.complete_job(job_id, result=result_data)
-
-        logger.info("视频合成完成: job_id=%s, plan_id=%s", job_id, plan_id)
-        return {"status": "completed", "job_id": job_id, "result": result_data}
+        if engine == "unified":
+            return _compose_with_unified_engine(self, job_service, job, plan_id, db)
+        else:
+            return _compose_with_legacy_engine(self, job_service, job, plan_id, db)
 
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
@@ -129,11 +83,140 @@ def compose_video(self, job_id: str, **kwargs):
         raise self.retry(exc=exc, countdown=60)
     finally:
         db.close()
-        # 清理临时文件
+
+
+def _compose_with_legacy_engine(task, job_service, job, plan_id: str, db) -> dict:
+    """旧引擎渲染路径（VideoComposeService）。"""
+    job_id = job.id
+
+    # 标记为 running
+    job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
+
+    # 延迟导入 VideoComposeService
+    from apps.api.app.services.video_compose_service import VideoComposeService
+
+    compose_svc = VideoComposeService(db)
+
+    # 校验合成条件
+    job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
+    validation = compose_svc.validate_compose(plan_id)
+    if not validation.valid:
+        error_msg = "; ".join(validation.errors)
+        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 构建合成命令
+    job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
+    _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
+    output_path = os.path.join(_output_dir, f"{job_id}.mp4")
+    compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
+
+    # 执行 FFmpeg
+    job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
+    logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id)
+
+    try:
+        subprocess.run(
+            compose_cmd.command,
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=3600,
+        )
+    except subprocess.CalledProcessError as e:
+        job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}")
+        raise
+
+    # 上传结果
+    job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
+    storage_key = f"rendered/{plan_id}/{job_id}.mp4"
+
+    from worker_app.tasks.edit_plan_generation import _upload_to_oss
+
+    output_url = _upload_to_oss(Path(output_path), storage_key)
+
+    # 更新 Job 状态为完成
+    result_data = {
+        "plan_id": plan_id,
+        "output_path": output_path,
+        "storage_key": storage_key,
+        "output_url": output_url or "",
+        "estimated_duration": compose_cmd.estimated_duration,
+        "clip_count": len(compose_cmd.clip_chains),
+        "engine": "legacy",
+    }
+    job_service.complete_job(job_id, result=result_data)
+
+    logger.info("视频合成完成(legacy): job_id=%s, plan_id=%s", job_id, plan_id)
+    return {"status": "completed", "job_id": job_id, "result": result_data}
+
+
+def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:
+    """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。"""
+    job_id = job.id
+
+    # 标记为 running
+    job_service.update_progress(job_id, progress=10.0, current_stage="初始化统一渲染引擎")
+
+    from video_processing.render_adapter import RenderAdapter
+
+    adapter = RenderAdapter(db)
+
+    # 校验合成条件
+    job_service.update_progress(job_id, progress=15.0, current_stage="校验合成条件")
+    valid, errors, warnings, ready_count, total_count = adapter.validate_plan(plan_id)
+    if not valid:
+        error_msg = "; ".join(errors)
+        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 进度回调
+    def progress_cb(progress: float, stage: str) -> None:
         try:
-            _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-            output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-            if Path(output_path).exists():
-                Path(output_path).unlink()
-        except Exception as e:
-            logger.warning(f"Operation failed in apps/worker/worker_app/tasks/compose_video.py: {e}", exc_info=True)
+            job_service.update_progress(job_id, progress=progress, current_stage=stage)
+        except Exception:
+            logger.exception("更新进度失败")
+
+    # 执行渲染
+    job_service.update_progress(job_id, progress=20.0, current_stage="开始渲染")
+    logger.info("统一渲染引擎开始: job_id=%s plan_id=%s", job_id, plan_id)
+
+    result = adapter.render_plan(
+        plan_id=plan_id,
+        job_id=job_id,
+        progress_cb=progress_cb,
+    )
+
+    if not result.success:
+        job_service.fail_job(job_id, f"渲染失败: {result.error_message}")
+        raise RuntimeError(result.error_message)
+
+    # 更新 Job 状态为完成
+    result_data = {
+        "plan_id": plan_id,
+        "output_path": str(result.output_path) if result.output_path else "",
+        "storage_key": f"rendered/{plan_id}/{job_id}.mp4",
+        "output_url": result.output_url,
+        "estimated_duration": result.duration,
+        "clip_count": result.clip_count,
+        "engine": "unified",
+        "width": result.width,
+        "height": result.height,
+        "file_size": result.file_size,
+    }
+    job_service.complete_job(job_id, result=result_data)
+
+    logger.info("视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", job_id, plan_id, result.duration)
+    return {"status": "completed", "job_id": job_id, "result": result_data}
+
+
+def _cleanup_output(job_id: str) -> None:
+    """清理临时输出文件。"""
+    try:
+        _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
+        output_path = os.path.join(_output_dir, f"{job_id}.mp4")
+        if Path(output_path).exists():
+            Path(output_path).unlink()
+    except Exception as e:
+        logger.warning(f"清理输出文件失败: {e}", exc_info=True)

tests/unit/test_render_adapter.py

@@ -0,0 +1,423 @@
+"""RenderAdapter 单元测试 — Phase 2.
+
+测试适配层的计划加载、素材下载、引擎调用、结果上传等逻辑。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.render_adapter import RenderAdapter, RenderAdapterResult
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeClip:
+    """模拟 EditPlanClip。"""
+
+    id: str
+    plan_id: str = "plan_001"
+    clip_type: str = "main"
+    order: int = 0
+    asset_id: str = ""
+    text_content: str = ""
+    start_time: float = 0.0
+    duration: float = 0.0
+    transition_effect: str = "cut"
+    status: str = "ready"
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class FakePlan:
+    """模拟 EditPlan。"""
+
+    id: str = "plan_001"
+    name: str = "测试计划"
+    status: str = "editing"
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+def _make_clip(
+    clip_id: str,
+    clip_type: str = "main",
+    order: int = 0,
+    asset_id: str | None = None,
+    duration: float = 5.0,
+    status: str = "ready",
+    transition_effect: str = "cut",
+    config: dict[str, Any] | None = None,
+) -> FakeClip:
+    # asset_id 为 None 时生成默认值，为空字符串时保留空串
+    if asset_id is None:
+        asset_id = f"asset_{clip_id}.mp4"
+    return FakeClip(
+        id=clip_id,
+        clip_type=clip_type,
+        order=order,
+        asset_id=asset_id,
+        duration=duration,
+        status=status,
+        transition_effect=transition_effect,
+        config=config or {},
+    )
+
+
+def _make_adapter(
+    plan: FakePlan | None = None,
+    clips: list[FakeClip] | None = None,
+) -> tuple[RenderAdapter, MagicMock, MagicMock]:
+    """创建测试用的 RenderAdapter 及 mock repo。
+
+    Returns:
+        (adapter, mock_plan_repo, mock_clip_repo)
+    """
+    mock_db = MagicMock()
+    adapter = RenderAdapter(mock_db)
+
+    # 替换内部 repo
+    mock_plan_repo = MagicMock()
+    mock_clip_repo = MagicMock()
+    adapter._plan_repo = mock_plan_repo
+    adapter._clip_repo = mock_clip_repo
+
+    # 设置默认返回
+    if plan is not None:
+        mock_plan_repo.get.return_value = plan
+    if clips is not None:
+        mock_clip_repo.list_by_plan.return_value = clips
+
+    return adapter, mock_plan_repo, mock_clip_repo
+
+
+# ── validate_plan 测试 ───────────────────────────────────────────────────────
+
+
+class TestValidatePlan:
+    def test_plan_not_found(self):
+        """计划不存在时校验失败。"""
+        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
+        mock_plan_repo.get.return_value = None
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert len(errors) == 1
+        assert "不存在" in errors[0]
+        assert ready_count == 0
+        assert total_count == 0
+
+    def test_no_clips(self):
+        """没有任何片段时校验失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        adapter, _, mock_clip_repo = _make_adapter(plan=plan, clips=[])
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有任何片段" in e for e in errors)
+
+    def test_no_ready_clips(self):
+        """没有 ready 片段时校验失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", status="pending"),
+            _make_clip("c2", status="pending"),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有就绪" in e for e in errors)
+        assert ready_count == 0
+        assert total_count == 2
+
+    def test_ready_clip_no_asset(self):
+        """ready 片段没有 asset_id 时报错。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", asset_id=""),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("没有分配素材" in e for e in errors)
+
+    def test_valid_plan(self):
+        """正常计划校验通过。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, duration=3.0),
+            _make_clip("c2", order=1, duration=4.0),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert valid
+        assert len(errors) == 0
+        assert ready_count == 2
+        assert total_count == 2
+
+    def test_wrong_status(self):
+        """计划状态不正确时报错。"""
+        plan = FakePlan(id="plan_001", status="draft")
+        clips = [_make_clip("c1")]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, _, _, _ = adapter.validate_plan("plan_001")
+
+        assert not valid
+        assert any("状态不正确" in e for e in errors)
+
+    def test_mixed_status_with_warnings(self):
+        """混合状态时有 pending/failed 警告。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, status="ready"),
+            _make_clip("c2", order=1, status="pending"),
+            _make_clip("c3", order=2, status="failed"),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001")
+
+        assert valid
+        assert any("pending" in w for w in warnings)
+        assert any("failed" in w for w in warnings)
+        assert ready_count == 1
+        assert total_count == 3
+
+
+# ── render_plan 测试 ─────────────────────────────────────────────────────────
+
+
+class TestRenderPlan:
+    def test_plan_not_found(self):
+        """计划不存在时返回失败。"""
+        adapter, mock_plan_repo, _ = _make_adapter(plan=None)
+        mock_plan_repo.get.return_value = None
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "不存在" in result.error_message
+
+    def test_no_ready_clips(self):
+        """没有 ready 片段时返回失败。"""
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [_make_clip("c1", status="pending")]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "没有可渲染" in result.error_message
+        assert result.clip_count == 0
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_assets_download_fail(self, mock_download):
+        """所有素材下载失败时返回失败。"""
+        mock_download.return_value = False
+
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan("plan_001")
+
+        assert not result.success
+        assert "素材下载失败" in result.error_message
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    def test_successful_render(self, mock_download, mock_render_cls, mock_upload, tmp_path):
+        """完整渲染流程成功。"""
+        # 素材下载成功
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video data")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        # 渲染成功
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "output.mp4",
+            duration=10.0,
+            file_size=102400,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+
+        # 上传成功
+        mock_upload.return_value = "https://oss.example.com/rendered/plan_001/job_001.mp4"
+
+        plan = FakePlan(id="plan_001", status="editing")
+        clips = [
+            _make_clip("c1", order=0, duration=5.0),
+            _make_clip("c2", order=1, duration=5.0),
+        ]
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan(
+            "plan_001",
+            job_id="job_001",
+            work_dir=tmp_path / "work",
+        )
+
+        assert result.success
+        assert result.output_url.startswith("https://")
+        assert result.duration == 10.0
+        assert result.width == 1280
+        assert result.height == 720
+        assert result.clip_count == 2
+
+        # 验证 UnifiedRenderService 被正确调用
+        mock_render_cls.assert_called_once()
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["plan"] is plan
+        assert len(call_kwargs.kwargs["clips"]) == 2
+        assert len(call_kwargs.kwargs["asset_path_map"]) == 2
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_progress_callback(self, mock_download, tmp_path):
+        """进度回调被正确触发。"""
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake data")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        # 模拟渲染异常，避免走到最后
+        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
+            mock_render = MagicMock()
+            mock_render.render.side_effect = RuntimeError("render error")
+            mock_render_cls.return_value = mock_render
+
+            plan = FakePlan(id="plan_001", status="editing")
+            clips = [_make_clip("c1", order=0, duration=5.0)]
+            adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+            progress_values = []
+
+            def progress_cb(progress: float, stage: str) -> None:
+                progress_values.append((progress, stage))
+
+            result = adapter.render_plan(
+                "plan_001",
+                work_dir=tmp_path / "work",
+                progress_cb=progress_cb,
+            )
+
+            # 即使渲染失败，前期进度也应该上报了
+            assert len(progress_values) > 0
+            # 第一个进度应该是加载计划
+            assert progress_values[0][1] == "加载剪辑计划"
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_partial_asset_download(self, mock_download, tmp_path):
+        """部分素材下载失败时，只使用成功的素材。"""
+        download_results = [True, False, True]  # 3个素材中2个成功
+
+        def _fake_download(asset_id, local_path):
+            idx = hash(asset_id) % 3
+            if download_results[idx]:
+                local_path.parent.mkdir(parents=True, exist_ok=True)
+                local_path.write_bytes(b"fake data")
+                return True
+            return False
+
+        mock_download.side_effect = _fake_download
+
+        with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls:
+            mock_render = MagicMock()
+            mock_render.render.return_value = MagicMock(
+                output_path=tmp_path / "out.mp4",
+                duration=5.0,
+                file_size=1024,
+                width=1280,
+                height=720,
+            )
+            mock_render_cls.return_value = mock_render
+
+            with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"):
+                plan = FakePlan(id="plan_001", status="editing")
+                clips = [
+                    _make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"),
+                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"),
+                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"),
+                ]
+                adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+                result = adapter.render_plan(
+                    "plan_001",
+                    work_dir=tmp_path / "work",
+                )
+
+                # 至少有部分素材成功，渲染应该进行
+                # （具体成功数量取决于 hash 结果，但至少1个成功就能渲染）
+                assert result.success or "素材下载失败" in result.error_message
+
+
+# ── _download_assets 测试 ────────────────────────────────────────────────────
+
+
+class TestDownloadAssets:
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_download_success(self, mock_download, tmp_path):
+        """全部素材下载成功。"""
+        mock_download.return_value = True
+
+        clips = [
+            _make_clip("c1", order=0, asset_id="key1.mp4"),
+            _make_clip("c2", order=1, asset_id="key2.mp4"),
+        ]
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 2
+        assert "key1.mp4" in result
+        assert "key2.mp4" in result
+        assert mock_download.call_count == 2
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_empty_asset_id_skipped(self, mock_download, tmp_path):
+        """空 asset_id 的片段被跳过。"""
+        clips = [
+            _make_clip("c1", order=0, asset_id=""),
+            _make_clip("c2", order=1, asset_id="key2.mp4"),
+        ]
+        mock_download.return_value = True
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 1
+        assert "key2.mp4" in result
+        assert mock_download.call_count == 1  # 只调用了一次下载
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_all_download_fail(self, mock_download, tmp_path):
+        """全部下载失败返回空字典。"""
+        mock_download.return_value = False
+
+        clips = [
+            _make_clip("c1", order=0, asset_id="key1.mp4"),
+        ]
+
+        result = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(result) == 0