apps/worker/video_processing/render_adapter.py

@@ -186,85 +186,15 @@ class RenderAdapter:
 
             self._report_progress(progress_cb, 35.0, "准备 BGM 音频")
 
-            # 3. 准备 BGM（从 plan.config.bgm 读取配置）
-            bgm_path = self._prepare_bgm(plan, work_dir, plan_id)
-
-            self._report_progress(progress_cb, 40.0, "执行视频渲染")
-
-            # 4. 初始化 ASR 服务（用于自动字幕）
-            asr_service = self._get_asr_service()
-
-            # 5. 从 plan.config.export 读取输出分辨率
-            plan_config = plan.config or {}
-            export_config = plan_config.get("export", {}) or {}
-            output_width, output_height = _parse_resolution(export_config.get("resolution"))
-            logger.info(
-                "渲染输出分辨率: plan_id=%s resolution=%dx%d source=%s",
-                plan_id,
-                output_width,
-                output_height,
-                "config" if export_config.get("resolution") else "default",
-            )
-
-            # 6. 执行统一渲染
-            render_svc = UnifiedRenderService(
+            # 3~6. 统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）
+            return self._do_render(
                 plan=plan,
                 clips=ready_clips,
                 asset_path_map=asset_path_map,
                 work_dir=work_dir,
-                output_width=output_width,
-                output_height=output_height,
-                bgm_path=bgm_path,
-                asr_service=asr_service,
-            )
-            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, 90.0, "生成封面缩略图")
-
-            # 5. 生成缩略图（在清理临时目录前）
-            thumbnail_url = ""
-            try:
-                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
-
-                thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
-                thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
-            except Exception as thumb_err:
-                logger.warning(
-                    "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
-                    plan_id,
-                    thumb_err,
-                )
-
-            self._report_progress(progress_cb, 100.0, "渲染完成")
-
-            logger.info(
-                "[render-adapter] render success: plan_id=%s job_id=%s engine=unified "
-                "duration=%.2fs file_size=%d resolution=%dx%d clip_count=%d",
-                plan_id,
-                job_id,
-                result.duration,
-                result.file_size,
-                result.width,
-                result.height,
-                len(ready_clips),
-            )
-
-            return RenderAdapterResult(
-                success=True,
-                output_url=output_url or "",
-                output_path=result.output_path,
-                thumbnail_url=thumbnail_url,
-                duration=result.duration,
-                file_size=result.file_size,
-                width=result.width,
-                height=result.height,
-                clip_count=len(ready_clips),
+                plan_id=plan_id,
+                job_id=job_id,
+                progress_cb=progress_cb,
                 rendered_clip_ids=rendered_clip_ids,
                 failed_clip_ids=failed_clip_ids,
             )
@@ -540,3 +470,215 @@ class RenderAdapter:
         except Exception as e:
             logger.warning("ASR 服务初始化失败，自动字幕将不可用: %s", e)
             return None
+
+    def _do_render(
+        self,
+        plan: Any,
+        clips: list[Any],
+        asset_path_map: dict[str, Path],
+        work_dir: Path,
+        *,
+        plan_id: str,
+        job_id: str = "",
+        progress_cb: ProgressCallback | None = None,
+        rendered_clip_ids: list[str] | None = None,
+        failed_clip_ids: list[str] | None = None,
+    ) -> RenderAdapterResult:
+        """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。
+
+        render_plan 和 render_from_memory 共用此方法。
+
+        Args:
+            rendered_clip_ids: 成功下载/准备的 clip id 列表（render_plan 从下载阶段传入）
+            failed_clip_ids: 失败的 clip id 列表
+
+        Returns:
+            RenderAdapterResult
+        """
+        # 1. 准备 BGM
+        bgm_path = self._prepare_bgm(plan, work_dir, plan_id)
+
+        self._report_progress(progress_cb, 40.0, "执行视频渲染")
+
+        # 2. 初始化 ASR
+        asr_service = self._get_asr_service()
+
+        # 3. 读取输出分辨率
+        plan_config = plan.config or {}
+        export_config = plan_config.get("export", {}) or {}
+        output_width, output_height = _parse_resolution(export_config.get("resolution"))
+        logger.info(
+            "渲染输出分辨率: plan_id=%s resolution=%dx%d source=%s",
+            plan_id,
+            output_width,
+            output_height,
+            "config" if export_config.get("resolution") else "default",
+        )
+
+        # 4. 执行统一渲染
+        render_svc = UnifiedRenderService(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_path_map,
+            work_dir=work_dir,
+            output_width=output_width,
+            output_height=output_height,
+            bgm_path=bgm_path,
+            asr_service=asr_service,
+        )
+        result = render_svc.render()
+
+        self._report_progress(progress_cb, 80.0, "上传渲染结果")
+
+        # 5. 上传结果
+        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, 90.0, "生成封面缩略图")
+
+        # 6. 生成缩略图
+        thumbnail_url = ""
+        try:
+            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+
+            thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
+            thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
+        except Exception as thumb_err:
+            logger.warning(
+                "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
+                plan_id,
+                thumb_err,
+            )
+
+        self._report_progress(progress_cb, 100.0, "渲染完成")
+
+        logger.info(
+            "[render-adapter] render success: plan_id=%s job_id=%s engine=unified "
+            "duration=%.2fs file_size=%d resolution=%dx%d clip_count=%d",
+            plan_id,
+            job_id,
+            result.duration,
+            result.file_size,
+            result.width,
+            result.height,
+            len(clips),
+        )
+
+        final_rendered_ids = (
+            rendered_clip_ids if rendered_clip_ids is not None else [c.id for c in clips if hasattr(c, "id")]
+        )
+        final_failed_ids = failed_clip_ids if failed_clip_ids is not None else []
+
+        return RenderAdapterResult(
+            success=True,
+            output_url=output_url or "",
+            output_path=result.output_path,
+            thumbnail_url=thumbnail_url,
+            duration=result.duration,
+            file_size=result.file_size,
+            width=result.width,
+            height=result.height,
+            clip_count=len(clips),
+            rendered_clip_ids=final_rendered_ids,
+            failed_clip_ids=final_failed_ids,
+        )
+
+    def render_from_memory(
+        self,
+        plan: Any,
+        clips: list[Any],
+        asset_path_map: dict[str, Path],
+        *,
+        plan_id: str = "",
+        job_id: str = "",
+        work_dir: Path | None = None,
+        progress_cb: ProgressCallback | None = None,
+    ) -> RenderAdapterResult:
+        """使用内存中的 plan/clips/asset_path_map 直接渲染。
+
+        适用于一键生成等不写DB剪辑计划的场景，复用统一的 BGM/ASR/分辨率/渲染/缩略图逻辑。
+
+        Args:
+            plan: 类 EditPlan 的对象（鸭子类型，需有 id/config 等属性）
+            clips: 类 EditPlanClip 的对象列表
+            asset_path_map: asset_id → local_path 映射
+            plan_id: 用于日志的计划标识（不传则用 plan.id）
+            job_id: 关联的 Job ID
+            work_dir: 工作目录，不传则用临时目录
+            progress_cb: 进度回调
+
+        Returns:
+            RenderAdapterResult
+        """
+        actual_plan_id = plan_id or getattr(plan, "id", "memory_plan")
+        temp_dir = None
+        try:
+            if work_dir is None:
+                temp_dir = tempfile.mkdtemp(prefix="render_mem_")
+                work_dir = Path(temp_dir)
+            work_dir.mkdir(parents=True, exist_ok=True)
+
+            if not clips:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="没有可渲染的片段",
+                    clip_count=0,
+                )
+
+            if not asset_path_map:
+                return RenderAdapterResult(
+                    success=False,
+                    error_message="素材路径映射为空",
+                    clip_count=len(clips),
+                )
+
+            logger.info(
+                "开始内存模式渲染: plan_id=%s job_id=%s clip_count=%d engine=unified",
+                actual_plan_id,
+                job_id,
+                len(clips),
+            )
+
+            self._report_progress(progress_cb, 35.0, "准备 BGM 音频")
+
+            return self._do_render(
+                plan=plan,
+                clips=clips,
+                asset_path_map=asset_path_map,
+                work_dir=work_dir,
+                plan_id=actual_plan_id,
+                job_id=job_id,
+                progress_cb=progress_cb,
+            )
+
+        except subprocess.CalledProcessError as exc:
+            stderr_text = (exc.stderr or "").strip()
+            logger.error(
+                "[render-adapter] 内存模式渲染失败: plan_id=%s exit_code=%d\nstderr:\n%s",
+                actual_plan_id,
+                exc.returncode,
+                stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
+            )
+            return RenderAdapterResult(
+                success=False,
+                error_message=f"FFmpeg渲染失败(exit={exc.returncode}): {stderr_text[:200]}",
+                error_detail=stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
+            )
+        except Exception as exc:
+            logger.exception(
+                "[render-adapter] 内存模式渲染失败: plan_id=%s error=%s",
+                actual_plan_id,
+                str(exc)[:200],
+            )
+            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 as cleanup_err:
+                    logger.warning("临时目录清理失败: path=%s error=%s", temp_dir, cleanup_err)

apps/worker/worker_app/tasks/generation.py

@@ -138,7 +138,6 @@ def _flush_logs(task_id: str, gen_task) -> None:
 
 # ── 共享工具模块导入 ──────────────────────────────────────────────────────────
 
-from services.asr_service_factory import get_asr_service
 from video_processing.dedup_helpers import create_video_record_and_dedup
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 from video_processing.oss_helpers import (
@@ -146,7 +145,6 @@ from video_processing.oss_helpers import (
     get_signed_download_url,
     upload_to_oss,
 )
-from video_processing.unified_render_service import UnifiedRenderService
 
 # ── 虚拟 Plan / Clip（内存中构建，不写数据库） ────────────────────────────────
 
@@ -943,13 +941,31 @@ def _download_library_assets(
 def _validate_template_exists(template_id: str) -> None:
     """校验 template_id 是否存在且可用。
 
+    优先读新模板系统（EditTemplate），找不到 fallback 到旧模板系统（TemplateModel）。
+
     Raises:
         ValueError: template_id 不存在或已禁用时抛出
     """
-    from packages.adapters.sqlalchemy_impl.models import TemplateModel
+    from packages.adapters.sqlalchemy_impl import (
+        SQLAlchemyEditTemplateRepository,
+    )
 
     session = SessionLocal()
     try:
+        # 优先读新模板系统
+        new_repo = SQLAlchemyEditTemplateRepository(session)
+        new_template = new_repo.get(template_id)
+        if new_template is not None:
+            status_val = new_template.status.value if hasattr(new_template.status, "value") else new_template.status
+            if status_val == "active":
+                logger.info("模板校验通过(新系统): template_id=%s name=%s", template_id, new_template.name)
+                return
+            else:
+                raise ValueError(f"模板已停用: template_id={template_id}")
+
+        # fallback: 旧模板系统
+        from packages.adapters.sqlalchemy_impl.models import TemplateModel
+
         template = (
             session.query(TemplateModel)
             .filter(
@@ -958,9 +974,11 @@ def _validate_template_exists(template_id: str) -> None:
             )
             .first()
         )
-        if template is None:
-            raise ValueError(f"模板不存在或已禁用: template_id={template_id}")
-        logger.info("模板校验通过: template_id=%s name=%s", template_id, template.name)
+        if template:
+            logger.info("模板校验通过(旧系统): template_id=%s name=%s", template_id, template.name)
+            return
+
+        raise ValueError(f"模板不存在或已禁用: template_id={template_id}")
     finally:
         session.close()
 
@@ -968,17 +986,51 @@ def _validate_template_exists(template_id: str) -> None:
 def _load_template_plan_config(template_id: str) -> dict:
     """从模板加载 plan 级配置（BGM、字幕、标题等效果层）。
 
-    TemplateModel 里 bgm_config / subtitle_config / title_config 是独立字段，
-    需要组装成 plan.config 的格式（{bgm, subtitle, title}）后再注入。
+    优先读新模板系统（EditTemplate.config + TemplateClipConfig），
+    找不到 fallback 到旧模板系统（TemplateModel 独立字段）。
+
     模板不存在时返回空 dict，不阻塞主流程。
     """
     if not template_id:
         return {}
     try:
-        from packages.adapters.sqlalchemy_impl.models import TemplateModel
+        from packages.adapters.sqlalchemy_impl import (
+            SQLAlchemyEditTemplateRepository,
+            SQLAlchemyTemplateClipConfigRepository,
+        )
 
         session = SessionLocal()
         try:
+            # 优先读新模板系统
+            tpl_repo = SQLAlchemyEditTemplateRepository(session)
+            clip_repo = SQLAlchemyTemplateClipConfigRepository(session)
+            template = tpl_repo.get(template_id)
+
+            if template is not None:
+                # 新系统：config 直接就是 plan.config 格式
+                plan_config = dict(template.config or {})
+
+                # 从片段配置中提取 intro/outro 配置
+                clip_configs = clip_repo.list_by_template(template_id, limit=200)
+                if clip_configs:
+                    intro_outro = _extract_intro_outro_from_clip_configs(clip_configs)
+                    if intro_outro:
+                        plan_config["intro_outro"] = intro_outro
+
+                # 把 editing_mode 也带过去
+                if template.editing_mode:
+                    plan_config["editing_mode"] = template.editing_mode
+
+                logger.info(
+                    "模板配置加载成功(新系统): template_id=%s keys=%s",
+                    template_id,
+                    list(plan_config.keys()),
+                )
+                return plan_config
+
+            # fallback: 旧模板系统
+            from packages.adapters.sqlalchemy_impl.models import TemplateModel
+
             template = (
                 session.query(TemplateModel)
                 .filter(
@@ -1005,7 +1057,7 @@ def _load_template_plan_config(template_id: str) -> dict:
                 plan_config["bgm"] = bgm_cfg
 
             logger.info(
-                "模板配置加载成功: template_id=%s keys=%s",
+                "模板配置加载成功(旧系统): template_id=%s keys=%s",
                 template_id,
                 list(plan_config.keys()),
             )
@@ -1108,13 +1160,15 @@ def _render_video(
 ) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
 
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+
     Returns:
         (output_path, render_duration)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
 
-    # 构建虚拟 plan + clips
+    # 构建虚拟 plan + clips + asset_path_map
     virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task(
         task_id=task_id,
         downloaded_paths=downloaded_videos,
@@ -1126,7 +1180,6 @@ def _render_video(
     if template_id:
         template_config = _load_template_plan_config(template_id)
         if template_config:
-            # 合并：现有 config 优先级更高（目前为空，模板配置直接生效）
             base_config = virtual_plan.config or {}
             virtual_plan.config = {**template_config, **base_config}
             logger.info(
@@ -1135,6 +1188,16 @@ def _render_video(
                 list(template_config.keys()),
             )
 
+    # 确保输出分辨率配置存在（一键生成默认横屏 1280x720）
+    # RenderAdapter 从 plan.config.export.resolution 读取，
+    # 如果模板没有配置则用默认值，这里显式设置保持和旧逻辑一致
+    plan_cfg = virtual_plan.config or {}
+    export_cfg = plan_cfg.get("export", {}) or {}
+    if not export_cfg.get("resolution"):
+        export_cfg["resolution"] = f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}"
+        plan_cfg["export"] = export_cfg
+        virtual_plan.config = plan_cfg
+
     total_duration = sum(c.duration for c in virtual_clips)
     logger.info(
         "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs",
@@ -1144,49 +1207,41 @@ def _render_video(
     )
 
     render_start = time.monotonic()
-    render_output_path = temp_path / f"rendered-{task_id}.mp4"
+    logger.info("[task_id=%s] [渲染] RenderAdapter 统一渲染开始", task_id)
 
-    logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
+    # 使用 RenderAdapter 统一渲染入口（复用 BGM/ASR/分辨率/缩略图逻辑）
+    from video_processing.render_adapter import RenderAdapter
+    from worker_app.db import SessionLocal
 
-    # ── 准备 BGM 音频 ──
-    bgm_path: str | None = None
-    plan_config = virtual_plan.config or {}
-    bgm_config = plan_config.get("bgm", {}) or {}
-    if bgm_config.get("enabled", False):
-        try:
-            bgm_path = _prepare_bgm_track(
-                bgm_config=bgm_config,
-                temp_path=temp_path,
-                task_id=task_id,
-            )
-        except Exception as bgm_err:
-            logger.warning("[task_id=%s] [BGM] 准备失败，跳过BGM: %s", task_id, bgm_err)
-            bgm_path = None
+    db = SessionLocal()
+    try:
+        adapter = RenderAdapter(db)
+        render_result = adapter.render_from_memory(
+            plan=virtual_plan,
+            clips=virtual_clips,
+            asset_path_map=asset_path_map,
+            plan_id=f"gen_{task_id}",
+            job_id=task_id,
+            work_dir=temp_path,
+        )
+    finally:
+        db.close()
+
+    if not render_result.success:
+        raise RuntimeError(f"渲染失败: {render_result.error_message}")
 
-    render_service = UnifiedRenderService(
-        plan=virtual_plan,
-        clips=virtual_clips,
-        asset_path_map=asset_path_map,
-        work_dir=temp_path,
-        output_width=OUTPUT_WIDTH,
-        output_height=OUTPUT_HEIGHT,
-        output_fps=int(OUTPUT_FPS),
-        asr_service=get_asr_service(),
-        bgm_path=bgm_path,
-    )
-    render_result = render_service.render()
     render_output_path = render_result.output_path
     render_duration = render_result.duration
 
     render_elapsed = time.monotonic() - render_start
     logger.info(
-        "[task_id=%s] [渲染] unified 引擎完成: 耗时=%.1fs, 时长=%.2fs",
+        "[task_id=%s] [渲染] RenderAdapter 完成: 耗时=%.1fs, 时长=%.2fs",
         task_id,
         render_elapsed,
         render_duration,
     )
 
-    # 配音混音
+    # 配音混音（素材库音频，后处理混音）
     if voice_path:
         final_path = temp_path / f"final-{task_id}.mp4"
         try:

tests/unit/test_render_adapter.py

@@ -1151,3 +1151,213 @@ class TestRenderPlanWithBgmAsr:
         assert "c_bad" in result.failed_clip_ids
         assert len(result.rendered_clip_ids) == 2
         assert len(result.failed_clip_ids) == 1
+
+
+# ── render_from_memory 测试 ──────────────────────────────────────────────────
+
+
+class TestRenderFromMemory:
+    """render_from_memory 内存模式渲染测试。"""
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_successful_render(self, mock_render_cls, mock_upload, tmp_path):
+        """内存模式渲染成功。"""
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=10.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        plan = FakePlan(id="mem_plan_001", config={"bgm": {"enabled": False}})
+        clips = [
+            _make_clip("c1", order=0, duration=5.0),
+            _make_clip("c2", order=1, duration=5.0),
+        ]
+        asset_path_map = {
+            "asset_c1.mp4": tmp_path / "c1.mp4",
+            "asset_c2.mp4": tmp_path / "c2.mp4",
+        }
+        # 创建假文件
+        for p in asset_path_map.values():
+            p.write_bytes(b"fake")
+
+        adapter, _, _ = _make_adapter()
+
+        result = adapter.render_from_memory(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_path_map,
+            plan_id="mem_plan_001",
+            work_dir=tmp_path / "work",
+        )
+
+        assert result.success
+        assert result.duration == 10.0
+        assert result.file_size == 1024
+        assert result.width == 1280
+        assert result.height == 720
+        assert result.clip_count == 2
+        assert len(result.rendered_clip_ids) == 2
+        assert len(result.failed_clip_ids) == 0
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_empty_clips_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
+        """clips 为空时返回失败。"""
+        plan = FakePlan(id="mem_empty")
+        adapter, _, _ = _make_adapter()
+
+        result = adapter.render_from_memory(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            plan_id="mem_empty",
+            work_dir=tmp_path / "work",
+        )
+
+        assert not result.success
+        assert "没有可渲染的片段" in result.error_message
+        mock_render_cls.assert_not_called()
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_empty_asset_map_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
+        """asset_path_map 为空时返回失败。"""
+        plan = FakePlan(id="mem_no_assets")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        adapter, _, _ = _make_adapter()
+
+        result = adapter.render_from_memory(
+            plan=plan,
+            clips=clips,
+            asset_path_map={},
+            plan_id="mem_no_assets",
+            work_dir=tmp_path / "work",
+        )
+
+        assert not result.success
+        assert "素材路径映射为空" in result.error_message
+        mock_render_cls.assert_not_called()
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_bgm_prepared_for_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
+        """内存模式下 BGM 配置也会被正确处理。"""
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=512,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        plan = FakePlan(
+            id="mem_bgm",
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "preset_id": "preset_001",
+                }
+            },
+        )
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
+        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
+
+        adapter, _, _ = _make_adapter()
+
+        # mock _prepare_bgm 返回一个假的 bgm 路径
+        fake_bgm_path = tmp_path / "bgm.mp3"
+        fake_bgm_path.write_bytes(b"fake bgm")
+        with patch.object(adapter, "_prepare_bgm", return_value=str(fake_bgm_path)):
+            result = adapter.render_from_memory(
+                plan=plan,
+                clips=clips,
+                asset_path_map=asset_path_map,
+                plan_id="mem_bgm",
+                work_dir=tmp_path / "work",
+            )
+
+        assert result.success
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["bgm_path"] == str(fake_bgm_path)
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_resolution_from_config_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
+        """内存模式下从 plan.config.export.resolution 读取分辨率。"""
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=512,
+            width=720,
+            height=1280,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        plan = FakePlan(
+            id="mem_res",
+            config={"export": {"resolution": "720x1280"}},
+        )
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
+        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
+
+        adapter, _, _ = _make_adapter()
+
+        result = adapter.render_from_memory(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_path_map,
+            plan_id="mem_res",
+            work_dir=tmp_path / "work",
+        )
+
+        assert result.success
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["output_width"] == 720
+        assert call_kwargs.kwargs["output_height"] == 1280
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    def test_ffmpeg_error_captured_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
+        """内存模式下 FFmpeg 错误也会被正确捕获。"""
+        import subprocess
+
+        mock_render = MagicMock()
+        mock_render.render.side_effect = subprocess.CalledProcessError(
+            returncode=1,
+            cmd=["ffmpeg", "-i", "input.mp4", "output.mp4"],
+            stderr="test error output",
+        )
+        mock_render_cls.return_value = mock_render
+
+        plan = FakePlan(id="mem_err")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
+        asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
+
+        adapter, _, _ = _make_adapter()
+
+        result = adapter.render_from_memory(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_path_map,
+            plan_id="mem_err",
+            work_dir=tmp_path / "work",
+        )
+
+        assert not result.success
+        assert "FFmpeg渲染失败" in result.error_message
+        assert "test error output" in result.error_detail