apps/api/app/api/routes/generation_preview.py

@@ -45,7 +45,6 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
-PREVIEW_RESOLUTION = "854x480"
 
 # 模板 mode → 视频比例映射
 _TEMPLATE_MODE_TO_RATIO = {
@@ -55,24 +54,7 @@ _TEMPLATE_MODE_TO_RATIO = {
 }
 
 
-def _calc_preview_resolution(video_ratio: str = "") -> str:
-    """根据视频比例计算预览分辨率（短边 480，长边按比例）。
-
-    支持的比例：16:9, 9:16, 1:1, 4:3, 3:4, 其他默认 16:9。
-    """
-    ratio_map = {
-        "16:9": "854x480",
-        "9:16": "480x854",
-        "1:1": "480x480",
-        "4:3": "640x480",
-        "3:4": "480x640",
-    }
-    return ratio_map.get(video_ratio.strip(), PREVIEW_RESOLUTION)
-
-
-def _infer_video_ratio_from_template(
-    template_id: str, db: Session, user_id: str = ""
-) -> str:
+def _infer_video_ratio_from_template(template_id: str, db: Session, user_id: str = "") -> str:
     """从模板 mode 推断视频比例，前端未传 video_ratio 时使用。
 
     Returns:
@@ -102,9 +84,7 @@ def _infer_video_ratio_from_template(
     return ""
 
 
-def _resolve_strategy_id_from_template(
-    template_id: str, db: Session, user_id: str = ""
-) -> str:
+def _resolve_strategy_id_from_template(template_id: str, db: Session, user_id: str = "") -> str:
     """从模板读取 editing_mode / mode 作为 strategy_id。
 
     优先查新模板系统（EditTemplate.editing_mode），fallback 旧模板（Template.mode）。
@@ -236,7 +216,7 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
         status=task.status.value if hasattr(task.status, "value") else str(task.status),
         progress=float(task.progress or 0.0),
         is_preview=bool(getattr(task, "is_preview", True)),
-        resolution=getattr(task, "resolution", PREVIEW_RESOLUTION) or PREVIEW_RESOLUTION,
+        resolution=getattr(task, "resolution", "") or "",
         video_url=video_url,
         duration=duration,
         file_size=file_size,
@@ -260,7 +240,7 @@ def create_preview_generation_task(
 ) -> PreviewGenerationTaskResponse:
     """创建预览生成任务。
 
-    预览为完整时长的低清版（480p + 低码率），效果与正式生成一致，仅清晰度降低。
+    预览渲染品质与正式生成一致（1080p, CRF 23, medium preset），确认生成时可直接复用预览产物。
 
     Args:
         request: 预览任务创建请求（template_id + asset_ids 等）
@@ -318,11 +298,11 @@ def create_preview_generation_task(
                 title_ids=list(request.title_ids),
                 voice_ids=list(request.voice_ids),
                 created_by_user_id=user_id,
-                source_edit_plan_id="",
+                source_edit_plan_id=request.source_edit_plan_id,
                 asset_select_mode="",
                 batch_id="",
                 video_title=request.video_title,
-                resolution=_calc_preview_resolution(video_ratio),
+                resolution="",
                 bgm_config=request.bgm_config or {},
                 auto_retry_enabled=False,
                 auto_retry_max=0,

apps/api/app/api/routes/generation_tasks.py

@@ -352,10 +352,11 @@ def confirm_generation(
     generation_task_repository: Any = Depends(get_generation_task_repository),
     project_repository: Any = Depends(get_project_repository),
 ) -> BatchGenerationTaskResponse:
-    """确认生成 — 基于预览任务创建正式生成任务。
+    """确认生成 -- 复用预览渲染产物（预览与正式品质一致）。
 
-    查找预览任务，复制其配置，创建新的正式生成任务（is_preview=False），
-    使用高分辨率，复用 worker.generate_video 渲染路径。
+    预览已使用 1080p / CRF 23 / medium 渲染，品质与正式生成一致。
+    确认时直接将预览任务标记为正式产出，无需重新渲染，实现秒出。
+    仅当预览任务未完成时，才创建新的正式任务走渲染流程。
     """
     # 1. 查找源预览任务
     source_task = generation_task_repository.get(task_id)
@@ -368,7 +369,43 @@ def confirm_generation(
     if source_task.project_id:
         check_project_access(source_task.project_id, authenticated_user.user.id, project_repository)
 
-    # 3. 创建正式生成任务，复制预览任务的配置
+    # 3. 如果预览任务已完成，检查分辨率一致性后复用产物（秒出）
+    if source_task.is_completed and getattr(source_task, "is_preview", False):
+        # 校验请求的分辨率是否与预览实际渲染的分辨率一致
+        req_w = request.output_width or 0
+        req_h = request.output_height or 0
+        src_w = getattr(source_task, "output_width", 0) or 0
+        src_h = getattr(source_task, "output_height", 0) or 0
+        resolution_match = (req_w == 0 or req_w == src_w) and (req_h == 0 or req_h == src_h)
+
+        if resolution_match:
+            source_task.mark_confirmed(
+                cover_url=request.cover_url,
+                custom_title=request.custom_title,
+                output_width=request.output_width,
+                output_height=request.output_height,
+            )
+            generation_task_repository.update(source_task)
+            logger.info(
+                "[确认生成] 复用预览产物: task_id=%s, user_id=%s",
+                task_id,
+                authenticated_user.user.id,
+            )
+            return BatchGenerationTaskResponse(
+                items=[_to_generation_task_response(source_task)],
+                total=1,
+            )
+        # 分辨率不一致，跳过复用，走新建任务流程
+        logger.info(
+            "[确认生成] 分辨率不一致，跳过复用: task_id=%s, src=%sx%s, req=%sx%s",
+            task_id,
+            src_w,
+            src_h,
+            req_w,
+            req_h,
+        )
+
+    # 4. 预览任务未完成，创建新的正式任务走渲染流程
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
     new_task = use_case.execute(
         CreateGenerationTaskCommand(
@@ -394,7 +431,7 @@ def confirm_generation(
         )
     )
 
-    # 4. 调度 worker.generate_video（同一条渲染路径）
+    # 5. 调度 worker
     try:
         if not safe_enqueue_generation_task(
             new_task,

apps/api/app/api/routes/templates_editor/generation.py

@@ -18,6 +18,7 @@ from app.dependencies import (
     get_asset_library_repository,
     get_asset_repository,
     get_db_session,
+    get_generated_video_repository,
 )
 from app.schemas.generation_task import GenerationTaskResponse
 from app.services.edit_plan_service import EditPlanService
@@ -28,6 +29,7 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.generation_task_repository import (
     SQLAlchemyGenerationTaskRepository,
 )
+from packages.application.generated_videos import ListGeneratedVideosByTaskUseCase
 from packages.application.generation_tasks import (
     CreateGenerationTaskCommand,
     CreateGenerationTaskUseCase,
@@ -80,6 +82,40 @@ def generate_editor_draft(
         user_id=str(current_user.user.id),
     )
 
+    # 检查是否可复用已完成的预览产物（预览品质已与正式一致）
+    gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+    reusable_task = _find_reusable_preview_task(gen_task_repo, plan_id, plan_check)
+    if reusable_task:
+        # 复用预览产物：标记为正式产出，跳过渲染
+        reusable_task.mark_confirmed()
+        gen_task_repo.update(reusable_task)
+
+        # 将产物 URL 写入 plan config
+        rendered_url = _get_task_output_url(reusable_task, gen_task_repo, db)
+        plan_svc.update_plan_config(
+            plan_id,
+            {
+                "generation_task_id": reusable_task.id,
+                "rendered_url": rendered_url,
+            },
+        )
+        plan_svc.transition_status(plan_id, EditPlanStatus.COMPLETED)
+
+        updated_plan = plan_svc.get_plan_or_raise(plan_id)
+        logger.info(
+            "模板编辑器复用预览产物: template_id=%s plan_id=%s task_id=%s by user=%s",
+            template_id,
+            plan_id,
+            reusable_task.id,
+            current_user.user.id,
+        )
+        return EditPlanGenerateResponse(
+            plan_id=plan_id,
+            plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
+            generation_task_id=reusable_task.id,
+            clip_count=len((plan_check.config or {}).get("clips", [])),
+        )
+
     # 检查是否可生成（含最后防线自动修复 + 诊断日志）
     try:
         can_gen, reason = plan_svc.can_generate(plan_id)
@@ -91,7 +127,6 @@ def generate_editor_draft(
     try:
         clip_count = plan_svc.mark_clips_ready(plan_id)
 
-        gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
         user_id = current_user.user.id
         _check_queue_limits(gen_task_repo, user_id)
 
@@ -147,6 +182,53 @@ def generate_editor_draft(
         ) from _e
 
 
+def _find_reusable_preview_task(gen_task_repo, plan_id: str, plan) -> "object | None":
+    """查找该 plan 关联的已完成预览任务，判断是否可复用。
+
+    复用条件：
+    1. 存在 source_edit_plan_id == plan_id 的已完成预览任务
+    2. plan 在预览完成后未被修改（updated_at <= 预览完成时间）
+
+    Returns:
+        可复用的 GenerationTask，或 None
+    """
+    try:
+        tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
+    except Exception:
+        return None
+
+    for task in tasks:
+        if not getattr(task, "is_preview", False):
+            continue
+        if not task.is_completed:
+            continue
+        # 检查 plan 是否在预览完成后被修改
+        completed_at = getattr(task, "completed_at", None)
+        if completed_at and hasattr(plan, "updated_at"):
+            plan_updated = plan.updated_at
+            # 如果 plan.updated_at 为空，无法判断是否修改过，跳过
+            if plan_updated is None:
+                continue
+            # 如果 plan 在预览完成后又被修改了，不能复用
+            if plan_updated > completed_at:
+                continue
+        return task
+    return None
+
+
+def _get_task_output_url(task, gen_task_repo, db) -> str:
+    """获取任务的输出视频 URL。"""
+    try:
+        video_repo = get_generated_video_repository(db)
+        use_case = ListGeneratedVideosByTaskUseCase(video_repo)
+        videos = use_case.execute(task.id)
+        if videos:
+            return getattr(videos[0], "file_url", "") or ""
+    except Exception:
+        pass
+    return ""
+
+
 @router.get("/generation-status", response_model=EditPlanGenerationStatusResponse)
 def get_editor_generation_status(
     template_id: str,

apps/api/app/schemas/generation_task.py

@@ -154,7 +154,7 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
     """创建预览生成任务请求。
 
     仅支持模板模式：template_id + asset_ids 等素材 ID 列表。
-    预览为完整时长低清版（480p + 低码率）。
+    预览渲染品质与正式生成一致（1080p, CRF 23, medium preset）。
     """
 
     template_id: str
@@ -177,6 +177,10 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
         le=10,
         description="预览视频生成数量，范围 1-10，默认 1",
     )
+    source_edit_plan_id: str = Field(
+        default="",
+        description="关联的编辑计划ID（可选），用于确认生成时复用预览产物",
+    )
 
     @model_validator(mode="after")
     def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":

apps/worker/video_processing/render_adapter.py

@@ -485,7 +485,6 @@ class RenderAdapter:
         rendered_clip_ids: list[str] | None = None,
         failed_clip_ids: list[str] | None = None,
         voiceover_audio_path: str | None = None,
-        is_preview: bool = False,
     ) -> RenderAdapterResult:
         """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。
 
@@ -505,10 +504,8 @@ class RenderAdapter:
         self._report_progress(progress_cb, 40.0, "执行视频渲染")
 
         # 2. 初始化 ASR
-        # 预览模式下，如果 plan.config 中存在 voice_id，仍需初始化 ASR 以支持配音
         plan_config = plan.config or {}
-        has_voice_id = bool(plan_config.get("voice_id"))
-        asr_service = None if (is_preview and not has_voice_id) else self._get_asr_service()
+        asr_service = self._get_asr_service()
 
         # 3. 读取输出分辨率
         export_config = plan_config.get("export", {}) or {}
@@ -532,27 +529,23 @@ class RenderAdapter:
             bgm_path=bgm_path,
             asr_service=asr_service,
             voiceover_audio_path=voiceover_audio_path,
-            is_preview=is_preview,
         )
         result = render_svc.render()
 
-        # 4.5 渲染后校验输出完整性（预览模式跳过，节省耗时）
-        if is_preview:
-            logger.info("[render-adapter] 预览模式：跳过输出校验")
-        else:
-            validation = validate_video_output(result.output_path)
-            if not validation.valid:
-                logger.error(
-                    "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s",
-                    plan_id,
-                    job_id,
-                    validation.error_message,
-                )
-                return RenderAdapterResult(
-                    success=False,
-                    error_message=f"渲染输出校验失败: {validation.error_message}",
-                    error_detail=validation.error_message,
-                )
+        # 4.5 渲染后校验输出完整性
+        validation = validate_video_output(result.output_path)
+        if not validation.valid:
+            logger.error(
+                "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s",
+                plan_id,
+                job_id,
+                validation.error_message,
+            )
+            return RenderAdapterResult(
+                success=False,
+                error_message=f"渲染输出校验失败: {validation.error_message}",
+                error_detail=validation.error_message,
+            )
         self._report_progress(progress_cb, 80.0, "上传渲染结果")
 
         # 5. 上传结果
@@ -561,20 +554,19 @@ class RenderAdapter:
 
         self._report_progress(progress_cb, 90.0, "生成封面缩略图")
 
-        # 6. 生成缩略图（预览模式跳过，节省耗时）
+        # 6. 生成封面缩略图
         thumbnail_url = ""
-        if not is_preview:
-            try:
-                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+        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,
-                )
+            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, "渲染完成")
 
@@ -620,7 +612,6 @@ class RenderAdapter:
         work_dir: Path | None = None,
         progress_cb: ProgressCallback | None = None,
         voiceover_audio_path: str | None = None,
-        is_preview: bool = False,
     ) -> RenderAdapterResult:
         """使用内存中的 plan/clips/asset_path_map 直接渲染。
 
@@ -679,7 +670,6 @@ class RenderAdapter:
                 job_id=job_id,
                 progress_cb=progress_cb,
                 voiceover_audio_path=voiceover_audio_path,
-                is_preview=is_preview,
             )
 
         except subprocess.CalledProcessError as exc:

apps/worker/video_processing/unified_render_service.py

@@ -151,7 +151,6 @@ class UnifiedRenderService:
         asr_service: Any = None,  # ASRService 实例，用于自动生成字幕
         bgm_path: str | None = None,  # BGM 本地文件路径
         voiceover_audio_path: str | None = None,  # 配音素材库音频本地路径
-        is_preview: bool = False,  # 预览模式：ultrafast 编码 + 跳过非必要步骤
     ):
         self.plan = plan
         self.clips = clips
@@ -164,7 +163,6 @@ class UnifiedRenderService:
         self.asr_service = asr_service
         self.bgm_path = bgm_path
         self.voiceover_audio_path = voiceover_audio_path
-        self.is_preview = is_preview
         self._transition_engine = TransitionEngine(default_duration=transition_duration)
         self._speed_engine = SpeedEngine()
         self._asr_timeline_cache: Any = None  # ASR 字幕结果缓存，避免重复调用
@@ -1250,9 +1248,9 @@ class UnifiedRenderService:
             "-c:v",
             "libx264",
             "-crf",
-            "28" if self.is_preview else "23",
+            "23",
             "-preset",
-            "ultrafast" if self.is_preview else "medium",
+            "medium",
             "-pix_fmt",
             "yuv420p",
             "-movflags",
@@ -1776,9 +1774,9 @@ class UnifiedRenderService:
             "-c:v",
             "libx264",
             "-crf",
-            "28" if self.is_preview else "23",
+            "23",
             "-preset",
-            "ultrafast" if self.is_preview else "medium",
+            "medium",
             "-pix_fmt",
             "yuv420p",
             "-movflags",
@@ -1787,11 +1785,10 @@ class UnifiedRenderService:
         ]
 
         logger.info(
-            "执行渲染: plan_id=%s inputs=%d output=%s preview=%s",
+            "执行渲染: plan_id=%s inputs=%d output=%s",
             self.plan.id,
             input_args.count("-i"),
             output_path,
-            self.is_preview,
         )
         try:
             run_ffmpeg(command)

apps/worker/worker_app/tasks/generation.py

@@ -1039,7 +1039,6 @@ def _load_task_info(task_id: str) -> dict | None:
             "video_title": getattr(gen_task, "video_title", "") or "",
             "resolution": getattr(gen_task, "resolution", "") or "",
             "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}),
-            "is_preview": bool(getattr(gen_task, "is_preview", False)),
             "source_task_id": getattr(gen_task, "source_task_id", "") or "",
             "output_width": getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH,
             "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT,
@@ -1123,7 +1122,6 @@ def _render_video(
     output_name: str,
     resolution: str = "",
     bgm_config: dict | None = None,
-    is_preview: bool = False,
     voice_ids: list[str] | None = None,
 ) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
@@ -1131,8 +1129,6 @@ def _render_video(
     使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
 
     Args:
-        is_preview: 是否为预览生成，若是则强制 480p + 低码率
-
     Returns:
         (output_path, render_duration)
     """
@@ -1179,18 +1175,7 @@ def _render_video(
     # 注意：必须拷贝字典，避免预览模式修改污染源对象（模板配置）
     plan_cfg = dict(virtual_plan.config or {})
     export_cfg = dict(plan_cfg.get("export", {}) or {})
-    if is_preview:
-        # 预览模式：短边 480p + 低码率，但尊重视频比例（竖屏模板不应强制横屏）
-        preview_res = resolution if resolution else "854x480"
-        export_cfg["resolution"] = preview_res
-        export_cfg["bitrate"] = "1M"
-        logger.info(
-            "[task_id=%s] [渲染] 预览模式：分辨率=%s, 码率=%s",
-            task_id,
-            preview_res,
-            "1M",
-        )
-    elif resolution:
+    if resolution:
         # 用户在 API 调用时指定的分辨率优先级最高
         export_cfg["resolution"] = resolution
     elif not export_cfg.get("resolution"):
@@ -1239,7 +1224,6 @@ def _render_video(
             job_id=task_id,
             work_dir=temp_path,
             voiceover_audio_path=voice_path,
-            is_preview=is_preview,
         )
     finally:
         db.close()
@@ -1474,7 +1458,6 @@ def generate_video(self, task_id: str) -> dict:
                 output_name=output_name,
                 resolution=_resolved_resolution,
                 bgm_config=task_info.get("bgm_config", {}),
-                is_preview=task_info.get("is_preview", False),
                 voice_ids=task_info.get("voice_ids", []),
             )
 

packages/domain/generation_task.py

@@ -296,6 +296,30 @@ class GenerationTask:
         self.transition_to(GenerationTaskStatus.CANCELLED)
         self.completed_at = datetime.now(timezone.utc)
 
+    def mark_confirmed(
+        self,
+        *,
+        cover_url: str = "",
+        custom_title: str = "",
+        output_width: int = 0,
+        output_height: int = 0,
+    ) -> None:
+        """将预览任务确认为正式产出。
+
+        预览渲染品质已与正式生成一致（1080p, CRF 23, medium），
+        确认时直接复用已有产物，无需重新渲染。
+        """
+        self.is_preview = False
+        if cover_url:
+            self.cover_url = cover_url
+        if custom_title:
+            self.custom_title = custom_title
+        if output_width > 0:
+            self.output_width = output_width
+        if output_height > 0:
+            self.output_height = output_height
+        self.updated_at = datetime.now(timezone.utc)
+
     # ── 日志辅助 ────────────────────────────────────────────────────────────
 
     _MAX_LOGS = 200

tests/unit/test_1280_preview_speedup.py

@@ -1,12 +1,10 @@
-"""#1280 预览视频生成加速 — 单元测试。
+"""统一渲染管线 — 预览与确认生成使用相同品质参数。
 
 验证点：
-1. UnifiedRenderService.is_preview 参数正确传递
-2. 预览模式使用 ultrafast preset + crf 28
-3. RenderAdapter.render_from_memory 正确传递 is_preview
-4. 预览模式跳过 ASR 初始化
-5. 预览模式跳过输出校验和缩略图
-6. generation.py 并行下载逻辑
+1. UnifiedRenderService 不再有 is_preview 参数
+2. 所有渲染统一使用 medium preset + CRF 23
+3. RenderAdapter 统一执行校验和缩略图生成
+4. generation.py 并行下载逻辑（保留）
 """
 
 from __future__ import annotations
@@ -18,13 +16,24 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-# ── 1. UnifiedRenderService is_preview 参数 ──
+# ── 1. UnifiedRenderService 无 is_preview 参数 ──
 
 
-class TestUnifiedRenderServicePreviewFlag:
-    """is_preview 参数正确传递和存储。"""
+class TestUnifiedRenderServiceNoPreviewParam:
+    """UnifiedRenderService 构造函数不再接受 is_preview 参数。"""
 
-    def test_default_is_preview_false(self):
+    def test_constructor_has_no_is_preview(self):
+        import inspect
+
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        sig = inspect.signature(UnifiedRenderService.__init__)
+        param_names = list(sig.parameters.keys())
+        assert (
+            "is_preview" not in param_names
+        ), f"is_preview should be removed from UnifiedRenderService.__init__, found params: {param_names}"
+
+    def test_no_is_preview_attribute(self):
         from video_processing.unified_render_service import UnifiedRenderService
 
         svc = UnifiedRenderService(
@@ -33,38 +42,16 @@ class TestUnifiedRenderServicePreviewFlag:
             asset_path_map={},
             work_dir=Path(tempfile.mkdtemp()),
         )
-        assert svc.is_preview is False
-
-    def test_is_preview_true(self):
-        from video_processing.unified_render_service import UnifiedRenderService
-
-        svc = UnifiedRenderService(
-            plan=MagicMock(id="test"),
-            clips=[],
-            asset_path_map={},
-            work_dir=Path(tempfile.mkdtemp()),
-            is_preview=True,
-        )
-        assert svc.is_preview is True
-
-    def test_is_preview_false_explicit(self):
-        from video_processing.unified_render_service import UnifiedRenderService
-
-        svc = UnifiedRenderService(
-            plan=MagicMock(id="test"),
-            clips=[],
-            asset_path_map={},
-            work_dir=Path(tempfile.mkdtemp()),
-            is_preview=False,
-        )
-        assert svc.is_preview is False
+        assert not hasattr(
+            svc, "is_preview"
+        ), "UnifiedRenderService should not have is_preview attribute after unification"
 
 
-# ── 2. 预览模式 FFmpeg 参数 ──
+# ── 2. FFmpeg 参数统一为 medium + CRF 23 ──
 
 
-class TestPreviewFFmpegPreset:
-    """预览模式使用 ultrafast preset + crf 28。"""
+class TestUnifiedFFmpegPreset:
+    """所有渲染统一使用 medium preset + CRF 23。"""
 
     def _make_clip(self):
         from video_processing.unified_render_service import ResolvedClip
@@ -84,46 +71,7 @@ class TestPreviewFFmpegPreset:
         )
 
     @patch("video_processing.unified_render_service.run_ffmpeg")
-    def test_execute_ffmpeg_preview_uses_ultrafast(self, mock_run):
-        from video_processing.unified_render_service import (
-            RenderLayer,
-            UnifiedRenderService,
-        )
-
-        plan = MagicMock()
-        plan.id = "test_plan"
-        plan.config = {"export": {"resolution": "854x480"}}
-
-        clip = self._make_clip()
-
-        svc = UnifiedRenderService(
-            plan=plan,
-            clips=[clip],
-            asset_path_map={"a1": Path("/tmp/fake.mp4")},
-            work_dir=Path(tempfile.mkdtemp()),
-            output_width=854,
-            output_height=480,
-            is_preview=True,
-        )
-
-        layers = [RenderLayer(role="main", clips=[clip])]
-        filter_complex, input_args = svc._build_filter_complex(layers)
-        output_path = Path(tempfile.mkdtemp()) / "out.mp4"
-        svc._execute_ffmpeg(filter_complex, input_args, output_path)
-
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
-
-        # Check preset is ultrafast
-        preset_idx = cmd.index("-preset")
-        assert cmd[preset_idx + 1] == "ultrafast", f"Expected ultrafast, got {cmd[preset_idx + 1]}"
-
-        # Check crf is 28
-        crf_idx = cmd.index("-crf")
-        assert cmd[crf_idx + 1] == "28", f"Expected crf 28, got {cmd[crf_idx + 1]}"
-
-    @patch("video_processing.unified_render_service.run_ffmpeg")
-    def test_execute_ffmpeg_normal_uses_medium(self, mock_run):
+    def test_execute_ffmpeg_uses_medium_crf23(self, mock_run):
         from video_processing.unified_render_service import (
             RenderLayer,
             UnifiedRenderService,
@@ -140,9 +88,8 @@ class TestPreviewFFmpegPreset:
             clips=[clip],
             asset_path_map={"a1": Path("/tmp/fake.mp4")},
             work_dir=Path(tempfile.mkdtemp()),
-            output_width=1280,
-            output_height=720,
-            is_preview=False,
+            output_width=1920,
+            output_height=1080,
         )
 
         layers = [RenderLayer(role="main", clips=[clip])]
@@ -153,137 +100,73 @@ class TestPreviewFFmpegPreset:
         mock_run.assert_called_once()
         cmd = mock_run.call_args[0][0]
 
+        # Check preset is medium (no conditional)
         preset_idx = cmd.index("-preset")
-        assert cmd[preset_idx + 1] == "medium"
+        assert cmd[preset_idx + 1] == "medium", f"Expected medium, got {cmd[preset_idx + 1]}"
 
+        # Check crf is 23 (no conditional)
         crf_idx = cmd.index("-crf")
-        assert cmd[crf_idx + 1] == "23"
+        assert cmd[crf_idx + 1] == "23", f"Expected crf 23, got {cmd[crf_idx + 1]}"
 
 
-# ── 3. RenderAdapter passes is_preview ──
+# ── 3. RenderAdapter 统一执行校验和缩略图 ──
 
 
-class TestRenderAdapterPreviewPassthrough:
-    """RenderAdapter 正确传递 is_preview 参数。"""
+class TestRenderAdapterUnifiedPostProcess:
+    """RenderAdapter 不再跳过校验和缩略图。"""
+
+    def test_render_adapter_no_is_preview_param(self):
+        import inspect
 
-    def test_render_from_memory_passes_is_preview(self):
         from video_processing.render_adapter import RenderAdapter
 
-        db = MagicMock()
-        adapter = RenderAdapter(db)
+        # Check render_from_memory signature
+        sig = inspect.signature(RenderAdapter.render_from_memory)
+        param_names = list(sig.parameters.keys())
+        assert (
+            "is_preview" not in param_names
+        ), f"is_preview should be removed from render_from_memory, found params: {param_names}"
 
-        plan = MagicMock()
-        plan.id = "test_plan"
-        plan.config = {"export": {"resolution": "854x480"}}
-
-        clip = MagicMock()
-        clip.id = "c1"
-
-        with patch.object(adapter, "_do_render") as mock_do_render:
-            mock_do_render.return_value = MagicMock(
-                success=True,
-                output_path=Path("/tmp/out.mp4"),
-                thumbnail_url="",
-                duration=5.0,
-                file_size=1000,
-                width=854,
-                height=480,
-                output_url="https://oss/test.mp4",
-                rendered_clip_ids=["c1"],
-                failed_clip_ids=[],
-            )
-
-            adapter.render_from_memory(
-                plan=plan,
-                clips=[clip],
-                asset_path_map={"a1": Path("/tmp/fake.mp4")},
-                is_preview=True,
-            )
-
-            mock_do_render.assert_called_once()
-            _, kwargs = mock_do_render.call_args
-            assert kwargs.get("is_preview") is True
-
-
-# ── 4. 预览模式跳过 ASR ──
-
-
-class TestPreviewSkipsASR:
-    """预览模式跳过 ASR 初始化。"""
-
-    def test_render_method_source_has_asr_skip(self):
-        """_do_render 在 is_preview=True 时不调用 _get_asr_service。"""
+    def test_no_preview_skip_validation_in_source(self):
+        """渲染适配器源码中不再包含预览跳过校验的逻辑。"""
         with open("apps/worker/video_processing/render_adapter.py") as f:
             source = f.read()
 
-        assert (
-            "None if (is_preview and not has_voice_id) else self._get_asr_service()" in source
-        ), "Should skip ASR initialization in preview mode unless voice_id is provided"
+        assert "预览模式：跳过输出校验" not in source, "Should not skip validation in any mode"
+        assert "if not is_preview:" not in source, "Thumbnail should always be generated"
 
 
-# ── 5. 并行下载逻辑 ──
+# ── 4. generation.py 不再有 is_preview 覆盖逻辑 ──
 
 
-class TestParallelDownload:
-    """generation.py 并行下载素材。"""
+class TestWorkerGenerationNoPreviewOverride:
+    """Worker generation.py 不再覆盖预览分辨率为 480p。"""
 
-    def test_parallel_download_uses_thread_pool(self):
+    def test_no_480p_override(self):
+        with open("apps/worker/worker_app/tasks/generation.py") as f:
+            source = f.read()
+
+        assert 'resolution = "854x480"' not in source, "Should not override resolution to 480p in preview mode"
+        assert 'bitrate = "1M"' not in source, "Should not override bitrate to 1M in preview mode"
+
+    def test_parallel_download_still_works(self):
+        """并行下载逻辑保留。"""
         with open("apps/worker/worker_app/tasks/generation.py") as f:
             source = f.read()
 
         assert "ThreadPoolExecutor" in source, "Should use ThreadPoolExecutor for parallel downloads"
         assert "as_completed" in source, "Should use as_completed for result collection"
 
-    def test_parallel_download_preserves_order(self):
-        with open("apps/worker/worker_app/tasks/generation.py") as f:
+
+# ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ──
+
+
+class TestPreviewNoLowQualityConstants:
+    """预览 API 不再定义低质量常量。"""
+
+    def test_no_preview_resolution_constant(self):
+        with open("apps/api/app/api/routes/generation_preview.py") as f:
             source = f.read()
 
-        assert "sorted(results_map.keys())" in source, "Should sort results by original index"
-
-
-# ── 6. generation.py _render_video passes is_preview ──
-
-
-class TestRenderVideoPassesPreview:
-    """_render_video 正确传递 is_preview 到 render_from_memory。"""
-
-    def test_render_video_passes_is_preview(self):
-        with open("apps/worker/worker_app/tasks/generation.py") as f:
-            source = f.read()
-
-        assert "is_preview=is_preview" in source, "Should pass is_preview to render_from_memory"
-
-
-# ── 7. Preview mode skips thumbnail and validation ──
-
-
-class TestPreviewSkipsThumbnailAndValidation:
-    """预览模式跳过缩略图生成和输出校验。"""
-
-    def test_render_adapter_skips_thumbnail_in_preview(self):
-        with open("apps/worker/video_processing/render_adapter.py") as f:
-            source = f.read()
-
-        assert "if not is_preview:" in source, "Thumbnail should be conditional on is_preview"
-
-    def test_render_adapter_skips_validation_in_preview(self):
-        with open("apps/worker/video_processing/render_adapter.py") as f:
-            source = f.read()
-
-        assert "预览模式：跳过输出校验" in source, "Should skip validation in preview mode"
-
-
-# ── 8. Pass-through rendering uses ultrafast in preview ──
-
-
-class TestPassThroughPreviewPreset:
-    """直通渲染在预览模式也使用 ultrafast。"""
-
-    def test_pass_through_has_preview_preset(self):
-        with open("apps/worker/video_processing/unified_render_service.py") as f:
-            source = f.read()
-
-        # The pass_through method should also use ultrafast for preview
-        # Count occurrences of "ultrafast" - should be at least 2 (execute_ffmpeg + pass_through)
-        count = source.count('"ultrafast" if self.is_preview')
-        assert count >= 2, f"Expected at least 2 ultrafast preset usages, found {count}"
+        assert "PREVIEW_RESOLUTION" not in source, "PREVIEW_RESOLUTION constant should be removed"
+        assert "_calc_preview_resolution" not in source, "_calc_preview_resolution function should be removed"

tests/unit/test_1294_preview_voice_injection.py

@@ -169,7 +169,6 @@ class TestRenderVideoVoiceInjection:
                 temp_path=Path("/tmp"),
                 output_name="test_output.mp4",
                 resolution="854x480",
-                is_preview=True,
                 voice_ids=["voice_abc"],
             )
 

tests/unit/test_confirm_generation.py

@@ -1,10 +1,10 @@
 """确认生成 API 单元测试.
 
 覆盖 POST /tasks/{task_id}/confirm 端点：
-  - 正常确认流程
+  - 预览任务已完成 → 直接复用（mark_confirmed），秒出
+  - 预览任务未完成 → 创建新任务走渲染流程
   - 预览任务不存在 → 404
   - 权限不足 → 403
-  - is_preview=False 及分辨率正确
   - cover_url 和 custom_title 正确传递
 """
 
@@ -164,7 +164,6 @@ def app(
     test_app.dependency_overrides[get_current_user] = override_get_current_user
     test_app.dependency_overrides[get_generation_task_repository] = override_get_generation_task_repository
     test_app.dependency_overrides[get_project_repository] = override_get_project_repository
-    # Stubs for repositories not used by confirm endpoint but required by router
     test_app.dependency_overrides[get_asset_library_repository] = lambda: MagicMock()
     test_app.dependency_overrides[get_asset_repository] = lambda: MagicMock()
     test_app.dependency_overrides[get_generated_video_repository] = lambda: MagicMock()
@@ -199,8 +198,8 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask:
         asset_select_mode="all",
         is_preview=True,
         source_task_id="",
-        output_width=1280,
-        output_height=720,
+        output_width=1080,
+        output_height=1920,
         cover_url="",
         custom_title="",
         video_title="",
@@ -214,45 +213,97 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask:
 # ── Tests ────────────────────────────────────────────────────────────────────
 
 
-class TestConfirmGeneration:
-    def test_confirm_success(
+class TestConfirmGenerationReuse:
+    """确认生成复用预览产物。"""
+
+    def test_confirm_reuses_completed_preview(
         self,
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ) -> None:
-        """正常确认流程：预览任务存在、权限正确 → 创建正式任务"""
+        """预览已完成 → 直接复用，返回同一个任务，不创建新任务"""
         preview = _make_preview_task()
         gen_task_repo.create(preview)
 
-        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True) as mock_enqueue:
-            resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
-                json={
-                    "output_width": 1080,
-                    "output_height": 1920,
-                    "cover_url": "https://example.com/cover.jpg",
-                    "custom_title": "我的视频",
-                },
-            )
+        initial_count = len(gen_task_repo._store)
+
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={
+                "output_width": 1080,
+                "output_height": 1920,
+                "cover_url": "https://example.com/cover.jpg",
+                "custom_title": "我的视频",
+            },
+        )
 
         assert resp.status_code == 200
         data = resp.json()
         assert data["total"] == 1
         item = data["items"][0]
+
+        # 返回的是同一个任务（复用）
+        assert item["id"] == preview.id
+        # is_preview 变为 False
         assert item["is_preview"] is False
-        assert item["source_task_id"] == preview.id
+        # 分辨率更新
         assert item["output_width"] == 1080
         assert item["output_height"] == 1920
+        # 封面和标题更新
         assert item["cover_url"] == "https://example.com/cover.jpg"
         assert item["custom_title"] == "我的视频"
-        # 复制了预览任务的配置
-        assert item["project_id"] == "project-001"
-        assert item["asset_library_id"] == "library-001"
-        assert item["strategy_id"] == "one_take"
-        assert item["asset_ids"] == ["asset-1"]
 
-        # 验证入队函数被调用
-        mock_enqueue.assert_called_once()
+        # 没有创建新任务
+        assert len(gen_task_repo._store) == initial_count
+
+    def test_confirm_updates_task_in_repo(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认后的任务在 repo 中被更新"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={"cover_url": "https://cdn.example.com/cover.png", "custom_title": "测试标题"},
+        )
+
+        assert resp.status_code == 200
+
+        # 验证 repo 中的任务已被更新
+        updated = gen_task_repo.get(preview.id)
+        assert updated is not None
+        assert updated.is_preview is False
+        assert updated.cover_url == "https://cdn.example.com/cover.png"
+        assert updated.custom_title == "测试标题"
+
+    def test_confirm_creates_new_task_when_preview_not_completed(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """预览任务未完成 → 创建新任务走渲染流程"""
+        preview = _make_preview_task(status=GenerationTaskStatus.RUNNING, progress=50.0)
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            resp = client.post(
+                f"/api/v1/tasks/{preview.id}/confirm",
+                json={"output_width": 1080, "output_height": 1920},
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        # 创建了新任务
+        assert item["id"] != preview.id
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
+
+
+class TestConfirmGenerationErrors:
+    """确认生成的错误处理。"""
 
     def test_confirm_not_found(self, client: TestClient) -> None:
         """预览任务不存在 → 404"""
@@ -284,7 +335,7 @@ class TestConfirmGeneration:
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ) -> None:
-        """确认后的任务 is_preview=False，分辨率已更新，其余配置从预览任务复制"""
+        """确认后任务保留预览任务的全部配置"""
         preview = _make_preview_task(
             voice_library_id="voice-001",
             template_id="tmpl-001",
@@ -293,22 +344,20 @@ class TestConfirmGeneration:
         )
         gen_task_repo.create(preview)
 
-        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
-            resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
-                json={"output_width": 1920, "output_height": 1080},
-            )
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={"output_width": 1920, "output_height": 1080},
+        )
 
         assert resp.status_code == 200
         item = resp.json()["items"][0]
         assert item["is_preview"] is False
-        assert item["source_task_id"] == preview.id
         assert item["output_width"] == 1920
         assert item["output_height"] == 1080
-        # 默认封面和标题
+        # 默认封面和标题为空
         assert item["cover_url"] == ""
         assert item["custom_title"] == ""
-        # 复制的配置
+        # 配置保留
         assert item["voice_library_id"] == "voice-001"
         assert item["template_id"] == "tmpl-001"
         assert item["title_ids"] == ["title-1", "title-2"]
@@ -323,16 +372,15 @@ class TestConfirmGeneration:
         preview = _make_preview_task()
         gen_task_repo.create(preview)
 
-        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
-            resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
-                json={
-                    "output_width": 1080,
-                    "output_height": 1920,
-                    "cover_url": "https://cdn.example.com/my-cover.png",
-                    "custom_title": "测试视频标题",
-                },
-            )
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={
+                "output_width": 1080,
+                "output_height": 1920,
+                "cover_url": "https://cdn.example.com/my-cover.png",
+                "custom_title": "测试视频标题",
+            },
+        )
 
         assert resp.status_code == 200
         item = resp.json()["items"][0]
@@ -348,24 +396,23 @@ class TestConfirmGeneration:
         preview = _make_preview_task()
         gen_task_repo.create(preview)
 
-        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
-            resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
-                json={},
-            )
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={},
+        )
 
         assert resp.status_code == 200
         item = resp.json()["items"][0]
         assert item["output_width"] == 1080
         assert item["output_height"] == 1920
 
-    def test_confirm_creates_new_task_in_repo(
+    def test_confirm_skips_reuse_when_resolution_mismatch(
         self,
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ) -> None:
-        """确认生成的任务确实被存入 repository"""
-        preview = _make_preview_task()
+        """请求的分辨率与预览渲染的分辨率不一致时，跳过复用，创建新任务"""
+        preview = _make_preview_task(output_width=1080, output_height=1920)
         gen_task_repo.create(preview)
 
         initial_count = len(gen_task_repo._store)
@@ -373,15 +420,13 @@ class TestConfirmGeneration:
         with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
             resp = client.post(
                 f"/api/v1/tasks/{preview.id}/confirm",
-                json={},
+                json={"output_width": 720, "output_height": 1280},
             )
 
         assert resp.status_code == 200
-        new_task_id = resp.json()["items"][0]["id"]
-        assert new_task_id != preview.id
+        item = resp.json()["items"][0]
+        # 创建了新任务（而非复用）
+        assert item["id"] != preview.id
+        assert item["is_preview"] is False
+        assert item["source_task_id"] == preview.id
         assert len(gen_task_repo._store) == initial_count + 1
-
-        new_task = gen_task_repo.get(new_task_id)
-        assert new_task is not None
-        assert new_task.is_preview is False
-        assert new_task.source_task_id == preview.id

tests/unit/test_generation_preview.py

@@ -563,7 +563,6 @@ from unittest.mock import PropertyMock, patch
 
 # 导入路由模块和被测函数
 from app.api.routes.generation_preview import (
-    PREVIEW_RESOLUTION,
     _to_preview_response,
     create_preview_generation_task,
     get_preview_generation_task,
@@ -1133,35 +1132,6 @@ class TestSignVideoUrl:
         assert result == "https://bucket.oss.example.com/video.mp4"
 
 
-class TestCalcPreviewResolution:
-    """_calc_preview_resolution 单元测试。"""
-
-    def test_portrait_9_16(self):
-        from app.api.routes.generation_preview import _calc_preview_resolution
-
-        assert _calc_preview_resolution("9:16") == "480x854"
-
-    def test_landscape_16_9(self):
-        from app.api.routes.generation_preview import _calc_preview_resolution
-
-        assert _calc_preview_resolution("16:9") == "854x480"
-
-    def test_square_1_1(self):
-        from app.api.routes.generation_preview import _calc_preview_resolution
-
-        assert _calc_preview_resolution("1:1") == "480x480"
-
-    def test_unknown_defaults_to_landscape(self):
-        from app.api.routes.generation_preview import _calc_preview_resolution
-
-        assert _calc_preview_resolution("unknown") == "854x480"
-
-    def test_empty_defaults_to_landscape(self):
-        from app.api.routes.generation_preview import _calc_preview_resolution
-
-        assert _calc_preview_resolution("") == "854x480"
-
-
 class TestInferVideoRatioFromTemplate:
     """_infer_video_ratio_from_template 单元测试。"""
 
@@ -1239,7 +1209,7 @@ class TestPreviewRouteAutoInfersVideoRatio:
     """验证预览路由在前端未传 video_ratio 时自动从模板推断。"""
 
     def test_auto_infer_pip_resolution(self):
-        """前端传 video_ratio=''，模板 mode=pip → resolution=480x854"""
+        """前端传 video_ratio=''，模板 mode=pip → resolution 为空（使用用户默认导出配置）"""
         from app.api.routes.generation_preview import create_preview_generation_task
         from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
 
@@ -1284,7 +1254,7 @@ class TestPreviewRouteAutoInfersVideoRatio:
         # Verify the resolution passed to CreateGenerationTaskCommand
         call_args = MockUC.return_value.execute.call_args
         cmd = call_args[0][0]
-        assert cmd.resolution == "480x854", f"Expected 480x854, got {cmd.resolution}"
+        assert cmd.resolution == "", f"Expected empty resolution, got {cmd.resolution}"
 
 
 # ═══════════════════════════════════════════════════════════════════════════════

tests/unit/test_unified_render_reuse.py

@@ -0,0 +1,287 @@
+"""统一渲染路径 — 编辑器预览产物复用逻辑单元测试。
+
+覆盖：
+- _find_reusable_preview_task: 查找可复用的预览任务
+- _get_task_output_url: 获取任务输出 URL
+- 编辑器 generate 接口复用预览产物路径
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+# ── Stub Repository ──────────────────────────────────────────────────────────
+
+
+class StubGenTaskRepo:
+    def __init__(self):
+        self._store = {}
+
+    def create(self, task):
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id):
+        return self._store.get(task_id)
+
+    def update(self, task):
+        self._store[task.id] = task
+        return task
+
+    def list_by_source_edit_plan(self, plan_id):
+        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]
+
+
+def _make_task(**kwargs):
+    defaults = dict(
+        id="task-001",
+        project_id="proj-1",
+        asset_library_id="lib-1",
+        strategy_id="one_take",
+        voice_library_id="",
+        template_id="tmpl-1",
+        asset_ids=["a1"],
+        title_ids=[],
+        voice_ids=[],
+        status=GenerationTaskStatus.COMPLETED,
+        progress=100.0,
+        result_count=1,
+        error_message="",
+        created_by_user_id="user-1",
+        source_edit_plan_id="plan-1",
+        asset_select_mode="all",
+        is_preview=True,
+        source_task_id="",
+        output_width=1920,
+        output_height=1080,
+        cover_url="",
+        custom_title="",
+        video_title="",
+        resolution="",
+        bgm_config={},
+        completed_at=datetime.now(timezone.utc),
+    )
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+def _make_plan(updated_at=None):
+    plan = MagicMock()
+    plan.id = "plan-1"
+    plan.updated_at = updated_at or datetime.now(timezone.utc)
+    plan.status = MagicMock()
+    plan.status.value = "editing"
+    plan.config = {"clips": [{"id": "c1"}, {"id": "c2"}]}
+    return plan
+
+
+# ── _find_reusable_preview_task ──────────────────────────────────────────────
+
+
+class TestFindReusablePreviewTask:
+    def test_returns_completed_preview_task(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        now = datetime.now(timezone.utc)
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=True,
+            status=GenerationTaskStatus.COMPLETED,
+            completed_at=now - timedelta(minutes=5),
+        )
+        repo.create(task)
+
+        plan = _make_plan(updated_at=now - timedelta(minutes=10))
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+
+        assert result is not None
+        assert result.id == "task-001"
+
+    def test_returns_none_when_no_tasks(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        plan = _make_plan()
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None
+
+    def test_returns_none_when_preview_not_completed(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=True,
+            status=GenerationTaskStatus.RUNNING,
+            completed_at=None,
+        )
+        repo.create(task)
+
+        plan = _make_plan()
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None
+
+    def test_returns_none_when_plan_modified_after_preview(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        now = datetime.now(timezone.utc)
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=True,
+            status=GenerationTaskStatus.COMPLETED,
+            completed_at=now - timedelta(minutes=10),
+        )
+        repo.create(task)
+
+        # Plan was updated AFTER preview completed
+        plan = _make_plan(updated_at=now)
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None
+
+    def test_skips_non_preview_tasks(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=False,  # not a preview task
+            status=GenerationTaskStatus.COMPLETED,
+        )
+        repo.create(task)
+
+        plan = _make_plan()
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None
+
+    def test_handles_repo_exception(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = MagicMock()
+        repo.list_by_source_edit_plan.side_effect = Exception("db error")
+        plan = _make_plan()
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None
+
+
+# ── _get_task_output_url ─────────────────────────────────────────────────────
+
+
+class TestGetTaskOutputUrl:
+    def test_returns_video_url(self):
+        from app.api.routes.templates_editor.generation import _get_task_output_url
+
+        task = _make_task()
+        repo = MagicMock()
+        db = MagicMock()
+
+        mock_video = MagicMock()
+        mock_video.file_url = "https://oss.example.com/video.mp4"
+
+        mock_use_case = MagicMock()
+        mock_use_case.execute.return_value = [mock_video]
+
+        with patch(
+            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
+            return_value=mock_use_case,
+        ):
+            result = _get_task_output_url(task, repo, db)
+
+        assert result == "https://oss.example.com/video.mp4"
+
+    def test_returns_empty_when_no_videos(self):
+        from app.api.routes.templates_editor.generation import _get_task_output_url
+
+        task = _make_task()
+        repo = MagicMock()
+        db = MagicMock()
+
+        mock_use_case = MagicMock()
+        mock_use_case.execute.return_value = []
+
+        with patch(
+            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
+            return_value=mock_use_case,
+        ):
+            result = _get_task_output_url(task, repo, db)
+
+        assert result == ""
+
+    def test_returns_empty_on_exception(self):
+        from app.api.routes.templates_editor.generation import _get_task_output_url
+
+        task = _make_task()
+        repo = MagicMock()
+        db = MagicMock()
+
+        with patch(
+            "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase",
+            side_effect=Exception("db error"),
+        ):
+            result = _get_task_output_url(task, repo, db)
+
+        assert result == ""
+
+
+# ── mark_confirmed ───────────────────────────────────────────────────────────
+
+
+class TestMarkConfirmed:
+    def test_sets_is_preview_false(self):
+        task = _make_task(is_preview=True)
+        task.mark_confirmed()
+        assert task.is_preview is False
+
+    def test_sets_cover_url(self):
+        task = _make_task()
+        task.mark_confirmed(cover_url="https://example.com/cover.jpg")
+        assert task.cover_url == "https://example.com/cover.jpg"
+
+    def test_sets_custom_title(self):
+        task = _make_task()
+        task.mark_confirmed(custom_title="My Video")
+        assert task.custom_title == "My Video"
+
+    def test_sets_output_dimensions(self):
+        task = _make_task()
+        task.mark_confirmed(output_width=1080, output_height=1920)
+        assert task.output_width == 1080
+        assert task.output_height == 1920
+
+    def test_zero_dimensions_not_applied(self):
+        task = _make_task(output_width=1920, output_height=1080)
+        task.mark_confirmed(output_width=0, output_height=0)
+        assert task.output_width == 1920
+        assert task.output_height == 1080
+
+    def test_skips_when_plan_updated_at_is_none(self):
+        from app.api.routes.templates_editor.generation import _find_reusable_preview_task
+
+        repo = StubGenTaskRepo()
+        now = datetime.now(timezone.utc)
+        task = _make_task(
+            source_edit_plan_id="plan-1",
+            is_preview=True,
+            status=GenerationTaskStatus.COMPLETED,
+            completed_at=now - timedelta(minutes=5),
+        )
+        repo.create(task)
+
+        plan = _make_plan(updated_at=None)
+        result = _find_reusable_preview_task(repo, "plan-1", plan)
+        assert result is None