ci: retry build (transient BuildKit infrastructure error)

PreviewVideoPanel 的 displayTitle.split("/") 改为正则 split(/[/／]/)，
与 useCanvasPlayer.ts 保持一致，中文输入法下的全角斜杠也能触发换行。

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

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import json
 import logging
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -265,22 +264,38 @@ def create_preview_generation_task(
     if not video_ratio and request.template_id:
         video_ratio = _infer_video_ratio_from_template(request.template_id, db, user_id)
 
+    # 根据 video_ratio 计算输出分辨率（默认竖屏 1080x1920）
+    output_width, output_height = 1080, 1920
+    if video_ratio:
+        parts = video_ratio.split(":")
+        if len(parts) == 2:
+            try:
+                w, h = int(parts[0]), int(parts[1])
+                base = 1920
+                if w < h:
+                    # 竖屏
+                    output_width = round(base * w / h)
+                    output_height = base
+                else:
+                    # 横屏
+                    output_width = base
+                    output_height = round(base * h / w)
+                # 对齐到偶数
+                output_width = output_width - output_width % 2
+                output_height = output_height - output_height % 2
+            except (ValueError, ZeroDivisionError):
+                output_width, output_height = 1080, 1920
+    resolution = f"{output_width}x{output_height}"
+
+    logger.info(
+        "[预览生成] 分辨率: video_ratio=%s → %s (%dx%d)",
+        video_ratio, resolution, output_width, output_height,
+    )
+
     # 从模板读取 editing_mode / mode 作为 strategy_id（渲染 pipeline 的 mode 参数）
     strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
 
-    # 处理标题配置：如果有标题文本，序列化到 custom_title 字段传递给 worker
     title_config = request.title_config or {}
-    title_text = (title_config.get("text") or "").strip()
-    custom_title_value = ""
-    if title_text:
-        # 将标题文本和样式配置序列化为 JSON 存入 custom_title
-        # Worker 端会解析 JSON 获取完整标题配置
-        custom_title_value = json.dumps(title_config, ensure_ascii=False)
-        logger.info(
-            "[预览生成] 标题配置: text=%s, config_keys=%s",
-            title_text[:30],
-            list(title_config.keys()),
-        )
 
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
 
@@ -300,12 +315,14 @@ def create_preview_generation_task(
                 asset_select_mode="",
                 batch_id="",
                 video_title=request.video_title,
-                resolution="",
+                resolution=resolution,
                 bgm_config=request.bgm_config or {},
                 auto_retry_enabled=False,
                 auto_retry_max=0,
                 is_preview=True,
-                custom_title=custom_title_value,
+                title_config=title_config,
+                output_width=output_width,
+                output_height=output_height,
             )
         )
     except ValueError as e:

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

@@ -70,7 +70,6 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         output_width=getattr(task, "output_width", 1280),
         output_height=getattr(task, "output_height", 720),
         cover_url=getattr(task, "cover_url", ""),
-        custom_title=getattr(task, "custom_title", ""),
         title_config=getattr(task, "title_config", {}) or {},
         logs=getattr(task, "logs", "[]"),
         status=task.status,
@@ -294,6 +293,89 @@ def create_generation_task(
                     detail="当前项目没有符合条件的视频素材，请先上传并等待导入完成后再生成。",
                 )
 
+    # ── 兜底复用预览产物 ──
+    # 前端刷新后 previewTaskId 丢失，降级调 create 接口时，
+    # 如果同一 edit_plan 有已完成的预览任务，直接复用（秒出）。
+    if request.source_edit_plan_id and not request.is_preview:
+        try:
+            from packages.adapters.sqlalchemy_impl.models import (
+                GenerationTaskModel,
+            )
+
+            _preview_model = (
+                db.query(GenerationTaskModel)
+                .filter(
+                    GenerationTaskModel.source_edit_plan_id == request.source_edit_plan_id,
+                    GenerationTaskModel.is_preview.is_(True),
+                    GenerationTaskModel.status == "completed",
+                    GenerationTaskModel.created_by_user_id == authenticated_user.user.id,
+                )
+                .order_by(GenerationTaskModel.created_at.desc())
+                .first()
+            )
+            if _preview_model is not None:
+                # 校验分辨率一致性（与 confirm 端点逻辑相同）
+                req_w = request.output_width or 0
+                req_h = request.output_height or 0
+                src_w = getattr(_preview_model, "output_width", 0) or 0
+                src_h = getattr(_preview_model, "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:
+                    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                        _to_domain,
+                    )
+
+                    preview_task = _to_domain(_preview_model)
+
+                    # 如果传了标题，更新 title_config
+                    fallback_title_config = None
+                    if request.title_config and request.title_config.get("text", "").strip():
+                        fallback_title_config = dict(preview_task.title_config or {})
+                        fallback_title_config.update(request.title_config)
+
+                    preview_task.mark_confirmed(
+                        cover_url=request.cover_url or preview_task.cover_url,
+                        output_width=request.output_width or preview_task.output_width,
+                        output_height=request.output_height or preview_task.output_height,
+                        title_config=fallback_title_config,
+                    )
+                    generation_task_repository.update(preview_task)
+
+                    # 同步标题到 EditPlan.config
+                    if fallback_title_config:
+                        _writeback_edit_plan_config(
+                            plan_id=request.source_edit_plan_id,
+                            task_id=preview_task.id,
+                            title_config=fallback_title_config,
+                            db=db,
+                        )
+
+                    logger.info(
+                        "[生成任务] 兜底复用预览产物: preview_task_id=%s, plan_id=%s",
+                        preview_task.id,
+                        request.source_edit_plan_id,
+                    )
+                    return BatchGenerationTaskResponse(
+                        items=[_to_generation_task_response(preview_task)],
+                        total=1,
+                    )
+                else:
+                    logger.info(
+                        "[生成任务] 兜底复用跳过（分辨率不一致）: plan_id=%s, src=%sx%s, req=%sx%s",
+                        request.source_edit_plan_id,
+                        src_w,
+                        src_h,
+                        req_w,
+                        req_h,
+                    )
+        except Exception:
+            logger.warning(
+                "[生成任务] 兜底复用预览产物异常（不影响主流程）: plan_id=%s",
+                request.source_edit_plan_id,
+                exc_info=True,
+            )
+
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
     count = request.count
     created_tasks = []
@@ -355,11 +437,53 @@ def create_generation_task(
                     output_width=request.output_width,
                     output_height=request.output_height,
                     cover_url=request.cover_url,
-                    custom_title=request.custom_title,
                     title_config=request.title_config or {},
                 )
             )
             try:
+                # 兜底关联编辑计划：前端未传 source_edit_plan_id 时，
+                # 通过 template_id + user_id 在 DB 层直接查找最新的 plan。
+                # 必须在 enqueue 之前执行，避免 worker 读取时 source_edit_plan_id 为空（竞态条件）
+                if not task.source_edit_plan_id and request.template_id:
+                    try:
+                        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+                        _plan_model = (
+                            db.query(EditPlanModel)
+                            .filter(
+                                EditPlanModel.template_id == request.template_id,
+                                EditPlanModel.created_by_user_id == user_id,
+                            )
+                            .order_by(EditPlanModel.created_at.desc())
+                            .first()
+                        )
+                        if _plan_model:
+                            task.source_edit_plan_id = _plan_model.id
+                            generation_task_repository.update(task)
+                            logger.info(
+                                "[生成任务] 自动关联编辑计划: task_id=%s plan_id=%s",
+                                task.id,
+                                _plan_model.id,
+                            )
+                    except Exception:
+                        logger.warning(
+                            "[生成任务] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
+                            task.id,
+                            exc_info=True,
+                        )
+
+                # 回写 plan.config：必须在 enqueue 之前执行，
+                # 确保 worker 读取 plan 时 config 中已包含 generation_task_id。
+                # 只在首个任务时回写一次，避免批量生成时循环覆盖。
+                _effective_plan_id = task.source_edit_plan_id
+                if _effective_plan_id and len(created_tasks) == 0:
+                    _writeback_edit_plan_config(
+                        plan_id=_effective_plan_id,
+                        task_id=task.id,
+                        title_config=request.title_config,
+                        db=db,
+                    )
+
                 if safe_enqueue_generation_task(
                     task,
                     generation_task_repository,
@@ -368,15 +492,6 @@ def create_generation_task(
                     log_task_status=True,
                 ):
                     created_tasks.append(task)
-                    # 只在首个成功任务时回写一次 plan.config，
-                    # 避免批量生成时循环覆盖 generation_task_id
-                    if request.source_edit_plan_id and len(created_tasks) == 1:
-                        _writeback_edit_plan_config(
-                            plan_id=request.source_edit_plan_id,
-                            task_id=task.id,
-                            title_config=request.title_config,
-                            db=db,
-                        )
                 else:
                     failed_tasks.append(task)
             except UserPendingLimitExceeded as _e:
@@ -413,6 +528,7 @@ def confirm_generation(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     generation_task_repository: Any = Depends(get_generation_task_repository),
     project_repository: Any = Depends(get_project_repository),
+    db: Session = Depends(get_db_session),
 ) -> BatchGenerationTaskResponse:
     """确认生成 -- 复用预览渲染产物（预览与正式品质一致）。
 
@@ -441,13 +557,29 @@ def confirm_generation(
         resolution_match = (req_w == 0 or req_w == src_w) and (req_h == 0 or req_h == src_h)
 
         if resolution_match:
+            # 如果用户传了 custom_title，同步更新 title_config
+            confirmed_title_config = None
+            if request.custom_title and request.custom_title.strip():
+                confirmed_title_config = dict(getattr(source_task, "title_config", {}) or {})
+                confirmed_title_config["text"] = request.custom_title.strip()
+
             source_task.mark_confirmed(
                 cover_url=request.cover_url,
-                custom_title=request.custom_title,
                 output_width=request.output_width,
                 output_height=request.output_height,
+                title_config=confirmed_title_config,
             )
             generation_task_repository.update(source_task)
+
+            # 同步标题到 EditPlan.config
+            if confirmed_title_config and source_task.source_edit_plan_id:
+                _writeback_edit_plan_config(
+                    plan_id=source_task.source_edit_plan_id,
+                    task_id=source_task.id,
+                    title_config=confirmed_title_config,
+                    db=db,
+                )
+
             logger.info(
                 "[确认生成] 复用预览产物: task_id=%s, user_id=%s",
                 task_id,
@@ -489,7 +621,6 @@ def confirm_generation(
             output_width=request.output_width,
             output_height=request.output_height,
             cover_url=request.cover_url,
-            custom_title=request.custom_title,
         )
     )
 
@@ -622,7 +753,6 @@ def retry_generation_task(
             output_width=getattr(task, "output_width", 1280),
             output_height=getattr(task, "output_height", 720),
             cover_url=getattr(task, "cover_url", ""),
-            custom_title=getattr(task, "custom_title", ""),
         )
     )
     try:

apps/api/app/api/routes/tts.py

@@ -381,15 +381,34 @@ def preview_tts(
     request: TTSPreviewRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
+    voice_clone_repo=Depends(get_voice_clone_profile_repository),
 ) -> TTSPreviewResponse:
     """TTS 预览（试听）——同步合成，立即返回音频 URL。
 
     用于前端预览配音效果，限制文本长度 200 字以内。
+    支持预设音色和克隆音色：克隆音色传的是 profile UUID，需解析为 CosyVoice voice_id。
     """
+    # 解析 voice_id：前端可能传 VoiceCloneProfile UUID 或预设音色 ID
+    actual_voice_id = request.voice_id
+    profile = voice_clone_repo.get(request.voice_id)
+    if profile is not None:
+        # 命中克隆音色 profile — 校验归属权限
+        if profile.user_id != authenticated_user.user.id:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="无权访问该音色",
+            )
+        if not profile.voice_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="音色克隆尚未完成，请稍后再试",
+            )
+        actual_voice_id = profile.voice_id
+
     try:
         result = cosyvoice_service.synthesize_speech(
             text=request.text,
-            voice_id=request.voice_id,
+            voice_id=actual_voice_id,
             speed=request.speed,
         )
     except CosyVoiceError as e:

apps/api/app/schemas/generation_task.py

@@ -10,7 +10,7 @@ class ConfirmGenerationRequest(BaseModel):
     output_width: int = Field(default=1080, ge=100, description="输出视频宽度")
     output_height: int = Field(default=1920, ge=100, description="输出视频高度")
     cover_url: str = Field(default="", description="自定义封面图片 URL")
-    custom_title: str = Field(default="", description="自定义视频标题")
+    custom_title: str = Field(default="", description="用户自定义标题文本，非空时同步到任务和编辑计划")
 
 
 class CreateGenerationTaskRequest(BaseModel):
@@ -33,7 +33,7 @@ class CreateGenerationTaskRequest(BaseModel):
     voice_ids: list[str] = Field(default_factory=list)
     # ── 来源剪辑计划 ──
     source_edit_plan_id: str = ""
-    # ── 标题配置（结构化，优先于 custom_title 纯文本）──
+    # ── 标题配置（结构化）──
     title_config: dict | None = Field(
         default=None,
         description="标题样式对象，包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。",
@@ -77,7 +77,6 @@ class CreateGenerationTaskRequest(BaseModel):
     output_width: int = Field(default=1280, description="输出视频宽度")
     output_height: int = Field(default=720, description="输出视频高度")
     cover_url: str = Field(default="", description="封面图片 URL")
-    custom_title: str = Field(default="", description="自定义视频标题")
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -113,7 +112,6 @@ class GenerationTaskResponse(BaseModel):
     output_width: int = 1280
     output_height: int = 720
     cover_url: str = ""
-    custom_title: str = ""
     title_config: dict = Field(default_factory=dict)
     status: str
     progress: float

apps/api/app/services/edit_plan_service.py

@@ -423,6 +423,9 @@ class EditPlanService:
                 )
                 db.add(model)
 
+            # flush 让新建 clip 写入当前事务（未 commit），后续查询才能找到它们
+            db.flush()
+
             # 3. 标记有 asset_id 的 clips 为 ready（不 commit）
             pending_with_asset = (
                 db.query(EditPlanClipModel)

apps/web/src/api/generation/types.ts

@@ -13,6 +13,10 @@ export interface CreatePreviewRequest {
   video_title?: string
   duration?: number
   video_ratio?: string
+  /** 输出视频宽度（与 video_ratio 匹配，如 9:16 → 1080） */
+  output_width?: number
+  /** 输出视频高度（与 video_ratio 匹配，如 9:16 → 1920） */
+  output_height?: number
   /* 标题烧录配置（可选，传入后 ASS 渲染标题到预览视频中） */
   title_config?: {
     text?: string
@@ -38,6 +42,8 @@ export interface CreatePreviewResponse {
   is_preview: boolean
   resolution: string
   created_at: string
+  /** 后端自动关联的编辑计划 ID（用于 fallback 路径传递 source_edit_plan_id） */
+  source_edit_plan_id?: string
 }
 
 /** 预览任务详情响应 */

apps/web/src/api/tasks/types.ts

@@ -84,6 +84,14 @@ export interface CreateGenerationTaskRequest {
   }
   /** 关联的草稿 ID（编辑流程数据链路用） */
   source_edit_plan_id?: string
+  /** 配音素材库 ID（用户上传的音频或 AI 配音素材） */
+  voice_library_id?: string
+  /** 自定义 BGM 配置，覆盖模板 BGM 设置 */
+  bgm_config?: {
+    enabled: boolean
+    preset_id?: string
+    volume?: number
+  }
 }
 
 /** 单个生成任务详情（对齐后端 GenerationTaskResponse） */

apps/web/src/pages/generate/GeneratePage.tsx

@@ -1,13 +1,12 @@
 /**
- * 智能剪辑页面 — V24 前端预览播放器架构改造
+ * 智能剪辑页面 — 前端实时预览架构
  * 7 步向导：选择模板 → 素材 → 配音 → 标题 → 预览 → 封面 → 确认生成
  * 左右布局：左侧 generate-form + 右侧 generate-preview
  *
- * 架构改造：
- * - Step5 预览改为前端素材切片播放（FrontendPreviewPlayer）
- * - 完全去除后端 FFmpeg 预览依赖
- * - 标题样式通过 CSS 层实时叠加，所见即所得
- * - 最终成片仍走后端 FFmpeg 渲染（Step7 确认生成）
+ * 架构：
+ * - Step4+ 右侧预览面板使用 FrontendPreviewPlayer 实时播放素材片段
+ * - 标题样式编辑时 CSS 层实时叠加预览，所见即所得
+ * - 点"确认生成"时调用 createGenerationTask 创建一次服务器渲染任务
  */
 import React, { useMemo, useState, useEffect, useRef } from "react"
 import { Modal, message } from "antd"
@@ -15,8 +14,6 @@ import { useNavigate } from "react-router-dom"
 import type { VoiceClone } from "@/api/voice-clone"
 import { useQuery } from "@tanstack/react-query"
 import { useCloneProgress } from "@/hooks/useCloneProgress"
-import { getAssetsByKind } from "@/api/assets"
-import { previewTts } from "@/api/tts"
 import CloneModal from "@/components/voice/CloneModal"
 import GenerateHeader from "./components/GenerateHeader"
 import {
@@ -25,7 +22,7 @@ import {
 } from "./utils/calculateTotalVideoDuration"
 import GenerateStepsBar from "./components/GenerateStepsBar"
 import GenerateResultPanel from "./components/GenerateResultPanel"
-import PreviewVideoPanel from "./components/PreviewVideoPanel"
+import FrontendPreviewPlayer from "./components/FrontendPreviewPlayer"
 import GenerateStepContent from "./components/GenerateStepContent"
 import GenerateStepActions from "./components/GenerateStepActions"
 import { useGenerateFormState } from "./hooks/useGenerateFormState"
@@ -33,6 +30,8 @@ import { useStepNavigation } from "./hooks/useStepNavigation"
 import { useGenerateVideo } from "./hooks/useGenerateVideo"
 import { usePreviewAssets } from "./hooks/usePreviewAssets"
 import { useTitleStyleUpdaters } from "./hooks/useStep4Title/useTitleStyleUpdaters"
+import { getAssetsByKind } from "@/api/assets"
+import { previewTts } from "@/api/tts"
 import "./generate.css"
 
 const GeneratePage: React.FC = () => {
@@ -78,50 +77,19 @@ const GeneratePage: React.FC = () => {
     setPreviewVideo,
     previewModalOpen,
     setPreviewModalOpen,
+    previewTaskId,
+    setPreviewTaskId,
+    storedSourceEditPlanId,
+    setStoredSourceEditPlanId,
   } = formState
 
-  /* ── 标题样式回调（Step5 样式面板 + 右侧预览 CSS 层共用） ── */
+  /* ── 标题样式回调 ── */
   const styleUpdaters = useTitleStyleUpdaters({
     titleSettings,
     onTitleSettingsChange: setTitleSettings,
   })
 
-  /* ── 克隆声音 ── */
-  const { clones: clonedVoices, addClone, hasProcessing } = useCloneProgress()
-
-  const handleCloneSuccess = (voice: VoiceClone) => {
-    addClone(voice)
-    setCloneModalOpen(false)
-    message.success("音色克隆成功！")
-  }
-
-  /* ── 前端预览：加载选中素材的视频文件信息 ── */
-  const previewAssetIds = useMemo(
-    () => (materialMode === "auto" ? smartSelectedIds : selectedMaterials),
-    [materialMode, smartSelectedIds, selectedMaterials],
-  )
-  const previewAssetsEnabled = currentStep >= 4 && previewAssetIds.length > 0
-  const {
-    assets: previewAssets,
-    loading: previewAssetsLoading,
-    ready: previewAssetsReady,
-  } = usePreviewAssets(previewAssetIds, previewAssetsEnabled)
-
-  /* ── 当前模板对象（传给前端预览播放器） ── */
-  const currentTemplate = useMemo(
-    () => userTemplates.find((t) => t.id === selectedTemplate) || null,
-    [userTemplates, selectedTemplate],
-  )
-
-  /* ── 视频总时长计算（用于配音时长校验） ── */
-  const totalVideoDuration = useMemo(() => {
-    // 优先用素材精确时长；素材未加载时用模板 segments 的 duration_max 之和估算
-    const exact = calculateTotalVideoDuration(previewAssets, currentTemplate ?? undefined)
-    if (exact > 0) return exact
-    return estimateTotalVideoDuration(currentTemplate ?? undefined)
-  }, [previewAssets, currentTemplate])
-
-  /* ── 配音音频 URL ── */
+  /* ── 配音预览音频（TTS 试听）── */
   const { data: voiceMaterials = [] } = useQuery({
     queryKey: ["assets", "voice"],
     queryFn: () => getAssetsByKind("voice", { limit: 50 }),
@@ -171,9 +139,49 @@ const GeneratePage: React.FC = () => {
       cancelled = true
       controller.abort()
     }
+    // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [selectedVoice, selectedClonedVoice, titleSettings.title, voiceMaterials])
 
-  const voiceAudioUrl = previewVoiceAudioUrl || undefined
+  /* ── 克隆声音 ── */
+  const { clones: clonedVoices, addClone, hasProcessing } = useCloneProgress()
+
+  const handleCloneSuccess = (voice: VoiceClone) => {
+    addClone(voice)
+    setCloneModalOpen(false)
+    message.success("音色克隆成功！")
+  }
+
+  /* ── 素材 ID 列表 ── */
+  const previewAssetIds = useMemo(
+    () => (materialMode === "auto" ? smartSelectedIds : selectedMaterials),
+    [materialMode, smartSelectedIds, selectedMaterials],
+  )
+
+  /* ── 当前模板对象 ── */
+  const currentTemplate = useMemo(
+    () => userTemplates.find((t) => t.id === selectedTemplate) || null,
+    [userTemplates, selectedTemplate],
+  )
+
+  /* ── BGM 配置 ── */
+  const bgmConfig = useMemo(
+    () => ({
+      enabled: bgm,
+      music_id: currentTemplate?.bgm_config?.music_id || "",
+    }),
+    [bgm, currentTemplate],
+  )
+
+  /* ── 加载素材详情（供前端预览播放器使用 + 配音时长校验） ── */
+  const previewAssetsEnabled = previewAssetIds.length > 0
+  const { assets: previewAssets } = usePreviewAssets(previewAssetIds, previewAssetsEnabled)
+
+  /* ── 视频总时长计算 ── */
+  const totalVideoDuration = useMemo(() => {
+    const exact = calculateTotalVideoDuration(previewAssets, currentTemplate ?? undefined)
+    if (exact > 0) return exact
+    return estimateTotalVideoDuration(currentTemplate ?? undefined)
+  }, [previewAssets, currentTemplate])
 
   /* ── 步骤导航 ── */
   const { goNext, goPrev } = useStepNavigation({
@@ -184,7 +192,6 @@ const GeneratePage: React.FC = () => {
     selectedMaterials,
     smartSelectedIds,
     titleSettings,
-    previewReady: previewAssetsReady,
   })
 
   /* ── 视频生成核心逻辑 ── */
@@ -215,22 +222,25 @@ const GeneratePage: React.FC = () => {
     autoSubtitles,
     bgm,
     generateCount,
-    sourceEditPlanId,
+    sourceEditPlanId: storedSourceEditPlanId || sourceEditPlanId,
+    previewTaskId,
+    bgmConfig,
+    onGenerationSuccess: () => {
+      setPreviewTaskId(null)
+      setStoredSourceEditPlanId(null)
+    },
   })
 
   /* ================================================================
-     渲染 — 主页面
+     渲染
      ================================================================ */
 
   return (
     <div className="xx-generate-page">
-      {/* ── 页头 ── */}
       <GenerateHeader fromEditPlan={!!editPlanId} />
 
-      {/* ── 步骤条 ── */}
       <GenerateStepsBar currentStep={currentStep} onStepClick={setCurrentStep} />
 
-      {/* ── 主布局 ── */}
       <div className="xx-generate-layout">
         {/* ════ 左侧：表单区 ════ */}
         <div className="xx-generate-form">
@@ -247,7 +257,6 @@ const GeneratePage: React.FC = () => {
             onSmartSelectedIdsChange={setSmartSelectedIds}
             titleSettings={titleSettings}
             onTitleSettingsChange={setTitleSettings}
-            /* 标题样式回调 */
             onUpdatePosition={styleUpdaters.updatePosition}
             onUpdateFont={styleUpdaters.updateFont}
             onUpdateSize={styleUpdaters.updateSize}
@@ -258,6 +267,10 @@ const GeneratePage: React.FC = () => {
             onApplyPreset={styleUpdaters.applyPreset}
             activePreset={styleUpdaters.activePreset}
             titlePresets={styleUpdaters.titlePresets}
+            onPreviewTaskCreated={setPreviewTaskId}
+            onSourceEditPlanIdExtracted={setStoredSourceEditPlanId}
+            bgm={bgm}
+            bgmConfig={bgmConfig}
             coverSettings={coverSettings}
             onCoverSettingsChange={setCoverSettings}
             duration={duration}
@@ -296,18 +309,26 @@ const GeneratePage: React.FC = () => {
           />
         </div>
 
-        {/* ════ 右侧：预览 + 生成结果 ════ */}
+        {/* ════ 右侧：预览 + 结果 ════ */}
         <div className="xx-generate-right-col">
-          {/* 预览视频面板（Step4+ 显示，含 CSS 标题实时预览层） */}
-          {currentStep >= 4 && (
-            <PreviewVideoPanel
+          {currentStep >= 4 && !!currentTemplate && (
+            <FrontendPreviewPlayer
               assets={previewAssets}
               template={currentTemplate}
               videoRatio={videoRatio}
-              assetsReady={previewAssetsReady}
-              assetsLoading={previewAssetsLoading}
-              titleSettings={titleSettings}
-              voiceAudioUrl={voiceAudioUrl}
+              ready={previewAssets.length > 0}
+              voiceAudioUrl={previewVoiceAudioUrl || undefined}
+              titleSettings={{
+                title: titleSettings.title,
+                size: titleSettings.size,
+                font: titleSettings.font,
+                color: titleSettings.color,
+                position: titleSettings.position as "top" | "center" | "bottom",
+                bold: titleSettings.bold,
+                italic: titleSettings.italic,
+                stroke: titleSettings.stroke,
+                shadow: titleSettings.shadow,
+              }}
             />
           )}
           {currentStep >= 6 && (
@@ -329,7 +350,7 @@ const GeneratePage: React.FC = () => {
         </div>
       </div>
 
-      {/* ── 视频预览弹窗 ── */}
+      {/* 视频预览弹窗 */}
       <Modal
         className="xx-preview-modal"
         open={previewModalOpen}
@@ -352,7 +373,7 @@ const GeneratePage: React.FC = () => {
         )}
       </Modal>
 
-      {/* ── 音色克隆弹窗 ── */}
+      {/* 音色克隆弹窗 */}
       <CloneModal
         open={cloneModalOpen}
         onClose={() => setCloneModalOpen(false)}

apps/web/src/pages/generate/components/FrontendPreviewPlayer.tsx

@@ -277,20 +277,29 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
   if (!ready || !assets.length) {
     return (
       <div
-        className="xx-preview-empty"
         style={{
-          position: "absolute",
-          inset: 0,
+          position: "relative",
+          width: "100%",
+          maxWidth: 280,
+          aspectRatio: "9 / 16",
+          background: "#0a0a0a",
+          borderRadius: 24,
+          overflow: "hidden",
+          boxShadow:
+            "0 4px 6px -1px rgba(0,0,0,0.3), 0 20px 50px -12px rgba(0,0,0,0.5), inset 0 0 0 1px rgba(255,255,255,0.06)",
           display: "flex",
           flexDirection: "column",
           alignItems: "center",
           justifyContent: "center",
-          zIndex: 1,
         }}
       >
-        <SoundOutlined style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }} />
-        <p className="xx-preview-empty-title">准备预览素材...</p>
-        <p className="xx-preview-empty-desc">加载素材后即可预览播放</p>
+        <SoundOutlined style={{ fontSize: 40, color: "rgba(255,255,255,0.3)", marginBottom: 12 }} />
+        <p style={{ color: "rgba(255,255,255,0.6)", fontSize: 14, margin: "0 0 4px" }}>
+          准备预览素材...
+        </p>
+        <p style={{ color: "rgba(255,255,255,0.35)", fontSize: 12, margin: 0 }}>
+          加载素材后即可预览播放
+        </p>
       </div>
     )
   }
@@ -300,31 +309,48 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
     const showDecodeError = forceVideoFallback && canvasState.hasDecodeError
     return (
       <div
-        className="xx-preview-empty"
         style={{
-          position: "absolute",
-          inset: 0,
+          position: "relative",
+          width: "100%",
+          maxWidth: 280,
+          aspectRatio: "9 / 16",
+          background: "#0a0a0a",
+          borderRadius: 24,
+          overflow: "hidden",
+          boxShadow:
+            "0 4px 6px -1px rgba(0,0,0,0.3), 0 20px 50px -12px rgba(0,0,0,0.5), inset 0 0 0 1px rgba(255,255,255,0.06)",
           display: "flex",
           flexDirection: "column",
           alignItems: "center",
           justifyContent: "center",
-          zIndex: 1,
+          padding: 24,
         }}
       >
         {isBuffering ? (
           <>
-            <LoadingOutlined style={{ fontSize: 48, color: "#fff", marginBottom: 12 }} spin />
-            <p style={{ color: "rgba(255,255,255,0.8)" }}>加载中...</p>
+            <LoadingOutlined style={{ fontSize: 40, color: "#fff", marginBottom: 12 }} spin />
+            <p style={{ color: "rgba(255,255,255,0.8)", fontSize: 14, margin: 0 }}>加载中...</p>
           </>
         ) : showDecodeError ? (
           <>
-            <PlayCircleOutlined style={{ fontSize: 48, color: "#ef4444", marginBottom: 12 }} />
-            <p className="xx-preview-empty-title" style={{ color: "rgba(255,255,255,0.9)" }}>
+            <PlayCircleOutlined style={{ fontSize: 40, color: "#ef4444", marginBottom: 12 }} />
+            <p
+              style={{
+                color: "rgba(255,255,255,0.9)",
+                fontSize: 14,
+                margin: "0 0 4px",
+                fontWeight: 500,
+              }}
+            >
               视频解码失败
             </p>
             <p
-              className="xx-preview-empty-desc"
-              style={{ color: "rgba(255,255,255,0.6)", maxWidth: 300, textAlign: "center" }}
+              style={{
+                color: "rgba(255,255,255,0.5)",
+                fontSize: 12,
+                margin: 0,
+                textAlign: "center",
+              }}
             >
               {canvasState.errorMessage || "当前浏览器不支持该视频编码格式，请刷新重试"}
             </p>
@@ -332,10 +358,14 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
         ) : (
           <>
             <PlayCircleOutlined
-              style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
+              style={{ fontSize: 40, color: "rgba(255,255,255,0.3)", marginBottom: 12 }}
             />
-            <p className="xx-preview-empty-title">暂无可播放素材</p>
-            <p className="xx-preview-empty-desc">请先在左侧选择素材</p>
+            <p style={{ color: "rgba(255,255,255,0.6)", fontSize: 14, margin: "0 0 4px" }}>
+              暂无可播放素材
+            </p>
+            <p style={{ color: "rgba(255,255,255,0.35)", fontSize: 12, margin: 0 }}>
+              请先在左侧选择素材
+            </p>
           </>
         )}
       </div>
@@ -343,7 +373,19 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
   }
 
   return (
-    <>
+    <div
+      style={{
+        position: "relative",
+        width: "100%",
+        maxWidth: 280,
+        aspectRatio: "9 / 16",
+        background: "#0a0a0a",
+        borderRadius: 24,
+        overflow: "hidden",
+        boxShadow:
+          "0 4px 6px -1px rgba(0,0,0,0.3), 0 20px 50px -12px rgba(0,0,0,0.5), inset 0 0 0 1px rgba(255,255,255,0.06)",
+      }}
+    >
       {/* ── Canvas 渲染层（WebCodecs 路径） ── */}
       {effectiveUseWebCodecs && (
         <div
@@ -392,54 +434,114 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
           />
         ))}
 
-      {/* 播放按钮 */}
+      {/* 标题CSS叠加层 — video fallback 路径也要渲染 */}
+      {titleSettings?.title && (
+        <div
+          style={{
+            position: "absolute",
+            left: 0,
+            right: 0,
+            zIndex: 5,
+            display: "flex",
+            justifyContent: "center",
+            pointerEvents: "none",
+            ...(titleSettings.position === "top"
+              ? { top: "10%" }
+              : titleSettings.position === "center"
+                ? { top: "50%", transform: "translateY(-50%)" }
+                : { bottom: "15%" }),
+          }}
+        >
+          <span
+            style={{
+              fontSize: titleSettings.size,
+              fontFamily: titleSettings.font || "思源黑体",
+              color: titleSettings.color || "#ffffff",
+              fontWeight: titleSettings.bold ? 700 : 400,
+              fontStyle: titleSettings.italic ? "italic" : "normal",
+              textShadow: [
+                titleSettings.shadow ? "0 2px 8px rgba(0,0,0,0.7)" : undefined,
+                titleSettings.stroke
+                  ? "1px 1px 0 rgba(0,0,0,0.5), -1px -1px 0 rgba(0,0,0,0.5), 1px -1px 0 rgba(0,0,0,0.5), -1px 1px 0 rgba(0,0,0,0.5)"
+                  : undefined,
+                "0 1px 3px rgba(0,0,0,0.4)",
+              ]
+                .filter(Boolean)
+                .join(", "),
+              maxWidth: "90%",
+              textAlign: "center",
+              lineHeight: 1.3,
+              wordBreak: "break-word",
+            }}
+          >
+            {titleSettings.title}
+          </span>
+        </div>
+      )}
+
+      {/* 中央播放按钮 */}
       {!isPlaying && (
         <button
-          className="xx-preview-play-btn"
           onClick={handleTogglePlay}
           style={{
             position: "absolute",
             top: "50%",
             left: "50%",
             transform: "translate(-50%, -50%)",
-            background: "rgba(0,0,0,0.5)",
-            border: "none",
+            background: "rgba(0,0,0,0.45)",
+            backdropFilter: "blur(12px)",
+            WebkitBackdropFilter: "blur(12px)",
+            border: "1px solid rgba(255,255,255,0.15)",
             borderRadius: "50%",
-            width: 56,
-            height: 56,
+            width: 52,
+            height: 52,
             cursor: "pointer",
             display: "flex",
             alignItems: "center",
             justifyContent: "center",
             color: "#fff",
-            fontSize: 28,
+            fontSize: 26,
             zIndex: 10,
+            transition: "transform 0.2s ease, background 0.2s ease",
+            boxShadow: "0 4px 20px rgba(0,0,0,0.4)",
+          }}
+          onMouseEnter={(e) => {
+            e.currentTarget.style.transform = "translate(-50%, -50%) scale(1.08)"
+            e.currentTarget.style.background = "rgba(0,0,0,0.6)"
+          }}
+          onMouseLeave={(e) => {
+            e.currentTarget.style.transform = "translate(-50%, -50%) scale(1)"
+            e.currentTarget.style.background = "rgba(0,0,0,0.45)"
           }}
         >
           <PlayCircleOutlined />
         </button>
       )}
 
-      {/* 片段指示器 */}
+      {/* 片段指示器 — 右上角胶囊 */}
       <div
         style={{
           position: "absolute",
           top: 8,
-          left: 8,
-          background: "rgba(0,0,0,0.6)",
-          color: "#fff",
-          fontSize: 11,
+          right: 8,
+          background: "rgba(0,0,0,0.45)",
+          backdropFilter: "blur(8px)",
+          WebkitBackdropFilter: "blur(8px)",
+          color: "rgba(255,255,255,0.9)",
+          fontSize: 10,
+          fontWeight: 500,
           padding: "2px 8px",
-          borderRadius: 4,
+          borderRadius: 999,
           zIndex: 10,
+          border: "1px solid rgba(255,255,255,0.1)",
+          letterSpacing: 0.3,
         }}
       >
-        {`片段 ${videoCurrentSegIdx + 1}/${segments.length}`}
+        {`${videoCurrentSegIdx + 1} / ${segments.length}`}
       </div>
 
-      {/* 控制条 */}
+      {/* 控制条 — 手机风格毛玻璃 */}
       <div
-        className="xx-preview-controls"
         style={{
           position: "absolute",
           bottom: 0,
@@ -447,23 +549,36 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
           right: 0,
           display: "flex",
           alignItems: "center",
-          gap: 8,
-          padding: "8px 12px",
-          background: "linear-gradient(transparent, rgba(0,0,0,0.6))",
+          gap: 10,
+          padding: "12px 16px 16px",
+          background: "linear-gradient(transparent, rgba(0,0,0,0.7))",
+          backdropFilter: "blur(4px)",
+          WebkitBackdropFilter: "blur(4px)",
           zIndex: 10,
         }}
       >
         <button
           onClick={handleTogglePlay}
           style={{
-            background: "none",
+            background: "rgba(255,255,255,0.15)",
             border: "none",
             color: "#fff",
-            fontSize: 18,
+            fontSize: 16,
             cursor: "pointer",
-            padding: 4,
+            width: 32,
+            height: 32,
+            borderRadius: "50%",
             display: "flex",
             alignItems: "center",
+            justifyContent: "center",
+            flexShrink: 0,
+            transition: "background 0.15s",
+          }}
+          onMouseEnter={(e) => {
+            e.currentTarget.style.background = "rgba(255,255,255,0.25)"
+          }}
+          onMouseLeave={(e) => {
+            e.currentTarget.style.background = "rgba(255,255,255,0.15)"
           }}
         >
           {isPlaying ? <PauseCircleOutlined /> : <PlayCircleOutlined />}
@@ -471,10 +586,11 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
 
         <span
           style={{
-            fontSize: 12,
-            color: "rgba(255,255,255,0.8)",
-            minWidth: 80,
+            fontSize: 11,
+            color: "rgba(255,255,255,0.85)",
+            minWidth: 72,
             fontVariantNumeric: "tabular-nums",
+            letterSpacing: 0.2,
           }}
         >
           {formatTime(currentTime)} / {formatTime(totalDuration)}
@@ -485,8 +601,8 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
           onMouseDown={handleMouseDown}
           style={{
             flex: 1,
-            height: 4,
-            background: "rgba(255,255,255,0.15)",
+            height: 3,
+            background: "rgba(255,255,255,0.2)",
             borderRadius: 2,
             cursor: "pointer",
             position: "relative",
@@ -496,7 +612,7 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
             style={{
               height: "100%",
               width: `${progressPercent}%`,
-              background: "#3b82f6",
+              background: "#fff",
               borderRadius: 2,
               transition: isDragging ? "none" : "width 0.1s linear",
             }}
@@ -510,15 +626,15 @@ const FrontendPreviewPlayer: React.FC<FrontendPreviewPlayerProps> = ({
               width: 10,
               height: 10,
               borderRadius: "50%",
-              background: "#3b82f6",
-              border: "2px solid #fff",
+              background: "#fff",
+              boxShadow: "0 0 6px rgba(255,255,255,0.5)",
               opacity: isDragging ? 1 : 0,
               transition: "opacity 0.15s",
             }}
           />
         </div>
       </div>
-    </>
+    </div>
   )
 }
 

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -76,6 +76,14 @@ export interface GenerateStepContentProps {
   onDismissError: () => void
   /* 其他 */
   presetVoices: PresetVoiceItem[]
+  /** 预览任务创建回调——传递给 Step6CoverSettings */
+  onPreviewTaskCreated?: (taskId: string) => void
+  /** 从预览响应中提取到 source_edit_plan_id 时的回调 */
+  onSourceEditPlanIdExtracted?: (planId: string) => void
+  /** BGM 开关 */
+  bgm: boolean
+  /** BGM 配置（来自模板） */
+  bgmConfig?: { enabled: boolean; music_id?: string }
 }
 
 export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) => {
@@ -121,6 +129,10 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
     onRetry,
     onDismissError,
     presetVoices,
+    onPreviewTaskCreated,
+    onSourceEditPlanIdExtracted,
+    bgm,
+    bgmConfig,
   } = props
 
   /* 当前模板的 segments，传给 Step2 构建 clips */
@@ -190,6 +202,13 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           assetIds={materialMode === "auto" ? smartSelectedIds : selectedMaterials}
           selectedTemplate={selectedTemplate}
           titleSettings={titleSettings}
+          onPreviewTaskCreated={onPreviewTaskCreated}
+          onSourceEditPlanIdExtracted={onSourceEditPlanIdExtracted}
+          voiceMode={voiceMode}
+          selectedVoice={selectedVoice}
+          selectedClonedVoice={selectedClonedVoice}
+          bgm={bgm}
+          bgmConfig={bgmConfig}
         />
       )
     case 7:

apps/web/src/pages/generate/components/PreviewVideoPanel.tsx

@@ -1,39 +1,42 @@
 /**
- * 右侧预览视频面板
- * Step4+: 显示预览视频面板
- * Step5: 前端实时预览 — 用原生 video 播放素材片段 + CSS 标题叠加
+ * 右侧预览视频面板 — 服务器渲染预览架构
  *
- * 架构改造：完全去除后端 FFmpeg 预览依赖
- * - 使用 FrontendPreviewPlayer 直接播放素材片段
- * - TitleOverlay CSS 层实时响应标题样式变化
+ * Step4+: 显示预览面板
+ * Step5: 播放服务器渲染的真实视频（POST /generation/preview）
  *
- * 布局：本组件提供 .xx-preview-video 容器（position: relative + overflow: hidden）
- * FrontendPreviewPlayer 的内容通过 absolute 定位填充容器
- * TitleOverlay 通过 absolute 定位 + z-index: 30 覆盖在最上层
+ * 架构：
+ * - 进入 Step4/5 时自动创建服务器预览渲染任务
+ * - 轮询完成后用 <video> 标签播放返回的 video_url
+ * - 标题样式编辑时 CSS TitleOverlay 实时叠加预览
+ * - 素材/配音/BGM 变更自动重新渲染
+ * - 标题文字/样式变更标记 stale，保留旧视频 + 显示"重新预览"按钮
+ *
+ * 点"确认生成"时走 confirm 路径，成品就是预览视频本身，100% 一致。
  */
 import React, { useMemo, useRef, useState, useEffect } from "react"
-import { LoadingOutlined } from "@ant-design/icons"
-import type { AssetItem } from "@/api/assets"
-import type { EditingTemplate } from "@/api/editing-planner"
+import { LoadingOutlined, ReloadOutlined, ExclamationCircleOutlined } from "@ant-design/icons"
+import { Button } from "antd"
 import type { TitleSettings } from "../types"
 import { getFontFamily } from "../constants"
-import FrontendPreviewPlayer from "./FrontendPreviewPlayer"
+import type { ServerPreviewStatus } from "../hooks/useServerPreview"
 
 interface PreviewVideoPanelProps {
-  /** 已加载的素材列表 */
-  assets: AssetItem[]
-  /** 当前模板 */
-  template: EditingTemplate | null
+  /** 服务器预览状态 */
+  previewStatus: ServerPreviewStatus
+  /** 服务器渲染视频 URL */
+  videoUrl: string | null
+  /** 渲染进度 0-100 */
+  progress: number
+  /** 错误信息 */
+  error: string | null
+  /** 重新预览回调 */
+  onRetry: () => void
   /** 视频比例 */
   videoRatio: string
-  /** 素材是否已加载就绪 */
-  assetsReady: boolean
-  /** 素材是否正在加载 */
-  assetsLoading: boolean
-  /** 标题设置 — 用于 CSS 实时预览层 */
+  /** 标题设置 — CSS 实时预览层 */
   titleSettings?: TitleSettings
-  /** 配音音频 URL */
-  voiceAudioUrl?: string
+  /** 素材数量 */
+  assetCount?: number
 }
 
 /* ── ASS 坐标系参数（与后端 ass_subtitle_builder.py 一致） ── */
@@ -42,13 +45,8 @@ const ASS_TITLE_MARGIN_TOP = 60
 const ASS_TITLE_MARGIN_BOTTOM = 60
 const ASS_TITLE_MARGIN_SIDE = 40
 
-/**
- * 根据 position 计算 CSS 垂直定位
- * 与后端 position_to_ass_alignment() 对齐：top→8, center→5, bottom→2
- */
 function getPositionStyle(position: string): React.CSSProperties {
   const sidePercent = (ASS_TITLE_MARGIN_SIDE / 1280) * 100
-
   switch (position) {
     case "bottom":
       return {
@@ -76,16 +74,11 @@ function getPositionStyle(position: string): React.CSSProperties {
   }
 }
 
-/**
- * 构建 CSS 标题层的样式
- * 所有渲染参数与后端 FFmpeg ASS 字幕一致
- */
 function buildTitleStyle(settings: TitleSettings, containerHeight: number): React.CSSProperties {
-  // 用 px 计算 fontSize，不再依赖父元素 font-size 的百分比
   const fontSizePx =
     containerHeight > 0
       ? (Math.min(settings.size, 96) / ASS_VIDEO_HEIGHT) * containerHeight
-      : (Math.min(settings.size, 96) / ASS_VIDEO_HEIGHT) * 400 // fallback
+      : (Math.min(settings.size, 96) / ASS_VIDEO_HEIGHT) * 400
 
   const base: React.CSSProperties = {
     fontFamily: getFontFamily(settings.font),
@@ -100,28 +93,16 @@ function buildTitleStyle(settings: TitleSettings, containerHeight: number): Reac
     paddingLeft: `${(ASS_TITLE_MARGIN_SIDE / 1280) * 100}%`,
     paddingRight: `${(ASS_TITLE_MARGIN_SIDE / 1280) * 100}%`,
   }
-
-  if (settings.stroke) {
-    base.WebkitTextStroke = "1px #000000"
-  }
-
-  if (settings.shadow) {
-    base.textShadow = "2px 2px 4px rgba(0,0,0,0.8)"
-  }
-
+  if (settings.stroke) base.WebkitTextStroke = "1px #000000"
+  if (settings.shadow) base.textShadow = "2px 2px 4px rgba(0,0,0,0.8)"
   return base
 }
 
-/**
- * CSS 标题预览覆盖层
- * 始终渲染：有标题显示标题，无标题显示占位文本"标题预览"
- * z-index: 20（在视频 z-index:1 和控制条 z-index:10 之上）
- */
+/** CSS 标题实时预览覆盖层 */
 const TitleOverlay: React.FC<{ titleSettings: TitleSettings }> = ({ titleSettings }) => {
   const containerRef = useRef<HTMLDivElement>(null)
-  const [containerHeight, setContainerHeight] = useState(400) // fallback
+  const [containerHeight, setContainerHeight] = useState(400)
 
-  // ResizeObserver 获取容器实际高度
   useEffect(() => {
     const el = containerRef.current
     if (!el) return
@@ -132,7 +113,6 @@ const TitleOverlay: React.FC<{ titleSettings: TitleSettings }> = ({ titleSetting
       }
     })
     ro.observe(el)
-    // 初始化也读一次
     const rect = el.getBoundingClientRect()
     if (rect.height > 0) setContainerHeight(rect.height)
     return () => ro.disconnect()
@@ -144,7 +124,7 @@ const TitleOverlay: React.FC<{ titleSettings: TitleSettings }> = ({ titleSetting
   )
   const titleStyle = useMemo(
     () => buildTitleStyle(titleSettings, containerHeight),
-    // eslint-disable-next-line react-hooks/exhaustive-deps -- 已逐字段列出 titleSettings 依赖
+    // eslint-disable-next-line react-hooks/exhaustive-deps
     [
       containerHeight,
       titleSettings.font,
@@ -170,14 +150,8 @@ const TitleOverlay: React.FC<{ titleSettings: TitleSettings }> = ({ titleSetting
         overflow: "hidden",
       }}
     >
-      <div
-        style={{
-          ...positionStyle,
-          ...titleStyle,
-          position: "absolute",
-        }}
-      >
-        {displayTitle.split("/").map((part, i) => (
+      <div style={{ ...positionStyle, ...titleStyle, position: "absolute" }}>
+        {displayTitle.split(/[/／]/).map((part, i) => (
           <span key={i}>
             {i > 0 && <br />}
             {part}
@@ -191,30 +165,47 @@ const TitleOverlay: React.FC<{ titleSettings: TitleSettings }> = ({ titleSetting
 /* ── 主组件 ── */
 
 export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
-  assets,
-  template,
+  previewStatus,
+  videoUrl,
+  progress,
+  error,
+  onRetry,
   videoRatio,
-  assetsReady,
-  assetsLoading,
   titleSettings,
-  voiceAudioUrl,
+  assetCount,
 }) => {
   const videoAspectStyle = { aspectRatio: (videoRatio || "9:16").replace(":", "/") }
+  const isLoading = previewStatus === "loading"
+  const isReady = previewStatus === "ready" || previewStatus === "stale"
+  const isFailed = previewStatus === "failed"
+  const isIdle = previewStatus === "idle"
+  const isStale = previewStatus === "stale"
 
   return (
     <div className="xx-generate-preview">
       <div className="xx-preview-header">
         <h3>预览视频</h3>
-        {assetsReady && assets.length > 0 && <span className="xx-preview-badge">实时预览</span>}
+        {isReady && !isStale && <span className="xx-preview-badge">服务器渲染</span>}
+        {isStale && (
+          <span
+            style={{
+              fontSize: 12,
+              color: "#faad14",
+              display: "flex",
+              alignItems: "center",
+              gap: 4,
+            }}
+          >
+            <ExclamationCircleOutlined /> 配置已变更
+          </span>
+        )}
+        {isLoading && <span className="xx-preview-badge">渲染中</span>}
       </div>
 
-      {/* ✅ 预览容器 — 唯一的 .xx-preview-video 容器
-          内部所有内容（视频、控制条、标题叠加层）通过 absolute 定位填充 */}
       <div className="xx-preview-video" style={{ ...videoAspectStyle, position: "relative" }}>
-        {/* 加载中状态 */}
-        {assetsLoading && (
+        {/* 加载中 */}
+        {isLoading && (
           <div
-            className="xx-preview-loading-center"
             style={{
               position: "absolute",
               inset: 0,
@@ -223,34 +214,130 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
               alignItems: "center",
               justifyContent: "center",
               zIndex: 5,
+              background: "rgba(0,0,0,0.6)",
             }}
           >
             <LoadingOutlined style={{ fontSize: 36, color: "#fff" }} spin />
-            <p style={{ marginTop: 12, color: "rgba(255,255,255,0.8)", fontSize: 14 }}>
-              加载素材中...
+            <p style={{ marginTop: 12, color: "rgba(255,255,255,0.9)", fontSize: 14 }}>
+              正在渲染预览视频{progress > 0 ? `...${progress}%` : "..."}
+            </p>
+            <p style={{ marginTop: 4, color: "rgba(255,255,255,0.5)", fontSize: 12 }}>
+              首次渲染约需 30-60 秒
             </p>
           </div>
         )}
 
-        {/* 前端播放器（视频 + 控制条 + 播放按钮）*/}
-        <FrontendPreviewPlayer
-          assets={assets}
-          template={template}
-          videoRatio={videoRatio}
-          ready={assetsReady}
-          voiceAudioUrl={voiceAudioUrl}
-        />
+        {/* 空闲状态（尚未触发预览） */}
+        {isIdle && (
+          <div
+            style={{
+              position: "absolute",
+              inset: 0,
+              display: "flex",
+              flexDirection: "column",
+              alignItems: "center",
+              justifyContent: "center",
+              background: "rgba(0,0,0,0.3)",
+              zIndex: 5,
+            }}
+          >
+            <p style={{ color: "rgba(255,255,255,0.7)", fontSize: 14 }}>等待素材选择...</p>
+          </div>
+        )}
 
-        {/* CSS 标题实时预览层 — z-index: 20，始终渲染在内容层之上 */}
-        {titleSettings && <TitleOverlay titleSettings={titleSettings} />}
+        {/* 服务器渲染的真实视频 */}
+        {isReady && videoUrl && (
+          <video
+            key={videoUrl}
+            src={videoUrl}
+            controls
+            autoPlay
+            loop
+            playsInline
+            style={{
+              width: "100%",
+              height: "100%",
+              objectFit: "contain",
+              position: "absolute",
+              inset: 0,
+              zIndex: 1,
+            }}
+          />
+        )}
+
+        {/* 标题样式实时预览层（仅在有视频时叠加） */}
+        {isReady && titleSettings && <TitleOverlay titleSettings={titleSettings} />}
+
+        {/* stale 遮罩：配置变更提示 */}
+        {isStale && (
+          <div
+            style={{
+              position: "absolute",
+              bottom: 0,
+              left: 0,
+              right: 0,
+              padding: "10px 16px",
+              background: "linear-gradient(transparent, rgba(0,0,0,0.85))",
+              display: "flex",
+              alignItems: "center",
+              justifyContent: "space-between",
+              gap: 8,
+              zIndex: 30,
+            }}
+          >
+            <span style={{ color: "rgba(255,255,255,0.9)", fontSize: 12 }}>
+              配置已变更，预览内容可能不是最新
+            </span>
+            <Button size="small" type="primary" icon={<ReloadOutlined />} onClick={onRetry}>
+              重新预览
+            </Button>
+          </div>
+        )}
+
+        {/* 错误状态 */}
+        {isFailed && (
+          <div
+            style={{
+              position: "absolute",
+              inset: 0,
+              display: "flex",
+              flexDirection: "column",
+              alignItems: "center",
+              justifyContent: "center",
+              background: "rgba(0,0,0,0.7)",
+              zIndex: 10,
+              padding: 24,
+            }}
+          >
+            <ExclamationCircleOutlined style={{ fontSize: 40, color: "#ff4d4f" }} />
+            <p
+              style={{
+                marginTop: 12,
+                color: "rgba(255,255,255,0.9)",
+                fontSize: 14,
+                textAlign: "center",
+              }}
+            >
+              {error || "预览渲染失败"}
+            </p>
+            <Button
+              type="primary"
+              icon={<ReloadOutlined />}
+              onClick={onRetry}
+              style={{ marginTop: 12 }}
+            >
+              重新预览
+            </Button>
+          </div>
+        )}
       </div>
 
       {/* 素材信息 */}
-      {assetsReady && assets.length > 0 && (
+      {assetCount !== undefined && assetCount > 0 && (
         <div className="xx-preview-info">
           <div className="xx-preview-info-row">
             <span>素材数</span>
-            <span>{assets.length} 个</span>
+            <span>{assetCount} 个</span>
           </div>
           <div className="xx-preview-info-row">
             <span>比例</span>

apps/web/src/pages/generate/components/Step5GeneratePreview.tsx

@@ -1,7 +1,10 @@
 /**
- * Step 5 生成预览组件
- * 架构改造：移除后端预览生成，改为前端实时预览
- * 左侧仅保留标题样式面板，视频在右侧 PreviewVideoPanel 实时播放
+ * Step 5 预览设置组件
+ *
+ * 前端实时预览架构：
+ * - 右侧面板使用 FrontendPreviewPlayer 实时播放素材片段
+ * - 标题样式可实时调整，CSS 层即时叠加预览
+ * - 点"确认生成"时触发一次服务器渲染
  */
 import React from "react"
 import { PlayCircleOutlined } from "@ant-design/icons"
@@ -10,7 +13,6 @@ import type { TitleSettings } from "../types"
 import TitleStylePanel from "./title/TitleStylePanel"
 
 interface Step5GeneratePreviewProps {
-  /* 标题样式 */
   titleSettings: TitleSettings
   onUpdatePosition: (position: string) => void
   onUpdateFont: (font: string) => void
@@ -41,9 +43,7 @@ const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
     <div className="xx-form-section">
       <h3>🎬 预览设置</h3>
 
-      {/* 前端实时预览提示 */}
       <div
-        className="xx-preview-tip"
         style={{
           display: "flex",
           alignItems: "center",
@@ -57,11 +57,10 @@ const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
       >
         <PlayCircleOutlined style={{ fontSize: 18, color: "#3b82f6" }} />
         <span style={{ fontSize: 13, color: "var(--text-secondary, #666)" }}>
-          右侧面板直接播放素材片段，调整标题样式可实时预览效果
+          右侧为实时预览，选完素材即可播放。确认生成后服务器渲染最终视频
         </span>
       </div>
 
-      {/* 标题样式面板 */}
       <TitleStylePanel
         settings={titleSettings}
         onUpdatePosition={onUpdatePosition}

apps/web/src/pages/generate/components/Step6CoverSettings.tsx

@@ -16,6 +16,20 @@ interface Step6CoverSettingsProps {
   selectedTemplate?: string
   /** Step4 标题设置，用于预览视频烧录标题 & 封面叠加标题 */
   titleSettings?: import("../types").TitleSettings
+  /** 预览任务创建回调——将 task_id 暴露给父组件供 confirmGeneration 复用 */
+  onPreviewTaskCreated?: (taskId: string) => void
+  /** 从预览响应中提取到 source_edit_plan_id 时的回调 */
+  onSourceEditPlanIdExtracted?: (planId: string) => void
+  /** 配音模式 */
+  voiceMode?: "preset" | "custom" | "clone"
+  /** 选中的配音素材 ID */
+  selectedVoice?: string
+  /** 选中的克隆音色 ID */
+  selectedClonedVoice?: string
+  /** BGM 开关 */
+  bgm?: boolean
+  /** BGM 配置 */
+  bgmConfig?: { enabled: boolean; music_id?: string }
 }
 
 const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
@@ -43,6 +57,13 @@ const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
     assetIds: props.assetIds,
     selectedTemplate: props.selectedTemplate,
     titleSettings: props.titleSettings,
+    onPreviewTaskCreated: props.onPreviewTaskCreated,
+    onSourceEditPlanIdExtracted: props.onSourceEditPlanIdExtracted,
+    voiceMode: props.voiceMode,
+    selectedVoice: props.selectedVoice,
+    selectedClonedVoice: props.selectedClonedVoice,
+    bgm: props.bgm,
+    bgmConfig: props.bgmConfig,
   })
 
   const handleAutoGenerate = () => {

apps/web/src/pages/generate/generate.css

@@ -2525,6 +2525,7 @@
 .xx-generate-right-col {
   display: flex;
   flex-direction: column;
+  align-items: center;
   gap: 16px;
 }
 

apps/web/src/pages/generate/hooks/generate-video/types.ts

@@ -21,6 +21,15 @@ export interface UseGenerateVideoProps {
   generateCount: number
   /** 当前草稿 ID（URL 参数 edit_plan_id，用于后端回写任务关联） */
   sourceEditPlanId?: string | null
+  /** 预览任务 ID（由 useStep6Cover 创建后写入，供 confirmGeneration 复用预览产物） */
+  previewTaskId?: string | null
+  /** BGM 配置（来自模板 bgm_config，受 bgm 开关控制） */
+  bgmConfig?: {
+    enabled: boolean
+    music_id?: string
+  }
+  /** 生成成功后的回调（用于清除持久化的 previewTaskId 等状态） */
+  onGenerationSuccess?: () => void
 }
 
 /** 生成阶段 */

apps/web/src/pages/generate/hooks/useGenerateFormState/index.ts

@@ -14,6 +14,7 @@ import { useTemplateSelection } from "./useTemplateSelection"
 import { useTitleCoverSync } from "./useTitleCoverSync"
 import { useVoiceState } from "./useVoiceState"
 import { usePlanConfigLoader } from "./usePlanConfigLoader"
+import { usePersistedState } from "../usePersistedState"
 
 const DEFAULT_TITLE_SETTINGS: TitleSettings = {
   aiAutoSelect: false,
@@ -95,6 +96,14 @@ export interface GenerateFormState {
   setPreviewVideo: (v: GeneratedVideo | null) => void
   previewModalOpen: boolean
   setPreviewModalOpen: (open: boolean) => void
+
+  /** 预览任务 ID（由 useStep6Cover 创建后写入，供 useGenerateVideo 复用） */
+  previewTaskId: string | null
+  setPreviewTaskId: (id: string | null) => void
+
+  /** 从预览响应中提取的 source_edit_plan_id（供 fallback 路径使用） */
+  storedSourceEditPlanId: string | null
+  setStoredSourceEditPlanId: (planId: string | null) => void
 }
 
 export const useGenerateFormState = (): GenerateFormState => {
@@ -160,6 +169,30 @@ export const useGenerateFormState = (): GenerateFormState => {
   const [previewVideo, setPreviewVideo] = useState<GeneratedVideo | null>(null)
   const [previewModalOpen, setPreviewModalOpen] = useState(false)
 
+  /* ── 预览任务 ID（useStep6Cover 创建预览时写入，useGenerateVideo 复用） ── */
+  // 持久化到 localStorage，key 按 editPlanId/templateId 区分，刷新页面后可恢复
+  const previewStorageKey = editPlanId
+    ? `preview_task_id_${editPlanId}`
+    : selectedTemplate
+      ? `preview_task_id_tpl_${selectedTemplate}`
+      : null
+  const [previewTaskId, setPreviewTaskId] = usePersistedState<string | null>(
+    previewStorageKey,
+    null,
+  )
+
+  /* ── 从预览响应中提取的 source_edit_plan_id（供 fallback 路径使用） ── */
+  // 持久化到 localStorage，刷新页面后 fallback 路径仍能正确传递 source_edit_plan_id
+  const planIdStorageKey = editPlanId
+    ? `source_edit_plan_id_${editPlanId}`
+    : selectedTemplate
+      ? `source_edit_plan_id_tpl_${selectedTemplate}`
+      : null
+  const [storedSourceEditPlanId, setStoredSourceEditPlanId] = usePersistedState<string | null>(
+    planIdStorageKey,
+    null,
+  )
+
   /* ── 从 URL / 编辑计划加载配置 ── */
   usePlanConfigLoader({
     editPlanId,
@@ -208,5 +241,9 @@ export const useGenerateFormState = (): GenerateFormState => {
     setPreviewVideo,
     previewModalOpen,
     setPreviewModalOpen,
+    previewTaskId,
+    setPreviewTaskId,
+    storedSourceEditPlanId,
+    setStoredSourceEditPlanId,
   }
 }

apps/web/src/pages/generate/hooks/useGenerateVideo.ts

@@ -10,10 +10,11 @@ import type { UseGenerateVideoProps } from "./generate-video/types"
 import { getGenerationPhase } from "./generate-video/phase"
 import { useGenerationPolling } from "./generate-video/useGenerationPolling"
 import { validateGenerateInputs } from "./generate-video/buildPayload"
+import { calculateResolution } from "../utils/calculateResolution"
 import { extractBackendError, translateError } from "./generate-video/errorUtils"
 
 export function useGenerateVideo(props: UseGenerateVideoProps) {
-  const { selectedTemplate } = props
+  const { selectedTemplate, onGenerationSuccess } = props
 
   /* ── 生成状态 ── */
   const [generating, setGenerating] = useState(false)
@@ -23,11 +24,16 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
   const [generatedVideos, setGeneratedVideos] = useState<GeneratedVideo[]>([])
 
   const handleProgress = useCallback((p: number) => setProgress(p), [])
-  const handleComplete = useCallback((videos: unknown[]) => {
-    setGenerating(false)
-    setGenerated(true)
-    setGeneratedVideos(videos as GeneratedVideo[])
-  }, [])
+  const handleComplete = useCallback(
+    (videos: unknown[]) => {
+      setGenerating(false)
+      setGenerated(true)
+      setGeneratedVideos(videos as GeneratedVideo[])
+      // 生成成功后清除持久化的预览状态，避免下次进入复用旧任务
+      onGenerationSuccess?.()
+    },
+    [onGenerationSuccess],
+  )
   const handleFailed = useCallback((errorMsg: string) => {
     setGenerating(false)
     setGenerateError(errorMsg)
@@ -54,37 +60,10 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
     clearTimer()
 
     try {
-      // 解析分辨率
-      const ratio = props.videoRatio || "9:16"
-      let outputWidth: number
-      let outputHeight: number
-
-      if (ratio.includes(":")) {
-        const [rw, rh] = ratio.split(":").map(Number)
-        if (rw > 0 && rh > 0) {
-          const [longSide, shortSide] = rw < rh ? [rh, rw] : [rw, rh]
-          const baseLong = 1920
-          const baseShort = Math.round((baseLong * shortSide) / longSide)
-          const evenShort = baseShort - (baseShort % 2)
-          if (rw < rh) {
-            outputWidth = evenShort
-            outputHeight = baseLong
-          } else {
-            outputWidth = baseLong
-            outputHeight = evenShort
-          }
-        } else {
-          outputWidth = 1080
-          outputHeight = 1920
-        }
-      } else if (ratio.includes("x")) {
-        const [wStr, hStr] = ratio.split("x")
-        outputWidth = parseInt(wStr, 10) || 1080
-        outputHeight = parseInt(hStr, 10) || 1920
-      } else {
-        outputWidth = 1080
-        outputHeight = 1920
-      }
+      // 解析分辨率（共享工具函数）
+      const { width: outputWidth, height: outputHeight } = calculateResolution(
+        props.videoRatio || "9:16",
+      )
 
       const assetIds =
         props.materialMode === "auto" ? props.smartSelectedIds : props.selectedMaterials
@@ -92,7 +71,11 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
       // 封面 URL：优先 AI 生成缩略图，兜底用户上传
       const coverUrl = props.coverSettings?.thumbnail_url || props.coverSettings?.upload_url || ""
 
-      // 直接创建正式生成任务
+      // 解析配音参数：voiceMode=clone 时用 selectedClonedVoice，否则用 selectedVoice
+      const voiceLibraryId =
+        props.voiceMode === "clone" ? props.selectedClonedVoice || "" : props.selectedVoice || ""
+
+      // 创建生成任务（服务器渲染）
       const taskResp = await createGenerationTask({
         template_id: selectedTemplate,
         asset_ids: assetIds,
@@ -102,6 +85,14 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
         custom_title: props.titleSettings?.title || "",
         duration: props.duration || undefined,
         video_ratio: props.videoRatio,
+        // 配音：优先用 voice_library_id（配音素材库 asset），兜底 voice_ids
+        ...(voiceLibraryId ? { voice_library_id: voiceLibraryId } : {}),
+        ...(props.selectedVoice && !voiceLibraryId ? { voice_ids: [props.selectedVoice] } : {}),
+        // BGM 配置：受 bgm 开关控制，enabled=false 时也显式传覆盖模板 BGM
+        bgm_config: {
+          enabled: props.bgm !== false,
+          ...(props.bgmConfig?.music_id ? { preset_id: props.bgmConfig.music_id } : {}),
+        },
         ...(props.sourceEditPlanId ? { source_edit_plan_id: props.sourceEditPlanId } : {}),
         ...(props.titleSettings?.title
           ? {
@@ -118,9 +109,8 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
             }
           : {}),
       })
-
-      // 从创建响应直接拿 task_id，改用新接口轮询
       const taskId = taskResp.items?.[0]?.id
+
       if (!taskId) {
         throw new Error("创建任务成功但未返回任务 ID，请稍后在任务列表查看")
       }

apps/web/src/pages/generate/hooks/usePersistedState.ts

@@ -0,0 +1,63 @@
+import { useState, useEffect, useCallback, useRef } from "react"
+
+/**
+ * 持久化到 localStorage 的 state hook
+ *
+ * 用于在页面刷新后恢复 previewTaskId / sourceEditPlanId 等关键状态。
+ * 当 localStorage 不可用（SSR、隐私模式等）时自动降级为普通 useState。
+ * 当 key 变化时（例如切换模板），自动从新 key 重新读取并更新 state。
+ */
+export function usePersistedState<T>(
+  key: string | null | undefined,
+  defaultValue: T,
+): [T, (value: T | ((prev: T) => T)) => void] {
+  const storageKey = key ? `xiaoxia_${key}` : null
+  // defaultValue 用 ref 持有，避免作为 useEffect 依赖导致频繁重跑
+  const defaultValueRef = useRef(defaultValue)
+  defaultValueRef.current = defaultValue
+
+  const readFromStorage = useCallback((k: string | null): T => {
+    if (!k) return defaultValueRef.current
+    try {
+      const stored = localStorage.getItem(k)
+      if (stored !== null) {
+        return JSON.parse(stored) as T
+      }
+    } catch (e) {
+      // localStorage 不可用或 JSON 解析失败，使用默认值
+      console.warn("[usePersistedState] 读取 localStorage 失败:", e)
+    }
+    return defaultValueRef.current
+  }, [])
+
+  const [state, setState] = useState<T>(() => readFromStorage(storageKey))
+
+  // key 变化时（如切换模板/草稿），从新 key 重新读取，避免状态与存储不同步
+  useEffect(() => {
+    setState(readFromStorage(storageKey))
+  }, [storageKey, readFromStorage])
+
+  const setPersistedState = useCallback(
+    (value: T | ((prev: T) => T)) => {
+      setState((prev) => {
+        const nextValue = typeof value === "function" ? (value as (prev: T) => T)(prev) : value
+        if (storageKey) {
+          try {
+            if (nextValue === null || nextValue === undefined || nextValue === "") {
+              localStorage.removeItem(storageKey)
+            } else {
+              localStorage.setItem(storageKey, JSON.stringify(nextValue))
+            }
+          } catch (e) {
+            // localStorage 写入失败（存储空间满/隐私模式），静默忽略
+            console.warn("[usePersistedState] 写入 localStorage 失败:", e)
+          }
+        }
+        return nextValue
+      })
+    },
+    [storageKey],
+  )
+
+  return [state, setPersistedState]
+}

apps/web/src/pages/generate/hooks/useServerPreview.ts

@@ -0,0 +1,294 @@
+/**
+ * 服务器渲染预览 Hook
+ *
+ * 核心职责：
+ * 1. 调用 POST /generation/preview 创建服务器预览渲染任务
+ * 2. 轮询 GET /generation/preview/{task_id} 直到完成
+ * 3. 返回服务器渲染的真实视频 URL（供 <video> 标签播放）
+ * 4. 检测配置变更，标记预览失效（stale）或自动重新渲染
+ * 5. 网络错误自动重试 2 次
+ *
+ * 状态机：
+ *   idle → loading → ready → stale (config changed)
+ *                   ↘ failed → idle (retry)
+ */
+import { useState, useCallback, useRef, useEffect } from "react"
+import { createPreview, getPreviewStatus } from "@/api/generation/preview"
+import type { CreatePreviewRequest } from "@/api/generation/types"
+
+export type ServerPreviewStatus = "idle" | "loading" | "ready" | "stale" | "failed"
+
+interface UseServerPreviewOptions {
+  /** 是否启用预览（Step4+ 且有素材和模板时） */
+  enabled: boolean
+  /** 构建预览请求参数（每次 render 调用，获取最新配置） */
+  buildRequest: () => CreatePreviewRequest
+  /** 预览任务创建成功回调 */
+  onPreviewTaskCreated?: (taskId: string, sourceEditPlanId?: string) => void
+}
+
+interface UseServerPreviewReturn {
+  status: ServerPreviewStatus
+  videoUrl: string | null
+  error: string | null
+  /** 进度 0-100 */
+  progress: number
+  /** 手动触发预览创建（"重新预览"按钮或标题变更后手动刷新） */
+  triggerPreview: () => void
+  /** 当前预览任务 ID */
+  taskId: string | null
+}
+
+const POLL_INTERVAL = 2000
+const POLL_TIMEOUT = 120_000
+const MAX_NETWORK_RETRIES = 2
+
+/**
+ * 对配置参数做指纹，用于检测配置是否变化
+ */
+function buildFingerprint(req: CreatePreviewRequest): string {
+  return JSON.stringify({
+    t: req.template_id,
+    a: [...req.asset_ids].sort(),
+    d: req.duration,
+    r: req.video_ratio,
+    v: req.voice_library_id,
+    b: req.bgm_config,
+    title: req.title_config,
+  })
+}
+
+export function useServerPreview({
+  enabled,
+  buildRequest,
+  onPreviewTaskCreated,
+}: UseServerPreviewOptions): UseServerPreviewReturn {
+  const [status, setStatus] = useState<ServerPreviewStatus>("idle")
+  const [videoUrl, setVideoUrl] = useState<string | null>(null)
+  const [error, setError] = useState<string | null>(null)
+  const [progress, setProgress] = useState(0)
+  const [taskId, setTaskId] = useState<string | null>(null)
+
+  const pollTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
+  const timeoutTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
+  const requestSeqRef = useRef(0)
+  const renderedFingerprintRef = useRef<string>("")
+  const mountedRef = useRef(true)
+  const networkRetriesRef = useRef(0)
+
+  // 始终持有最新的 buildRequest 和回调
+  const buildRequestRef = useRef(buildRequest)
+  buildRequestRef.current = buildRequest
+  const onCreatedRef = useRef(onPreviewTaskCreated)
+  onCreatedRef.current = onPreviewTaskCreated
+
+  /* ── 清理 ── */
+  const clearTimers = useCallback(() => {
+    if (pollTimerRef.current) {
+      clearTimeout(pollTimerRef.current)
+      pollTimerRef.current = null
+    }
+    if (timeoutTimerRef.current) {
+      clearTimeout(timeoutTimerRef.current)
+      timeoutTimerRef.current = null
+    }
+  }, [])
+
+  useEffect(() => {
+    mountedRef.current = true
+    return () => {
+      mountedRef.current = false
+      clearTimers()
+    }
+  }, [clearTimers])
+
+  /* ── 创建预览 + 轮询 ── */
+  const createAndPoll = useCallback(
+    async (request: CreatePreviewRequest, seq: number) => {
+      setStatus("loading")
+      setProgress(0)
+      setError(null)
+      networkRetriesRef.current = 0
+
+      try {
+        const resp = await createPreview(request)
+        if (seq !== requestSeqRef.current || !mountedRef.current) return
+
+        setTaskId(resp.task_id)
+        onCreatedRef.current?.(resp.task_id, resp.source_edit_plan_id)
+
+        let completed = false
+
+        // 超时保护
+        timeoutTimerRef.current = setTimeout(() => {
+          if (completed || seq !== requestSeqRef.current || !mountedRef.current) return
+          completed = true
+          clearTimers()
+          setStatus("failed")
+          setError("预览渲染超时（120秒），请重试")
+        }, POLL_TIMEOUT)
+
+        const poll = async () => {
+          if (completed || seq !== requestSeqRef.current || !mountedRef.current) return
+
+          try {
+            const st = await getPreviewStatus(resp.task_id)
+            if (completed || seq !== requestSeqRef.current || !mountedRef.current) return
+
+            if (st.status === "completed" && st.video_url) {
+              completed = true
+              clearTimers()
+              renderedFingerprintRef.current = buildFingerprint(request)
+              setVideoUrl(st.video_url)
+              setProgress(100)
+              setStatus("ready")
+              setError(null)
+              return
+            }
+
+            if (st.status === "failed" || st.status === "cancelled") {
+              completed = true
+              clearTimers()
+              setStatus("failed")
+              setError(
+                st.status === "cancelled"
+                  ? "预览任务已取消"
+                  : st.error_message || "预览渲染失败，请重试",
+              )
+              return
+            }
+
+            // pending / generating
+            if (typeof st.progress === "number") setProgress(st.progress)
+            pollTimerRef.current = setTimeout(poll, POLL_INTERVAL)
+          } catch (pollErr) {
+            if (completed || seq !== requestSeqRef.current || !mountedRef.current) return
+            if (networkRetriesRef.current < MAX_NETWORK_RETRIES) {
+              networkRetriesRef.current += 1
+              console.warn(
+                `[ServerPreview] 轮询网络错误，第 ${networkRetriesRef.current} 次重试`,
+                pollErr,
+              )
+              pollTimerRef.current = setTimeout(poll, POLL_INTERVAL * 2)
+            } else {
+              completed = true
+              clearTimers()
+              setStatus("failed")
+              setError("网络错误，无法获取预览状态，请重试")
+            }
+          }
+        }
+
+        poll()
+      } catch (createErr) {
+        if (seq !== requestSeqRef.current || !mountedRef.current) return
+        console.error("[ServerPreview] 创建预览任务失败:", createErr)
+
+        const isNetworkError =
+          !!(createErr as { request?: unknown })?.request ||
+          (createErr as { code?: string })?.code === "ERR_NETWORK"
+
+        if (isNetworkError && networkRetriesRef.current < MAX_NETWORK_RETRIES) {
+          networkRetriesRef.current += 1
+          console.warn(`[ServerPreview] 创建任务网络错误，第 ${networkRetriesRef.current} 次重试`)
+          setTimeout(() => {
+            if (seq === requestSeqRef.current && mountedRef.current) {
+              createAndPoll(request, seq)
+            }
+          }, POLL_INTERVAL * 2)
+          return
+        }
+
+        const errData = (
+          createErr as { response?: { data?: { detail?: string; message?: string } } }
+        )?.response?.data
+        setStatus("failed")
+        setError(errData?.detail || errData?.message || "预览任务创建失败，请重试")
+      }
+    },
+    [clearTimers],
+  )
+
+  /* ── 手动触发预览 ── */
+  const triggerPreview = useCallback(() => {
+    if (!enabled) return
+    const request = buildRequestRef.current()
+    if (!request.template_id || request.asset_ids.length === 0) return
+
+    clearTimers()
+    const seq = ++requestSeqRef.current
+    setVideoUrl(null)
+    setTaskId(null)
+    createAndPoll(request, seq)
+  }, [enabled, clearTimers, createAndPoll])
+
+  /* ── 自动触发 + 配置变更检测 ── */
+  // 每次 render 都检查最新配置 fingerprint，与已渲染的 fingerprint 比较
+  const request = enabled ? buildRequest() : null
+  const currentFingerprint = request
+    ? request.template_id && request.asset_ids.length > 0
+      ? buildFingerprint(request)
+      : ""
+    : ""
+
+  // 首次进入自动触发
+  const didInitRef = useRef(false)
+  useEffect(() => {
+    if (!enabled || !currentFingerprint) {
+      didInitRef.current = false
+      // 禁用时取消进行中的轮询，避免回到前序步骤后仍在后台轮询
+      requestSeqRef.current += 1
+      clearTimers()
+      return
+    }
+    if (!didInitRef.current) {
+      didInitRef.current = true
+      renderedFingerprintRef.current = currentFingerprint
+      triggerPreview()
+    }
+  }, [enabled, currentFingerprint, triggerPreview, clearTimers])
+
+  // 配置变更检测：素材/配音/BGM 等变化 → 自动重渲染；标题样式变化 → 标记 stale
+  const prevFingerprintRef = useRef(currentFingerprint)
+  useEffect(() => {
+    if (!enabled || !currentFingerprint) return
+    const prev = prevFingerprintRef.current
+    prevFingerprintRef.current = currentFingerprint
+
+    if (!prev || prev === currentFingerprint) return
+    if (currentFingerprint === renderedFingerprintRef.current) return
+
+    // 配置已变更
+    // 判断是标题样式变更还是素材/配音/BGM 变更
+    const prevParsed = JSON.parse(prev) as Record<string, unknown>
+    const currParsed = JSON.parse(currentFingerprint) as Record<string, unknown>
+    const nonTitleChanged =
+      prevParsed.t !== currParsed.t ||
+      prevParsed.a !== currParsed.a ||
+      prevParsed.d !== currParsed.d ||
+      prevParsed.r !== currParsed.r ||
+      prevParsed.v !== currParsed.v ||
+      JSON.stringify(prevParsed.b) !== JSON.stringify(currParsed.b)
+
+    if (nonTitleChanged) {
+      // 素材/配音/BGM/模板等变化 → 自动重新渲染
+      renderedFingerprintRef.current = currentFingerprint
+      triggerPreview()
+    } else {
+      // 仅标题文字/样式变化 → 标记 stale，不自动重渲染（避免频繁请求）
+      // 实时预览由 CSS TitleOverlay 提供
+      setStatus((s) => (s === "ready" ? "stale" : s))
+    }
+  }, [enabled, currentFingerprint, triggerPreview])
+
+  return {
+    status,
+    videoUrl,
+    error,
+    progress,
+    triggerPreview,
+    taskId,
+  }
+}
+
+export default useServerPreview

apps/web/src/pages/generate/hooks/useStep6Cover.ts

@@ -2,7 +2,7 @@
  * Step 6 封面设置 Hook
  * 封装封面设置的交互逻辑，对接后端封面模板 CRUD API
  */
-import { useCallback, useEffect, useState } from "react"
+import { useCallback, useEffect, useRef, useState } from "react"
 import { message } from "antd"
 import type { CoverConfig, CoverTemplate } from "../types/cover"
 import { generateCover } from "@/api/generation"
@@ -26,6 +26,20 @@ interface UseStep6CoverProps {
   selectedTemplate?: string
   /** Step4 标题设置，用于预览视频烧录标题 & 封面叠加标题 */
   titleSettings?: TitleSettings
+  /** 预览任务创建回调——将 task_id 暴露给父组件供 confirmGeneration 复用 */
+  onPreviewTaskCreated?: (taskId: string) => void
+  /** 从预览响应中提取到 source_edit_plan_id 时的回调 */
+  onSourceEditPlanIdExtracted?: (planId: string) => void
+  /** 配音模式 */
+  voiceMode?: "preset" | "custom" | "clone"
+  /** 选中的配音素材 ID（配音素材库 asset ID） */
+  selectedVoice?: string
+  /** 选中的克隆音色 ID */
+  selectedClonedVoice?: string
+  /** BGM 开关 */
+  bgm?: boolean
+  /** BGM 配置（来自模板） */
+  bgmConfig?: { enabled: boolean; music_id?: string }
 }
 
 export function useStep6Cover({
@@ -35,8 +49,17 @@ export function useStep6Cover({
   assetIds = [],
   selectedTemplate = "",
   titleSettings,
+  onPreviewTaskCreated,
+  onSourceEditPlanIdExtracted,
+  voiceMode,
+  selectedVoice,
+  selectedClonedVoice,
+  bgm,
+  bgmConfig,
 }: UseStep6CoverProps) {
   const [generating, setGenerating] = useState(false)
+  // 防竞态：记录当前预览生成的参数指纹，任务完成时校验一致性
+  const previewParamsRef = useRef<string>("")
 
   // ── 封面设置弹窗状态 ──
   const [showCoverSettings, setShowCoverSettings] = useState(false)
@@ -151,10 +174,22 @@ export function useStep6Cover({
         console.log("[Step6] 检测到预览缺失，尝试自动创建预览渲染任务...")
         message.info("正在准备预览视频，请稍候...")
         try {
+          // 记录当前参数指纹，用于任务完成时校验一致性（防竞态）
+          previewParamsRef.current = JSON.stringify({ selectedTemplate, assetIds, titleSettings })
+          // 解析配音参数：voiceMode=clone 时用 selectedClonedVoice，否则用 selectedVoice
+          const previewVoiceLibraryId =
+            voiceMode === "clone" ? selectedClonedVoice || "" : selectedVoice || ""
           const previewResp = await createPreview({
             template_id: selectedTemplate,
             asset_ids: assetIds,
             duration: duration || 30,
+            // 配音：voice_library_id 是配音素材库 asset ID（用户上传的音频或 AI 配音）
+            ...(previewVoiceLibraryId ? { voice_library_id: previewVoiceLibraryId } : {}),
+            // BGM 配置：受 bgm 开关控制
+            bgm_config: {
+              enabled: bgm !== false,
+              ...(bgmConfig?.music_id ? { preset_id: bgmConfig.music_id } : {}),
+            },
             ...(titleSettings?.title
               ? {
                   title_config: {
@@ -170,6 +205,15 @@ export function useStep6Cover({
                 }
               : {}),
           })
+          // 将预览任务 ID 暴露给父组件，供 Step7 确认生成时复用（confirmGeneration）
+          const currentFingerprint = JSON.stringify({ selectedTemplate, assetIds, titleSettings })
+          if (previewResp.task_id && previewParamsRef.current === currentFingerprint) {
+            onPreviewTaskCreated?.(previewResp.task_id)
+            // 提取后端自动关联的 source_edit_plan_id，供 fallback 路径使用
+            if (previewResp.source_edit_plan_id) {
+              onSourceEditPlanIdExtracted?.(previewResp.source_edit_plan_id)
+            }
+          }
           // 轮询等待预览渲染完成：递归 setTimeout 避免请求重叠 + 120s 超时兜底
           await new Promise<void>((resolve, reject) => {
             let finished = false
@@ -282,6 +326,13 @@ export function useStep6Cover({
     generating,
     duration,
     titleSettings,
+    onPreviewTaskCreated,
+    onSourceEditPlanIdExtracted,
+    voiceMode,
+    selectedVoice,
+    selectedClonedVoice,
+    bgm,
+    bgmConfig,
   ])
 
   // ── 模板操作方法 ──

apps/web/src/pages/generate/hooks/useStepNavigation.ts

@@ -3,7 +3,7 @@
  * 管理步骤切换与各步骤的前置校验
  * 步骤顺序：模板(1) → 素材(2) → 配音(3) → 标题(4) → 预览(5) → 封面(6) → 确认(7)
  *
- * V24: previewReady 改为前端素材加载状态
+ * 前端实时预览架构：Step5 无需等待服务器渲染
  */
 import { message } from "antd"
 import type { TitleSettings } from "../types"
@@ -16,8 +16,6 @@ export interface UseStepNavigationOptions {
   selectedMaterials: string[]
   smartSelectedIds: string[]
   titleSettings: TitleSettings
-  /** 预览是否就绪（前端素材已加载） */
-  previewReady: boolean
 }
 
 export interface UseStepNavigationReturn {
@@ -34,7 +32,6 @@ export const useStepNavigation = (options: UseStepNavigationOptions): UseStepNav
     selectedMaterials,
     smartSelectedIds,
     titleSettings,
-    previewReady,
   } = options
 
   const goNext = () => {
@@ -50,15 +47,10 @@ export const useStepNavigation = (options: UseStepNavigationOptions): UseStepNav
       message.warning("请先进行智能匹配并选择素材")
       return
     }
-    // Step3 配音：配音为可选项，不强制校验，用户可跳过
     if (currentStep === 4 && !titleSettings.title.trim()) {
       message.warning("请选择或输入标题")
       return
     }
-    if (currentStep === 5 && !previewReady) {
-      message.warning("请先选择素材以预览效果")
-      return
-    }
     if (currentStep < 7) {
       setCurrentStep((s) => s + 1)
     }

apps/web/src/pages/generate/utils/calculateResolution.ts

@@ -0,0 +1,46 @@
+/**
+ * 根据视频比例计算输出分辨率
+ *
+ * 规则：
+ * - 长边固定 1920
+ * - 短边按 1920 × (短/长) 计算，对齐到偶数
+ * - 9:16 → 1080×1920（竖屏）
+ * - 16:9 → 1920×1080（横屏）
+ * - 1:1   → 1920×1920
+ * - 无法解析时 fallback 到 1080×1920
+ */
+export interface Resolution {
+  width: number
+  height: number
+}
+
+export function calculateResolution(ratio: string): Resolution {
+  if (!ratio) return { width: 1080, height: 1920 }
+
+  if (ratio.includes(":")) {
+    const [rw, rh] = ratio.split(":").map(Number)
+    if (rw > 0 && rh > 0) {
+      const [longSide, shortSide] = rw < rh ? [rh, rw] : [rw, rh]
+      const baseLong = 1920
+      const baseShort = Math.round((baseLong * shortSide) / longSide)
+      const evenShort = baseShort - (baseShort % 2)
+      if (rw < rh) {
+        // 竖屏：短边是宽，长边是高
+        return { width: evenShort, height: baseLong }
+      } else {
+        // 横屏：长边是宽，短边是高
+        return { width: baseLong, height: evenShort }
+      }
+    }
+    return { width: 1080, height: 1920 }
+  }
+
+  if (ratio.includes("x")) {
+    const [wStr, hStr] = ratio.split("x")
+    const w = parseInt(wStr, 10) || 1080
+    const h = parseInt(hStr, 10) || 1920
+    return { width: w, height: h }
+  }
+
+  return { width: 1080, height: 1920 }
+}

apps/worker/video_processing/render_adapter.py

@@ -84,6 +84,7 @@ class RenderAdapterResult:
     cover_candidates: list[dict] | None = (
         None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
     )
+    temp_dir: str | None = None  # 渲染临时目录，成功时由调用方清理，失败时由 finally 清理
 
     def __post_init__(self):
         if self.rendered_clip_ids is None:
@@ -200,7 +201,7 @@ class RenderAdapter:
             self._report_progress(progress_cb, 35.0, "准备 BGM 音频")
 
             # 3~6. 统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）
-            return self._do_render(
+            result = self._do_render(
                 plan=plan,
                 clips=ready_clips,
                 asset_path_map=asset_path_map,
@@ -212,6 +213,11 @@ class RenderAdapter:
                 failed_clip_ids=failed_clip_ids,
                 voiceover_audio_path=voiceover_audio_path,
             )
+            # 成功时将临时目录所有权转移给调用方，阻止 finally 清理
+            if result.success and temp_dir:
+                result.temp_dir = temp_dir
+                temp_dir = None  # 阻止 finally 块清理
+            return result
 
         except subprocess.CalledProcessError as exc:
             stderr_text = (exc.stderr or "").strip()

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -41,7 +41,6 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         output_width=getattr(model, "output_width", 1280) or 1280,
         output_height=getattr(model, "output_height", 720) or 720,
         cover_url=getattr(model, "cover_url", "") or "",
-        custom_title=getattr(model, "custom_title", "") or "",
         title_config=dict(getattr(model, "title_config", {}) or {}),
         logs=model.logs or "[]",
         created_at=model.created_at,
@@ -86,7 +85,6 @@ class SQLAlchemyGenerationTaskRepository:
             output_width=task.output_width,
             output_height=task.output_height,
             cover_url=task.cover_url or "",
-            custom_title=task.custom_title or "",
             title_config=dict(task.title_config) if task.title_config else {},
             logs=task.logs,
             created_at=task.created_at,
@@ -274,7 +272,6 @@ class SQLAlchemyGenerationTaskRepository:
         model.output_width = task.output_width
         model.output_height = task.output_height
         model.cover_url = task.cover_url or ""
-        model.custom_title = task.custom_title or ""
         model.title_config = dict(task.title_config) if task.title_config else {}
         model.logs = task.logs
         self.session.commit()

packages/application/generation_tasks.py

@@ -31,7 +31,6 @@ class CreateGenerationTaskCommand:
     output_width: int = 1280
     output_height: int = 720
     cover_url: str = ""
-    custom_title: str = ""
     title_config: dict = field(default_factory=dict)
 
 
@@ -68,7 +67,6 @@ class CreateGenerationTaskUseCase:
             output_width=command.output_width,
             output_height=command.output_height,
             cover_url=command.cover_url,
-            custom_title=command.custom_title,
             title_config=command.title_config,
         )
         return self.generation_task_repository.create(task)

packages/domain/generation_task.py

@@ -120,7 +120,6 @@ class GenerationTask:
     output_width: int = 1280
     output_height: int = 720
     cover_url: str = ""
-    custom_title: str = ""
     title_config: dict = field(default_factory=dict)
     extra_meta: dict = field(default_factory=dict)
     logs: str = "[]"
@@ -153,7 +152,6 @@ class GenerationTask:
         output_width: int = 1280,
         output_height: int = 720,
         cover_url: str = "",
-        custom_title: str = "",
         title_config: dict | None = None,
         extra_meta: dict | None = None,
     ) -> "GenerationTask":
@@ -185,7 +183,6 @@ class GenerationTask:
             output_width=output_width,
             output_height=output_height,
             cover_url=cover_url,
-            custom_title=custom_title,
             title_config=dict(title_config) if title_config else {},
             extra_meta=dict(extra_meta) if extra_meta else {},
         )
@@ -306,10 +303,10 @@ class GenerationTask:
         self,
         *,
         cover_url: str = "",
-        custom_title: str = "",
         extra_meta: dict | None = None,
         output_width: int = 0,
         output_height: int = 0,
+        title_config: dict | None = None,
     ) -> None:
         """将预览任务确认为正式产出。
 
@@ -319,12 +316,12 @@ class GenerationTask:
         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
+        if title_config:
+            self.title_config = dict(title_config)
         if extra_meta:
             self.extra_meta.update(extra_meta)
         self.updated_at = datetime.now(timezone.utc)

tests/integration/test_four_mode_rendering.py

@@ -1,234 +0,0 @@
-"""剪辑模式渲染集成测试.
-
-验证剪辑模式（ONE_TAKE / VOICE_OVER）通过
-_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。
-注：PIP / VOICE_PIP 已下线，统一映射为 ONE_TAKE。
-
-需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。
-"""
-
-from __future__ import annotations
-
-import shutil
-import subprocess
-import tempfile
-from pathlib import Path
-
-import pytest
-from video_processing.unified_render_service import (
-    RenderResult,
-    UnifiedRenderService,
-    _resolve_layer_role,
-)
-from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-pytestmark = pytest.mark.skipif(
-    not shutil.which("ffmpeg"),
-    reason="ffmpeg not available",
-)
-
-
-# ── 辅助函数 ──────────────────────────────────────────────────────────────────
-
-
-def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None:
-    """生成一个纯色测试视频。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"color=c={color}:s=640x360:d={duration}:r=25",
-        "-c:v",
-        "libx264",
-        "-pix_fmt",
-        "yuv420p",
-        "-movflags",
-        "+faststart",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-def _render_with_mode(
-    mode: str,
-    num_clips: int = 3,
-    duration: float = 2.0,
-) -> tuple[RenderResult, Path]:
-    """用指定模式生成测试视频并渲染，返回 (result, work_dir)。
-
-    调用方负责清理 work_dir。
-    """
-    work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_"))
-
-    # 生成测试视频素材
-    colors = ["red", "green", "blue", "yellow", "purple"]
-    downloaded_paths: list[Path] = []
-    for i in range(num_clips):
-        p = work_dir / f"test_{i:03d}.mp4"
-        _generate_test_video(p, duration=duration, color=colors[i % len(colors)])
-        downloaded_paths.append(p)
-
-    # 构建虚拟 plan + clips
-    task_id = f"test_task_{mode}"
-    plan, clips, asset_path_map = _build_plan_and_clips_from_task(
-        task_id=task_id,
-        downloaded_paths=downloaded_paths,
-        mode=mode,
-    )
-
-    # 渲染
-    service = UnifiedRenderService(
-        plan=plan,
-        clips=clips,
-        asset_path_map=asset_path_map,
-        work_dir=work_dir,
-        output_width=640,
-        output_height=360,
-        output_fps=25,
-    )
-    result = service.render()
-    return result, work_dir
-
-
-# ── 测试 _build_plan_and_clips_from_task ──────────────────────────────────────
-
-
-class TestBuildPlanAndClips:
-    """测试 4 种模式的虚拟 plan 构建。"""
-
-    def _make_paths(self, n: int) -> list[Path]:
-        return [Path(f"/tmp/test_{i}.mp4") for i in range(n)]
-
-    def test_one_take_mode(self):
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take")
-
-        assert plan.id == "t1"
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-        assert len(asset_map) == 3
-
-    def test_pip_mode_maps_to_one_take(self):
-        """PIP 已下线，映射为 one_take → 全部 main clips。"""
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip")
-
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_voice_over_mode(self):
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over")
-
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-        assert all(c.config.get("role") == "b_roll" for c in clips)
-
-    def test_voice_pip_mode_maps_to_one_take(self):
-        """VOICE_PIP 已下线，映射为 one_take → 全部 main clips。"""
-        paths = self._make_paths(4)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip")
-
-        assert len(clips) == 4
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_unknown_mode_defaults_to_one_take(self):
-        paths = self._make_paths(2)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode")
-
-        assert len(clips) == 2
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_asset_path_map_keys_match_clip_asset_ids(self):
-        paths = self._make_paths(3)
-        _, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take")
-
-        clip_asset_ids = {c.asset_id for c in clips}
-        map_keys = set(asset_map.keys())
-        assert clip_asset_ids == map_keys
-
-
-# ── 测试图层分组（4 模式） ────────────────────────────────────────────────────
-
-
-class TestFourModeLayerGrouping:
-    """验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。"""
-
-    def test_one_take_layers(self):
-        """ONE_TAKE: 3 main → 1 main layer。"""
-        paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-    def test_pip_layers_now_one_take(self):
-        """PIP 已下线 → one_take: 3 main → 1 main layer。"""
-        paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-    def test_voice_over_layers(self):
-        """VOICE_OVER: 3 main(b_roll) → broll。"""
-        paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"broll"}
-
-    def test_voice_pip_layers_now_one_take(self):
-        """VOICE_PIP 已下线 → one_take: 4 main → main layer。"""
-        paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)]
-        _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-
-# ── 端到端渲染测试（需要 ffmpeg） ─────────────────────────────────────────────
-
-
-class TestEndToEndRendering:
-    """4 种模式的完整渲染测试，验证输出文件存在且时长合理。"""
-
-    def test_one_take_render(self):
-        result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-            assert result.width == 640
-            assert result.height == 360
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_pip_render(self):
-        result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_voice_over_render(self):
-        result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_voice_pip_render(self):
-        result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)

tests/integration/test_full_pipeline.py

@@ -1,189 +0,0 @@
-"""全链路集成测试.
-
-验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。
-"""
-
-from __future__ import annotations
-
-import shutil
-import subprocess
-import tempfile
-from pathlib import Path
-
-import pytest
-from video_processing.unified_render_service import (
-    UnifiedRenderService,
-)
-from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-pytestmark = pytest.mark.skipif(
-    not shutil.which("ffmpeg"),
-    reason="ffmpeg not available",
-)
-
-
-def _generate_test_video(path: Path, duration: float = 3.0) -> None:
-    """生成一个测试视频。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"color=c=blue:s=640x360:d={duration}:r=25",
-        "-c:v",
-        "libx264",
-        "-pix_fmt",
-        "yuv420p",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-def _generate_test_audio(path: Path, duration: float = 5.0) -> None:
-    """生成一个测试音频文件。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"sine=frequency=440:duration={duration}",
-        "-c:a",
-        "aac",
-        "-b:a",
-        "128k",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-class TestFullPipeline:
-    """验证从虚拟 plan 构建到渲染输出的完整流程。"""
-
-    def test_one_take_pipeline(self):
-        """ONE_TAKE 模式完整流程。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            # 生成测试素材
-            paths = []
-            for i in range(3):
-                p = work_dir / f"clip_{i}.mp4"
-                _generate_test_video(p, duration=2.0)
-                paths.append(p)
-
-            # 构建虚拟 plan
-            plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take")
-
-            # 渲染
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            result = service.render()
-
-            assert result.output_path.exists()
-            assert result.duration > 0
-            assert result.file_size > 0
-            assert result.width == 640
-            assert result.height == 360
-
-    def test_pipeline_with_audio_mux(self):
-        """渲染 + 混音后处理。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            # 生成测试素材
-            video_path = work_dir / "clip_0.mp4"
-            _generate_test_video(video_path, duration=3.0)
-
-            # 构建虚拟 plan
-            plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take")
-
-            # 渲染
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            render_result = service.render()
-
-            # 混音 - 直接用 ffmpeg（_mux_audio_track 已被清理）
-            audio_path = work_dir / "voice.aac"
-            _generate_test_audio(audio_path, duration=5.0)
-
-            final_path = work_dir / "final.mp4"
-            mux_cmd = [
-                "ffmpeg",
-                "-y",
-                "-i",
-                str(render_result.output_path),
-                "-i",
-                str(audio_path),
-                "-c:v",
-                "copy",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(final_path),
-            ]
-            subprocess.run(mux_cmd, check=True, capture_output=True, timeout=30)
-
-            assert final_path.exists()
-            assert final_path.stat().st_size > 0
-
-    def test_single_clip_pipeline(self):
-        """单 clip 渲染（无转场）。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            video_path = work_dir / "single.mp4"
-            _generate_test_video(video_path, duration=5.0)
-
-            plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take")
-
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            result = service.render()
-
-            assert result.output_path.exists()
-            assert result.duration > 0
-
-    def test_dedup_helper_integration(self):
-        """验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。"""
-        # 只验证函数存在且签名正确（不实际调用，需要数据库）
-        import inspect
-
-        from video_processing.dedup_helpers import create_video_record_and_dedup
-
-        sig = inspect.signature(create_video_record_and_dedup)
-        params = set(sig.parameters.keys())
-        expected = {
-            "generation_task_id",
-            "project_id",
-            "batch_id",
-            "file_url",
-            "file_size",
-            "duration",
-            "video_path",
-            "mode",
-            "session",
-            "width",
-            "height",
-            "fps",
-        }
-        assert expected.issubset(params), f"Missing params: {expected - params}"

tests/unit/test_1280_preview_speedup.py

@@ -149,14 +149,6 @@ class TestWorkerGenerationNoPreviewOverride:
         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"
-
 
 # ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ──
 

tests/unit/test_1294_preview_voice_injection.py

@@ -2,7 +2,6 @@
 
 验证：
 1. _load_task_info 正确加载 voice_ids
-2. _render_video 接受 voice_ids 参数
 3. voice_ids 正确注入到 plan config 中（实际执行代码路径，diff-cover 可达）
 """
 
@@ -74,185 +73,3 @@ class TestLoadTaskInfoVoiceIds:
 
             result = _load_task_info("test_task_id")
             assert result["voice_ids"] == []
-
-
-class TestRenderVideoVoiceInjection:
-    """验证 _render_video 正确注入 voice_id 到 plan config（实际执行代码路径）"""
-
-    def test_render_video_accepts_voice_ids(self):
-        """_render_video 签名包含 voice_ids 参数"""
-        import inspect
-
-        from worker_app.tasks.generation import _render_video
-
-        sig = inspect.signature(_render_video)
-        assert "voice_ids" in sig.parameters
-
-    def test_voice_ids_default_none(self):
-        """voice_ids 参数默认为 None"""
-        import inspect
-
-        from worker_app.tasks.generation import _render_video
-
-        sig = inspect.signature(_render_video)
-        param = sig.parameters["voice_ids"]
-        assert param.default is None
-
-    def test_voice_ids_injected_into_plan_config(self):
-        """voice_ids 非空时，voice_id 和 subtitle.auto_generated 被注入到 plan config。
-
-        此测试实际执行 _render_video 的配音注入代码路径，确保 diff-cover 覆盖新增行。
-        """
-        from dataclasses import dataclass, field
-
-        @dataclass
-        class MockClip:
-            """模拟 VirtualClip，至少需要 duration 属性。"""
-
-            id: str = "clip_1"
-            duration: float = 5.0
-            config: dict = field(default_factory=dict)
-
-        @dataclass
-        class MockPlan:
-            """模拟 VirtualPlan，至少需要 config 属性。"""
-
-            id: str = "test_plan"
-            name: str = "test"
-            config: dict = field(default_factory=dict)
-
-        mock_plan = MockPlan(config={"some_key": "some_value"})
-        mock_clips = [MockClip(duration=5.0), MockClip(duration=3.0)]
-        mock_asset_path_map = {"asset_1": Path("/tmp/video1.mp4")}
-
-        # Mock RenderAdapter 和 render 结果
-        mock_render_result = MagicMock()
-        mock_render_result.success = True
-        mock_render_result.output_path = Path("/tmp/output.mp4")
-        mock_render_result.duration = 8.0
-
-        mock_adapter_cls = MagicMock(return_value=MagicMock())
-        mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result
-
-        mock_db = MagicMock()
-
-        with (
-            patch(
-                "worker_app.tasks.generation._build_plan_and_clips_from_task",
-                return_value=(mock_plan, mock_clips, mock_asset_path_map),
-            ),
-            patch(
-                "worker_app.tasks.generation._load_template_plan_config",
-                return_value=None,
-            ),
-            patch(
-                "video_processing.render_adapter.RenderAdapter",
-                mock_adapter_cls,
-            ),
-            patch(
-                "worker_app.tasks.generation.SessionLocal",
-                return_value=mock_db,
-            ),
-        ):
-            from worker_app.tasks.generation import _render_video
-
-            from packages.domain import EditingMode
-
-            output_path, render_duration, cover_candidates = _render_video(
-                task_id="test_task_123",
-                downloaded_videos=[Path("/tmp/video1.mp4")],
-                voice_path=None,
-                editing_mode=EditingMode.ONE_TAKE,
-                project_id="proj_1",
-                template_id="tmpl_1",
-                user_id="user_1",
-                temp_path=Path("/tmp"),
-                output_name="test_output.mp4",
-                resolution="854x480",
-                voice_ids=["voice_abc"],
-            )
-
-            # 验证 voice_id 被注入到 plan config（覆盖新增代码行）
-            assert mock_plan.config.get("voice_id") == "voice_abc"
-            # 验证 subtitle.auto_generated 被设置为 True
-            assert mock_plan.config.get("subtitle", {}).get("auto_generated") is True
-            # 验证 RenderAdapter 被调用
-            mock_adapter_cls.return_value.render_from_memory.assert_called_once()
-            # 验证返回值
-            assert output_path == Path("/tmp/output.mp4")
-            assert render_duration == 8.0
-
-    def test_voice_ids_empty_skips_injection(self):
-        """voice_ids 为空时，不注入 voice_id 到 plan config"""
-        from dataclasses import dataclass, field
-
-        @dataclass
-        class MockClip:
-            id: str = "clip_1"
-            duration: float = 5.0
-
-        @dataclass
-        class MockPlan:
-            id: str = "test_plan"
-            name: str = "test"
-            config: dict = field(default_factory=dict)
-
-        mock_plan = MockPlan(config={"export": {"resolution": "854x480"}})
-        mock_clips = [MockClip(duration=5.0)]
-
-        mock_render_result = MagicMock()
-        mock_render_result.success = True
-        mock_render_result.output_path = Path("/tmp/output.mp4")
-        mock_render_result.duration = 5.0
-
-        mock_adapter_cls = MagicMock(return_value=MagicMock())
-        mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result
-
-        with (
-            patch(
-                "worker_app.tasks.generation._build_plan_and_clips_from_task",
-                return_value=(mock_plan, mock_clips, {}),
-            ),
-            patch(
-                "worker_app.tasks.generation._load_template_plan_config",
-                return_value=None,
-            ),
-            patch(
-                "video_processing.render_adapter.RenderAdapter",
-                mock_adapter_cls,
-            ),
-            patch(
-                "worker_app.tasks.generation.SessionLocal",
-                return_value=MagicMock(),
-            ),
-        ):
-            from worker_app.tasks.generation import _render_video
-
-            from packages.domain import EditingMode
-
-            _render_video(
-                task_id="test_task_456",
-                downloaded_videos=[Path("/tmp/video1.mp4")],
-                voice_path=None,
-                editing_mode=EditingMode.ONE_TAKE,
-                project_id="proj_1",
-                template_id="",
-                user_id="user_1",
-                temp_path=Path("/tmp"),
-                output_name="test_output.mp4",
-                voice_ids=[],
-            )
-
-            # 验证 voice_id 没有被注入
-            assert "voice_id" not in mock_plan.config
-
-
-class TestGenerateVideoPassesVoiceIds:
-    """验证 generate_video 调用 _render_video 时传递 voice_ids"""
-
-    def test_generate_video_passes_voice_ids(self):
-        """generate_video 中 _render_video 调用包含 voice_ids 参数"""
-        with open("apps/worker/worker_app/tasks/generation.py", "r") as f:
-            content = f.read()
-
-        assert 'voice_ids=task_info.get("voice_ids", [])' in content

tests/unit/test_bool_config_defense.py

@@ -1,130 +0,0 @@
-"""
-测试：模板 config 字段存储了非 dict 值（如 True / False / str）时，
-渲染链路不会崩溃（'bool' object has no attribute 'get'）。
-
-覆盖两个关键文件：
-1. generation.py — _load_template_plan_config 旧系统路径
-2. unified_render_service.py — _maybe_generate_ass
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# Add worker app to path
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-
-class TestLoadTemplatePlanConfigBoolDefense:
-    """_load_template_plan_config 旧系统路径对非 dict 值的防护。"""
-
-    def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg):
-        """通过 mock 新模板系统返回 None，强制走旧模板系统 fallback 路径。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        mock_old_template = MagicMock()
-        mock_old_template.title_config = title_cfg
-        mock_old_template.subtitle_config = subtitle_cfg
-        mock_old_template.bgm_config = bgm_cfg
-
-        mock_session = MagicMock()
-        # 旧系统 query 返回 mock template
-        mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template
-
-        # Mock 新模板系统 repo.get() 返回 None（强制走 fallback）
-        mock_repo_cls = MagicMock()
-        mock_repo_cls.return_value.get.return_value = None
-
-        with (
-            patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session),
-            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls),
-            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()),
-        ):
-            return _load_template_plan_config("fake-id")
-
-    def test_bool_values_return_empty(self):
-        """title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉"""
-        result = self._call_old_path(True, False, "not_a_dict")
-        assert isinstance(result, dict)
-        assert "title" not in result
-        assert "subtitle" not in result
-        assert "bgm" not in result
-
-    def test_valid_dict_passes_through(self):
-        """正常 dict 正常传递"""
-        result = self._call_old_path(
-            {"text": "标题", "enabled": True},
-            {"text": "副标题"},
-            {"enabled": True, "source": "test.mp3"},
-        )
-        assert result["title"] == {"text": "标题", "enabled": True}
-        assert result["subtitle"] == {"text": "副标题"}
-        assert result["bgm"] == {"enabled": True, "source": "test.mp3"}
-
-    def test_none_returns_empty(self):
-        """None → 空 dict"""
-        result = self._call_old_path(None, None, None)
-        assert result == {}
-
-    def test_mixed_valid_and_invalid(self):
-        """部分有效、部分无效时只保留有效的"""
-        result = self._call_old_path({"text": "OK"}, True, None)
-        assert "title" in result
-        assert "subtitle" not in result
-        assert "bgm" not in result
-
-    def test_int_and_list_also_filtered(self):
-        """int / list 类型也被过滤"""
-        result = self._call_old_path(42, [1, 2, 3], 0)
-        assert result == {}
-
-
-class TestUnifiedRenderBoolConfigDefense:
-    """_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。"""
-
-    def _make_service(self, config):
-        from video_processing.unified_render_service import UnifiedRenderService
-
-        service = UnifiedRenderService.__new__(UnifiedRenderService)
-        mock_plan = MagicMock()
-        mock_plan.config = config
-        service.plan = mock_plan
-        service.task_id = "test-task"
-        return service
-
-    def test_bool_title_does_not_crash(self):
-        """config['title']=True → 不崩溃，返回 None"""
-        service = self._make_service({"title": True, "subtitle": {}})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_bool_subtitle_does_not_crash(self):
-        """config['subtitle']=False → 不崩溃，返回 None"""
-        service = self._make_service({"title": {}, "subtitle": False})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_str_title_does_not_crash(self):
-        """config['title']='plain string' → 不崩溃"""
-        service = self._make_service({"title": "plain string", "subtitle": {}})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_none_config_does_not_crash(self):
-        """config=None → 不崩溃"""
-        service = self._make_service(None)
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_int_title_does_not_crash(self):
-        """config['title']=42 → 不崩溃"""
-        service = self._make_service({"title": 42, "subtitle": 0})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None

tests/unit/test_confirm_generation.py

@@ -5,7 +5,7 @@
   - 预览任务未完成 → 创建新任务走渲染流程
   - 预览任务不存在 → 404
   - 权限不足 → 403
-  - cover_url 和 custom_title 正确传递
+  - cover_url 正确传递
 """
 
 from __future__ import annotations
@@ -201,7 +201,6 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask:
         output_width=1080,
         output_height=1920,
         cover_url="",
-        custom_title="",
         video_title="",
         resolution="",
         bgm_config={},
@@ -233,7 +232,6 @@ class TestConfirmGenerationReuse:
                 "output_width": 1080,
                 "output_height": 1920,
                 "cover_url": "https://example.com/cover.jpg",
-                "custom_title": "我的视频",
             },
         )
 
@@ -251,7 +249,6 @@ class TestConfirmGenerationReuse:
         assert item["output_height"] == 1920
         # 封面和标题更新
         assert item["cover_url"] == "https://example.com/cover.jpg"
-        assert item["custom_title"] == "我的视频"
 
         # 没有创建新任务
         assert len(gen_task_repo._store) == initial_count
@@ -277,7 +274,6 @@ class TestConfirmGenerationReuse:
         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,
@@ -356,7 +352,6 @@ class TestConfirmGenerationErrors:
         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"
@@ -368,7 +363,7 @@ class TestConfirmGenerationErrors:
         client: TestClient,
         gen_task_repo: StubGenerationTaskRepository,
     ) -> None:
-        """cover_url 和 custom_title 正确传递"""
+        """cover_url 正确传递"""
         preview = _make_preview_task()
         gen_task_repo.create(preview)
 
@@ -378,14 +373,12 @@ class TestConfirmGenerationErrors:
                 "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]
         assert item["cover_url"] == "https://cdn.example.com/my-cover.png"
-        assert item["custom_title"] == "测试视频标题"
 
     def test_confirm_default_resolution(
         self,

tests/unit/test_create_reuse_preview_fallback.py

@@ -0,0 +1,409 @@
+"""create 端点兜底复用预览产物 + confirm 标题同步 — 单元测试."""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+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
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+# ── Stubs ────────────────────────────────────────────────────────────────────
+
+
+class StubGenerationTaskRepository:
+    def __init__(self) -> None:
+        self._store: dict[str, Any] = {}
+
+    def create(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id: str) -> Optional[Any]:
+        return self._store.get(task_id)
+
+    def update(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def list_by_user(self, user_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.created_by_user_id == user_id]
+
+    def count_pending_by_user(self, user_id: str) -> int:
+        return len(
+            [
+                t
+                for t in self._store.values()
+                if t.created_by_user_id == user_id and t.status == GenerationTaskStatus.PENDING
+            ]
+        )
+
+    def count_pending_total(self) -> int:
+        return len([t for t in self._store.values() if t.status == GenerationTaskStatus.PENDING])
+
+    def count_by_user(self, user_id: str) -> int:
+        return len([t for t in self._store.values() if t.created_by_user_id == user_id])
+
+
+@dataclass
+class FakeProject:
+    id: str = "project-001"
+    owner_user_id: str = "user-001"
+    shared_users: list[str] = field(default_factory=list)
+    name: str = "Test Project"
+
+    def can_access(self, user_id: str) -> bool:
+        return user_id == self.owner_user_id or user_id in self.shared_users
+
+
+class StubProjectRepository:
+    def __init__(self) -> None:
+        self._projects: dict[str, FakeProject] = {}
+
+    def add(self, project: FakeProject) -> None:
+        self._projects[project.id] = project
+
+    def find_by_id(self, project_id: str) -> Optional[FakeProject]:
+        return self._projects.get(project_id)
+
+
+@dataclass
+class FakeUser:
+    id: str = "user-001"
+    email: str = "test@example.com"
+
+
+@dataclass
+class FakeAuthenticatedUser:
+    user: FakeUser = field(default_factory=FakeUser)
+    session_id: str | None = None
+    token_type: str | None = None
+
+
+def _make_preview_task(**kwargs: Any) -> GenerationTask:
+    defaults = dict(
+        id="preview-task-001",
+        project_id="project-001",
+        asset_library_id="library-001",
+        strategy_id="one_take",
+        voice_library_id="",
+        template_id="tmpl-001",
+        asset_ids=["asset-1"],
+        title_ids=[],
+        voice_ids=[],
+        status=GenerationTaskStatus.COMPLETED,
+        progress=100.0,
+        result_count=1,
+        error_message="",
+        created_by_user_id="user-001",
+        source_edit_plan_id="plan-001",
+        asset_select_mode="all",
+        is_preview=True,
+        source_task_id="",
+        output_width=1080,
+        output_height=1920,
+        cover_url="",
+        video_title="",
+        resolution="",
+        bgm_config={},
+        title_config={"text": "预览标题"},
+    )
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+def _make_db_with_preview(preview: GenerationTask):
+    """Create a mock DB that returns a model-like object for the preview."""
+    db = MagicMock()
+    mock_model = MagicMock()
+    mock_model.id = preview.id
+    mock_model.output_width = preview.output_width
+    mock_model.output_height = preview.output_height
+    # Set up chain: db.query(...).filter(...).order_by(...).first()
+    chain = db.query.return_value
+    chain.filter.return_value = chain
+    chain.order_by.return_value = chain
+    chain.first.return_value = mock_model
+    return db, mock_model
+
+
+def _make_db_empty():
+    """Create a mock DB that returns None (no preview found)."""
+    db = MagicMock()
+    chain = db.query.return_value
+    chain.filter.return_value = chain
+    chain.order_by.return_value = chain
+    chain.first.return_value = None
+    return db
+
+
+# ── Fixtures ─────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def gen_task_repo() -> StubGenerationTaskRepository:
+    return StubGenerationTaskRepository()
+
+
+@pytest.fixture
+def project_repo() -> StubProjectRepository:
+    repo = StubProjectRepository()
+    repo.add(FakeProject())
+    return repo
+
+
+@pytest.fixture
+def app(
+    gen_task_repo: StubGenerationTaskRepository,
+    project_repo: StubProjectRepository,
+) -> FastAPI:
+    from app.api.routes.generation_tasks import router
+    from app.auth import get_current_user
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_db_session,
+        get_generated_video_repository,
+        get_generation_task_repository,
+        get_project_repository,
+    )
+
+    test_app = FastAPI()
+    test_app.include_router(router, prefix="/api/v1/generation")
+
+    test_app.dependency_overrides[get_current_user] = lambda: FakeAuthenticatedUser()
+    test_app.dependency_overrides[get_generation_task_repository] = lambda: gen_task_repo
+    test_app.dependency_overrides[get_project_repository] = lambda: project_repo
+    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()
+    # db_session will be overridden per-test
+
+    yield test_app
+    test_app.dependency_overrides.clear()
+
+
+def _make_client(app: FastAPI, db: MagicMock) -> TestClient:
+    from app.dependencies import get_db_session
+
+    app.dependency_overrides[get_db_session] = lambda: db
+    return TestClient(app)
+
+
+# ── Domain: mark_confirmed with title_config ─────────────────────────────────
+
+
+class TestMarkConfirmedTitleConfig:
+    def test_mark_confirmed_sets_title_config(self):
+        task = _make_preview_task()
+        task.mark_confirmed(title_config={"text": "新标题"})
+        assert task.is_preview is False
+        assert task.title_config["text"] == "新标题"
+
+    def test_mark_confirmed_without_title_config_preserves_existing(self):
+        task = _make_preview_task(title_config={"text": "原标题"})
+        task.mark_confirmed()
+        assert task.title_config["text"] == "原标题"
+
+
+# ── Create endpoint fallback ─────────────────────────────────────────────────
+
+
+class TestCreateEndpointFallback:
+    """create 端点兜底复用预览产物。"""
+
+    def test_reuse_completed_preview(
+        self,
+        app: FastAPI,
+        gen_task_repo: StubGenerationTaskRepository,
+    ):
+        """带 source_edit_plan_id + is_preview=False → 复用已完成预览"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+        db, _ = _make_db_with_preview(preview)
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generation_task_repository._to_domain",
+                return_value=preview,
+            ),
+            patch(
+                "app.api.routes.generation_tasks._writeback_edit_plan_config",
+            ),
+        ):
+            client = _make_client(app, db)
+            resp = client.post(
+                "/api/v1/generation/tasks",
+                json={
+                    "project_id": "project-001",
+                    "template_id": "tmpl-001",
+                    "asset_ids": ["asset-1"],
+                    "source_edit_plan_id": "plan-001",
+                    "is_preview": False,
+                    "output_width": 1080,
+                    "output_height": 1920,
+                },
+            )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] == 1
+        assert data["items"][0]["id"] == preview.id
+        assert data["items"][0]["is_preview"] is False
+
+    def test_no_preview_found_creates_new_task(
+        self,
+        app: FastAPI,
+        gen_task_repo: StubGenerationTaskRepository,
+    ):
+        """没有已完成预览 → 正常创建新任务"""
+        db = _make_db_empty()
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            client = _make_client(app, db)
+            resp = client.post(
+                "/api/v1/generation/tasks",
+                json={
+                    "project_id": "project-001",
+                    "template_id": "tmpl-001",
+                    "asset_ids": ["asset-1"],
+                    "source_edit_plan_id": "plan-002",
+                    "is_preview": False,
+                },
+            )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] == 1
+        assert data["items"][0]["id"] != "preview-task-001"
+        assert data["items"][0]["is_preview"] is False
+
+    def test_preview_request_does_not_use_fallback(
+        self,
+        app: FastAPI,
+    ):
+        """is_preview=True → 不走兜底，正常创建预览任务"""
+        db = _make_db_empty()
+
+        with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
+            client = _make_client(app, db)
+            resp = client.post(
+                "/api/v1/generation/tasks",
+                json={
+                    "project_id": "project-001",
+                    "template_id": "tmpl-001",
+                    "asset_ids": ["asset-1"],
+                    "source_edit_plan_id": "plan-001",
+                    "is_preview": True,
+                },
+            )
+
+        assert resp.status_code == 200
+        assert resp.json()["items"][0]["is_preview"] is True
+        # 兜底查询 GenerationTaskModel 不应被调用（只可能查 EditPlanModel 做自动关联）
+        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+        for call_args in db.query.call_args_list:
+            assert (
+                call_args[0][0] is not GenerationTaskModel
+            ), "fallback should not query GenerationTaskModel for preview requests"
+
+    def test_fallback_resolution_mismatch_creates_new(
+        self,
+        app: FastAPI,
+        gen_task_repo: StubGenerationTaskRepository,
+    ):
+        """兜底找到预览但分辨率不一致 → 跳过复用，创建新任务"""
+        preview = _make_preview_task(output_width=1080, output_height=1920)
+        gen_task_repo.create(preview)
+        db, _ = _make_db_with_preview(preview)
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generation_task_repository._to_domain",
+                return_value=preview,
+            ),
+            patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True),
+        ):
+            client = _make_client(app, db)
+            resp = client.post(
+                "/api/v1/generation/tasks",
+                json={
+                    "project_id": "project-001",
+                    "template_id": "tmpl-001",
+                    "asset_ids": ["asset-1"],
+                    "source_edit_plan_id": "plan-001",
+                    "is_preview": False,
+                    "output_width": 1920,
+                    "output_height": 1080,
+                },
+            )
+
+        assert resp.status_code == 200
+        assert resp.json()["items"][0]["id"] != preview.id
+
+
+# ── Confirm endpoint title sync ──────────────────────────────────────────────
+
+
+class TestConfirmTitleSync:
+    """confirm 端点 custom_title 同步到 title_config。"""
+
+    def test_confirm_with_custom_title_updates_title_config(
+        self,
+        app: FastAPI,
+        gen_task_repo: StubGenerationTaskRepository,
+    ):
+        """传了 custom_title → title_config.text 被更新"""
+        preview = _make_preview_task(
+            title_config={"text": "旧标题", "font_size": 32},
+            source_edit_plan_id="plan-001",
+        )
+        gen_task_repo.create(preview)
+        db = _make_db_empty()
+
+        with patch("app.api.routes.generation_tasks._writeback_edit_plan_config"):
+            client = _make_client(app, db)
+            resp = client.post(
+                f"/api/v1/generation/tasks/{preview.id}/confirm",
+                json={
+                    "output_width": 1080,
+                    "output_height": 1920,
+                    "custom_title": "新标题",
+                },
+            )
+
+        assert resp.status_code == 200
+        item = resp.json()["items"][0]
+        assert item["title_config"]["text"] == "新标题"
+        assert item["title_config"]["font_size"] == 32
+
+    def test_confirm_without_custom_title_preserves_title(
+        self,
+        app: FastAPI,
+        gen_task_repo: StubGenerationTaskRepository,
+    ):
+        """没传 custom_title → title_config 不变"""
+        preview = _make_preview_task(title_config={"text": "原标题"})
+        gen_task_repo.create(preview)
+        db = _make_db_empty()
+
+        client = _make_client(app, db)
+        resp = client.post(
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+
+        assert resp.status_code == 200
+        assert resp.json()["items"][0]["title_config"]["text"] == "原标题"

tests/unit/test_empty_plan_id_marks_failed.py

@@ -0,0 +1,32 @@
+"""回归测试：source_edit_plan_id 为空时任务必须被标记为 failed。
+
+背景 (2026-08-24)：确认生成卡在 10%。根因是 worker 在
+source_edit_plan_id 为空时直接 return failed，但没有调用 mark_failed，
+导致 DB 状态永远停在 running。
+"""
+
+from pathlib import Path
+
+GENERATION_PY = Path(__file__).resolve().parents[2] / "apps" / "worker" / "worker_app" / "tasks" / "generation.py"
+
+
+def test_mark_failed_in_else_branch():
+    """generation.py 中 source_edit_plan_id 为空的 else 分支必须调用 mark_failed。"""
+    source = GENERATION_PY.read_text(encoding="utf-8")
+
+    # 定位 else 分支：紧跟在 'source_edit_plan_id 为空' 日志之后的 else 块
+    marker = "source_edit_plan_id 为空"
+    idx = source.find(marker)
+    assert idx != -1, f"generation.py 中未找到 '{marker}'"
+
+    # 从 marker 位置向后搜索到下一个 return 语句
+    after_marker = source[idx:]
+    return_idx = after_marker.find("return {")
+    assert return_idx != -1, "else 分支中未找到 return 语句"
+
+    # 关键断言：marker 和 return 之间必须包含 mark_failed
+    block = source[idx : idx + return_idx]
+    assert "mark_failed" in block, (
+        "else 分支在 return 之前必须调用 _update_task_status(task_id, "
+        "'mark_failed', ...) 以更新 DB 状态，否则任务永远卡在 running"
+    )

tests/unit/test_generation_p3_optimizations.py

@@ -1,389 +0,0 @@
-"""P3 优化单元测试 — generation.py 三项优化.
-
-覆盖：
-  P3-1: _download_library_assets strict 模式
-  P3-2: 归属校验合并到同一 DB session
-  P3-3: _verify_url_accessible HEAD 重试
-"""
-
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
-
-# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
-# worker_app.db 在模块级别调用 ensure_database_exists() 尝试连接 PostgreSQL，
-# 增量测试单独跑这些文件时会失败。与 test_voice_clone_task.py 同理。
-_mock_db_module = MagicMock()
-_mock_db_module.SessionLocal = MagicMock()
-sys.modules.setdefault("worker_app.db", _mock_db_module)
-if "worker_app" in sys.modules:
-    sys.modules["worker_app"].db = _mock_db_module
-
-
-# ── P3-3: _verify_url_accessible 重试 ───────────────────────────────────────
-
-
-class TestVerifyUrlAccessibleRetry:
-    """_verify_url_accessible 重试逻辑."""
-
-    @patch("time.sleep")
-    @patch("urllib.request.OpenerDirector.open")
-    def test_first_attempt_success(self, mock_open, mock_sleep):
-        """首次成功，不重试."""
-        from worker_app.tasks.generation import _verify_url_accessible
-
-        mock_resp = MagicMock()
-        mock_resp.status = 200
-        mock_resp.__enter__ = MagicMock(return_value=mock_resp)
-        mock_resp.__exit__ = MagicMock(return_value=False)
-        mock_open.return_value = mock_resp
-
-        assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_open.call_count == 1
-        mock_sleep.assert_not_called()
-
-    @patch("time.sleep")
-    @patch("urllib.request.OpenerDirector.open")
-    def test_retry_then_success(self, mock_open, mock_sleep):
-        """首次失败，重试后成功."""
-        from worker_app.tasks.generation import _verify_url_accessible
-
-        # 第一次失败（网络异常），第二次成功
-        mock_resp_ok = MagicMock()
-        mock_resp_ok.status = 200
-        mock_resp_ok.__enter__ = MagicMock(return_value=mock_resp_ok)
-        mock_resp_ok.__exit__ = MagicMock(return_value=False)
-
-        mock_open.side_effect = [
-            OSError("connection reset"),
-            mock_resp_ok,
-        ]
-
-        assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_open.call_count == 2
-        mock_sleep.assert_called_once_with(1)
-
-    @patch("time.sleep")
-    @patch("urllib.request.OpenerDirector.open")
-    def test_all_retries_exhausted(self, mock_open, mock_sleep):
-        """全部重试耗尽，返回 False."""
-        from worker_app.tasks.generation import _verify_url_accessible
-
-        mock_open.side_effect = OSError("connection refused")
-
-        assert _verify_url_accessible("https://example.com/file.mp4") is False
-        # 1 首次 + 2 重试 = 3 次
-        assert mock_open.call_count == 3
-        assert mock_sleep.call_count == 2
-
-    @patch("time.sleep")
-    @patch("urllib.request.OpenerDirector.open")
-    def test_http_500_then_success(self, mock_open, mock_sleep):
-        """HTTP 500 后重试成功."""
-        from worker_app.tasks.generation import _verify_url_accessible
-
-        mock_resp_500 = MagicMock()
-        mock_resp_500.status = 500
-        mock_resp_500.__enter__ = MagicMock(return_value=mock_resp_500)
-        mock_resp_500.__exit__ = MagicMock(return_value=False)
-
-        mock_resp_200 = MagicMock()
-        mock_resp_200.status = 200
-        mock_resp_200.__enter__ = MagicMock(return_value=mock_resp_200)
-        mock_resp_200.__exit__ = MagicMock(return_value=False)
-
-        mock_open.side_effect = [mock_resp_500, mock_resp_200]
-
-        assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_open.call_count == 2
-
-    @patch("time.sleep")
-    @patch("urllib.request.OpenerDirector.open")
-    def test_custom_retries_zero(self, mock_open, mock_sleep):
-        """retries=0 时不重试."""
-        from worker_app.tasks.generation import _verify_url_accessible
-
-        mock_open.side_effect = OSError("timeout")
-
-        assert _verify_url_accessible("https://example.com/file.mp4", retries=0) is False
-        assert mock_open.call_count == 1
-        mock_sleep.assert_not_called()
-
-
-# ── P3-1: _download_library_assets strict 模式 ──────────────────────────────
-
-
-def _make_mock_asset(
-    asset_id: str,
-    name: str,
-    file_url: str | None,
-    asset_library_id: str = "lib-1",
-    project_id: str = "",
-):
-    """构造 mock AssetModel 实例."""
-    return SimpleNamespace(
-        id=asset_id,
-        name=name,
-        file_url=file_url,
-        asset_library_id=asset_library_id,
-        project_id=project_id,
-        status="ready",
-        file_type="video",
-        created_at="2026-01-01",
-    )
-
-
-def _setup_mock_session(assets):
-    """构造 mock session，返回 (mock_session, mock_query_chain)."""
-    mock_session = MagicMock()
-    mock_query = MagicMock()
-
-    # chain: session.query().filter().filter().order_by().all()
-    mock_session.query.return_value = mock_query
-    mock_query.filter.return_value = mock_query
-    mock_query.order_by.return_value = mock_query
-    mock_query.all.return_value = assets
-
-    return mock_session
-
-
-class TestDownloadLibraryAssetsStrictMode:
-    """_download_library_assets strict 模式."""
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_strict_mode_raises_on_download_failure(self, mock_download, mock_session_factory):
-        """strict=True 时，单个素材下载失败立即抛 RuntimeError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
-            _make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-
-        # 第一个成功，第二个失败
-        mock_download.side_effect = [True, False]
-
-        with pytest.raises(RuntimeError, match="素材下载失败"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-                asset_ids=["a1", "a2"],
-                strict=True,
-            )
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_non_strict_mode_returns_partial_results(self, mock_download, mock_session_factory):
-        """strict=False 时，跳过失败素材，返回成功列表."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
-            _make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"),
-            _make_mock_asset("a3", "video3.mp4", "uploads/video3.mp4"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-
-        # 第一个成功，第二个失败，第三个成功
-        mock_download.side_effect = [True, False, True]
-
-        result = _download_library_assets(
-            Path("/tmp"),
-            asset_library_id="lib-1",
-            asset_ids=["a1", "a2", "a3"],
-            strict=False,
-        )
-
-        assert len(result) == 2
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_non_strict_all_fail_raises(self, mock_download, mock_session_factory):
-        """strict=False 但全部失败时仍抛 RuntimeError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-        mock_download.return_value = False
-
-        with pytest.raises(RuntimeError, match="全部下载失败"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-                asset_ids=["a1"],
-                strict=False,
-            )
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_strict_mode_raises_on_missing_file_url(self, mock_download, mock_session_factory):
-        """strict=True 时，素材缺少 file_url 立即抛异常."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", None),  # file_url 为空
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-
-        with pytest.raises(RuntimeError, match="素材缺少 file_url"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-                strict=True,
-            )
-
-        # download_asset 不应被调用
-        mock_download.assert_not_called()
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_default_is_strict(self, mock_download, mock_session_factory):
-        """默认 strict=True."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-        mock_download.return_value = False
-
-        # 不传 strict 参数，默认严格模式
-        with pytest.raises(RuntimeError, match="素材下载失败"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-            )
-
-
-# ── P3-2: 归属校验合并到同一 session ────────────────────────────────────────
-
-
-class TestDownloadLibraryAssetsOwnershipValidation:
-    """归属校验合并到 _download_library_assets 同一 session."""
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_ownership_mismatch_raises_value_error(self, mock_download, mock_session_factory):
-        """asset_ids 不属于指定素材库时抛 ValueError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        # asset 属于 lib-2，但请求的是 lib-1
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-2"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-
-        with pytest.raises(ValueError, match="素材不属于指定素材库"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-                asset_ids=["a1"],
-            )
-
-        # 不应调用 download_asset（校验在下载前）
-        mock_download.assert_not_called()
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_missing_asset_ids_raises_value_error(self, mock_download, mock_session_factory):
-        """指定的 asset_ids 不存在时抛 ValueError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        # DB 返回空（asset_ids 不存在，query 过滤后无结果）
-        mock_session = _setup_mock_session([])
-        mock_session_factory.return_value = mock_session
-
-        with pytest.raises(RuntimeError, match="未找到视频素材"):
-            _download_library_assets(
-                Path("/tmp"),
-                asset_library_id="lib-1",
-                asset_ids=["nonexistent-id"],
-            )
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_project_ownership_mismatch_raises(self, mock_download, mock_session_factory):
-        """项目级模式下归属不匹配抛 ValueError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset(
-                "a1",
-                "video1.mp4",
-                "uploads/video1.mp4",
-                asset_library_id="",
-                project_id="proj-2",
-            ),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-
-        with pytest.raises(ValueError, match="素材不属于指定项目"):
-            _download_library_assets(
-                Path("/tmp"),
-                project_id="proj-1",
-                asset_ids=["a1"],
-            )
-
-        mock_download.assert_not_called()
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_ownership_pass_then_download(self, mock_download, mock_session_factory):
-        """归属校验通过后正常下载."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-        mock_download.return_value = True
-
-        result = _download_library_assets(
-            Path("/tmp"),
-            asset_library_id="lib-1",
-            asset_ids=["a1"],
-        )
-
-        assert len(result) == 1
-        mock_download.assert_called_once()
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_single_session_used(self, mock_download, mock_session_factory):
-        """验证只创建了一个 DB session（P3-2 核心）."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        assets = [
-            _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"),
-        ]
-        mock_session = _setup_mock_session(assets)
-        mock_session_factory.return_value = mock_session
-        mock_download.return_value = True
-
-        _download_library_assets(
-            Path("/tmp"),
-            asset_library_id="lib-1",
-            asset_ids=["a1"],
-        )
-
-        # SessionLocal 只调用一次（合并前会调用两次：校验 + 下载）
-        assert mock_session_factory.call_count == 1

tests/unit/test_generation_preview.py

@@ -1206,7 +1206,10 @@ class TestPreviewRouteAutoInfersVideoRatio:
         # Verify the resolution passed to CreateGenerationTaskCommand
         call_args = MockUC.return_value.execute.call_args
         cmd = call_args[0][0]
-        assert cmd.resolution == "", f"Expected empty resolution, got {cmd.resolution}"
+        # video_ratio inferred from pip → 9:16 → resolution=1080x1920
+        assert cmd.resolution == "1080x1920", f"Expected 1080x1920, got {cmd.resolution}"
+        assert cmd.output_width == 1080, f"Expected output_width=1080, got {cmd.output_width}"
+        assert cmd.output_height == 1920, f"Expected output_height=1920, got {cmd.output_height}"
 
 
 # ═══════════════════════════════════════════════════════════════════════════════

tests/unit/test_generation_worker_fixes.py

@@ -1,228 +0,0 @@
-"""Tests for generation.py worker-side fixes: segment durations + preview resolution."""
-
-from __future__ import annotations
-
-import os
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-# Add worker app to path
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-
-def _patch_session_local(mock_session):
-    """Patch worker_app.db.SessionLocal robustly even when other tests
-    have pre-registered a MagicMock for worker_app.db in sys.modules.
-    Uses patch.dict to inject a clean module so that
-    'from worker_app.db import SessionLocal' resolves correctly."""
-    from types import ModuleType
-
-    _fresh_db = ModuleType("worker_app.db")
-    _fresh_db.SessionLocal = lambda *a, **kw: mock_session
-    return patch.dict(sys.modules, {"worker_app.db": _fresh_db})
-
-
-class TestLoadTemplateSegmentDurations:
-    """_load_template_segment_durations 单元测试 (covers lines 198-226)."""
-
-    def test_empty_template_id(self):
-        """空 template_id 直接返回空列表。"""
-        from worker_app.tasks.generation import _load_template_segment_durations
-
-        result = _load_template_segment_durations("")
-        assert result == []
-
-    def test_loads_durations_ordered(self):
-        """按 segment_order 排序返回 duration_max 列表。"""
-        from worker_app.tasks.generation import _load_template_segment_durations
-
-        mock_seg1 = MagicMock(duration_max=5.0)
-        mock_seg2 = MagicMock(duration_max=8.0)
-        mock_seg3 = MagicMock(duration_max=3.0)
-
-        mock_query = MagicMock()
-        mock_query.filter.return_value.order_by.return_value.all.return_value = [
-            mock_seg1,
-            mock_seg2,
-            mock_seg3,
-        ]
-        mock_session = MagicMock()
-        mock_session.query.return_value = mock_query
-
-        with _patch_session_local(mock_session):
-            result = _load_template_segment_durations("tpl_123")
-
-        assert result == [5.0, 8.0, 3.0]
-
-    def test_filters_zero_and_negative(self):
-        """duration_max <= 0 的 segment 被过滤。"""
-        from worker_app.tasks.generation import _load_template_segment_durations
-
-        mock_seg_valid = MagicMock(duration_max=5.0)
-        mock_seg_zero = MagicMock(duration_max=0.0)
-        mock_seg_none = MagicMock(duration_max=None)
-
-        mock_query = MagicMock()
-        mock_query.filter.return_value.order_by.return_value.all.return_value = [
-            mock_seg_valid,
-            mock_seg_zero,
-            mock_seg_none,
-        ]
-        mock_session = MagicMock()
-        mock_session.query.return_value = mock_query
-
-        with _patch_session_local(mock_session):
-            result = _load_template_segment_durations("tpl_456")
-
-        assert result == [5.0]
-
-    def test_db_error_returns_empty(self):
-        """数据库异常返回空列表，不抛出。"""
-        from types import ModuleType
-
-        from worker_app.tasks.generation import _load_template_segment_durations
-
-        _err_db = ModuleType("worker_app.db")
-
-        def _raise(*a, **kw):
-            raise Exception("DB down")
-
-        _err_db.SessionLocal = _raise
-        with patch.dict(sys.modules, {"worker_app.db": _err_db}):
-            result = _load_template_segment_durations("tpl_789")
-
-        assert result == []
-
-    def test_empty_segments_returns_empty(self):
-        """没有 segment 时返回空列表。"""
-        from worker_app.tasks.generation import _load_template_segment_durations
-
-        mock_query = MagicMock()
-        mock_query.filter.return_value.order_by.return_value.all.return_value = []
-        mock_session = MagicMock()
-        mock_session.query.return_value = mock_query
-
-        with _patch_session_local(mock_session):
-            result = _load_template_segment_durations("tpl_empty")
-
-        assert result == []
-
-
-class TestDurationCappingInBuildPlan:
-    """_build_plan_and_clips_from_task 时长约束测试 (covers lines 322-333)."""
-
-    def _make_temp_video(self, tmpdir: Path, name: str = "v.mp4") -> Path:
-        p = tmpdir / name
-        p.write_bytes(b"\x00" * 100)
-        return p
-
-    def test_clips_capped_by_segment_max(self):
-        """clip 时长超过 segment duration_max 时截断。"""
-        import tempfile
-
-        from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            tmpdir = Path(tmpdir)
-            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
-
-            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
-                with patch(
-                    "worker_app.tasks.generation._load_template_segment_durations",
-                    return_value=[5.0, 4.0, 3.0],
-                ):
-                    with patch(
-                        "worker_app.tasks.generation._load_template_clip_configs",
-                        return_value=[],
-                    ):
-                        _, clips, _ = _build_plan_and_clips_from_task(
-                            task_id="test_cap",
-                            downloaded_paths=paths,
-                            mode="one_take",
-                            template_id="tpl_test",
-                        )
-
-            assert clips[0].duration == 5.0
-            assert clips[1].duration == 4.0
-            assert clips[2].duration == 3.0
-
-    def test_clips_not_capped_when_under_max(self):
-        """clip 时长小于 segment duration_max 时不截断。"""
-        import tempfile
-
-        from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            tmpdir = Path(tmpdir)
-            paths = [self._make_temp_video(tmpdir)]
-
-            with patch("worker_app.tasks.generation.probe_duration", return_value=3.0):
-                with patch(
-                    "worker_app.tasks.generation._load_template_segment_durations",
-                    return_value=[5.0],
-                ):
-                    with patch(
-                        "worker_app.tasks.generation._load_template_clip_configs",
-                        return_value=[],
-                    ):
-                        _, clips, _ = _build_plan_and_clips_from_task(
-                            task_id="test_no_cap",
-                            downloaded_paths=paths,
-                            mode="one_take",
-                            template_id="tpl_test",
-                        )
-
-            assert clips[0].duration == 3.0
-
-    def test_no_capping_without_template(self):
-        """无 template_id 时不截断。"""
-        import tempfile
-
-        from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            tmpdir = Path(tmpdir)
-            paths = [self._make_temp_video(tmpdir)]
-
-            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
-                _, clips, _ = _build_plan_and_clips_from_task(
-                    task_id="test_no_tpl",
-                    downloaded_paths=paths,
-                    mode="one_take",
-                    template_id="",
-                )
-
-            assert clips[0].duration == 30.0
-
-    def test_partial_segments_only_caps_matching(self):
-        """segment 数量少于 clip 时，只截断有对应 segment 的 clip。"""
-        import tempfile
-
-        from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-        with tempfile.TemporaryDirectory() as tmpdir:
-            tmpdir = Path(tmpdir)
-            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
-
-            with patch("worker_app.tasks.generation.probe_duration", return_value=20.0):
-                with patch(
-                    "worker_app.tasks.generation._load_template_segment_durations",
-                    return_value=[5.0],  # only 1 segment for 3 clips
-                ):
-                    with patch(
-                        "worker_app.tasks.generation._load_template_clip_configs",
-                        return_value=[],
-                    ):
-                        _, clips, _ = _build_plan_and_clips_from_task(
-                            task_id="test_partial",
-                            downloaded_paths=paths,
-                            mode="one_take",
-                            template_id="tpl_test",
-                        )
-
-            assert clips[0].duration == 5.0  # capped
-            assert clips[1].duration == 20.0  # not capped (no matching segment)
-            assert clips[2].duration == 20.0  # not capped

tests/unit/test_oneclick_gen_p0_fixes.py

@@ -1,7 +1,6 @@
 """P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验.
 
 覆盖：
-  P0-1: _download_library_assets 双模式查询（asset_library_id / project_id）
   P0-2: OSS 上传失败抛异常 + URL 可访问性校验
   P0-3: FFmpeg 失败时完整 stderr 日志
   P1:   template_id 存在性校验 + asset_ids 归属校验
@@ -26,130 +25,6 @@ if str(_WORKER_ROOT) not in sys.path:
 # ── P0-1: _download_library_assets ────────────────────────────────────────────
 
 
-class TestDownloadLibraryAssets:
-    """P0-1: 素材下载双模式 + 错误处理."""
-
-    def _make_asset(self, id_: str, file_url: str, project_id: str = "p1", library_id: str = "lib1"):
-        mock = MagicMock()
-        mock.id = id_
-        mock.file_url = file_url
-        mock.name = f"asset_{id_}"
-        mock.project_id = project_id
-        mock.asset_library_id = library_id
-        return mock
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_asset_library_mode(self, mock_download, mock_session_factory):
-        """素材库模式：按 asset_library_id 查询."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        session = MagicMock()
-        mock_session_factory.return_value = session
-        query = MagicMock()
-        session.query.return_value = query
-        filter_result = MagicMock()
-        query.filter.return_value = filter_result
-        in_filter = MagicMock()
-        filter_result.filter.return_value = in_filter
-        assets = [self._make_asset("a1", "video/a1.mp4")]
-        in_filter.order_by.return_value.all.return_value = assets
-
-        mock_download.return_value = True
-
-        with patch("worker_app.tasks.generation.AssetModel", create=True):
-            result = _download_library_assets(
-                Path("/tmp/test"),
-                asset_library_id="lib1",
-            )
-
-        assert len(result) == 1
-        mock_download.assert_called_once()
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_project_mode(self, mock_download, mock_session_factory):
-        """项目级模式：asset_library_id 为空时按 project_id 查询."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        session = MagicMock()
-        mock_session_factory.return_value = session
-        query = MagicMock()
-        session.query.return_value = query
-        filter_result = MagicMock()
-        query.filter.return_value = filter_result
-        proj_filter = MagicMock()
-        filter_result.filter.return_value = proj_filter
-        assets = [self._make_asset("a1", "video/a1.mp4", project_id="proj1")]
-        proj_filter.order_by.return_value.all.return_value = assets
-
-        mock_download.return_value = True
-
-        result = _download_library_assets(
-            Path("/tmp/test"),
-            project_id="proj1",
-        )
-
-        assert len(result) == 1
-
-    def test_both_empty_raises(self):
-        """asset_library_id 和 project_id 都为空时抛 ValueError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        with pytest.raises(ValueError, match="至少需要提供一个"):
-            _download_library_assets(Path("/tmp/test"))
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    def test_no_assets_found_raises(self, mock_session_factory):
-        """查不到素材时抛 RuntimeError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        session = MagicMock()
-        mock_session_factory.return_value = session
-        query = MagicMock()
-        session.query.return_value = query
-        filter_result = MagicMock()
-        query.filter.return_value = filter_result
-        in_filter = MagicMock()
-        filter_result.filter.return_value = in_filter
-        in_filter.order_by.return_value.all.return_value = []
-
-        with pytest.raises(RuntimeError, match="未找到视频素材"):
-            _download_library_assets(
-                Path("/tmp/test"),
-                asset_library_id="lib1",
-            )
-
-    @patch("worker_app.tasks.generation.SessionLocal")
-    @patch("worker_app.tasks.generation.download_asset")
-    def test_all_asset_ids_fail_raises(self, mock_download, mock_session_factory):
-        """指定 asset_ids 但全部下载失败时抛 RuntimeError."""
-        from worker_app.tasks.generation import _download_library_assets
-
-        session = MagicMock()
-        mock_session_factory.return_value = session
-        query = MagicMock()
-        session.query.return_value = query
-        filter_result = MagicMock()
-        query.filter.return_value = filter_result
-        id_filter = MagicMock()
-        filter_result.filter.return_value = id_filter
-        assets = [self._make_asset("a1", "video/a1.mp4")]
-        id_filter.order_by.return_value.all.return_value = assets
-
-        mock_download.return_value = False  # 全部下载失败
-
-        with pytest.raises(RuntimeError, match="素材下载失败"):
-            _download_library_assets(
-                Path("/tmp/test"),
-                asset_library_id="lib1",
-                asset_ids=["a1"],
-            )
-
-
-# ── P0-2: OSS 上传 + URL 校验 ────────────────────────────────────────────────
-
-
 class TestOSSUploadAndVerify:
     """P0-2: OSS 上传失败抛异常 + URL 可访问性校验."""
 
@@ -290,318 +165,3 @@ class TestP1Validations:
 
 
 # ── P1: 一键生成 clip 级效果层映射 ───────────────────────────────────────────
-
-
-class TestTemplateClipEffectMapping:
-    """P1: 模板 clip 级效果层映射到一键生成素材 clips."""
-
-    def _make_virtual_clip(self, idx: int, clip_type: str = "main", config: dict | None = None):
-        from dataclasses import dataclass, field
-
-        _clip_type_val = clip_type
-
-        @dataclass
-        class FakeClip:
-            id: str = f"vc_{idx:03d}"
-            plan_id: str = "task_001"
-            clip_type: str = _clip_type_val
-            order: int = idx
-            asset_id: str = f"asset_{idx}"
-            duration: float = 5.0
-            transition_effect: str = "cut"
-            transition_duration: float = 0.0
-            playback_speed: float = 1.0
-            config: dict = field(default_factory=dict)
-
-        return FakeClip(config=config or {})
-
-    def _make_template_clip_config(self, clip_type: str = "main", transition: str = "cut", config: dict | None = None):
-        mock = MagicMock()
-        mock.clip_type = clip_type
-        mock.transition_effect = transition
-        mock.config = config or {}
-        mock.default_duration = 3.0
-        mock.text_template = ""
-        return mock
-
-    def test_transition_effect_mapped(self):
-        """转场效果正确映射到素材 clips."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(i) for i in range(3)]
-        clip_configs = [
-            self._make_template_clip_config("main", transition="fade"),
-            self._make_template_clip_config("main", transition="dissolve"),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        # 前两个按顺序映射，第三个用最后一个模板配置
-        assert clips[0].transition_effect == "fade"
-        assert clips[1].transition_effect == "dissolve"
-        assert clips[2].transition_effect == "dissolve"  # 复用最后一个
-
-    def test_color_grade_mapped(self):
-        """滤镜配置正确映射到 clip.config.color_grade."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(i) for i in range(2)]
-        clip_configs = [
-            self._make_template_clip_config(
-                "main", config={"color_grade": {"enabled": True, "filter": "vintage", "brightness": 0.1}}
-            ),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        assert clips[0].config["color_grade"]["filter"] == "vintage"
-        assert clips[0].config["color_grade"]["brightness"] == 0.1
-        # 第二个素材复用第一个模板配置
-        assert clips[1].config["color_grade"]["filter"] == "vintage"
-
-    def test_existing_config_preserved(self):
-        """已有 clip.config 内容（如 role）被保留."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(0, config={"role": "b_roll"})]
-        clip_configs = [
-            self._make_template_clip_config("main", config={"color_grade": {"enabled": True, "filter": "warm"}}),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "voice_over")
-
-        assert clips[0].config["role"] == "b_roll"  # 保留原有配置
-        assert clips[0].config["color_grade"]["filter"] == "warm"  # 新增滤镜配置
-
-    def test_empty_clip_configs_no_change(self):
-        """空模板配置时 clips 保持不变."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(i) for i in range(2)]
-        _apply_template_clip_effects(clips, [], "one_take")
-
-        assert clips[0].transition_effect == "cut"
-        assert clips[1].transition_effect == "cut"
-
-    def test_cut_transition_not_overwritten(self):
-        """模板转场为 cut 时不覆盖（保持默认）."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(0)]
-        clips[0].transition_effect = "fade"  # 已有非默认值
-        clip_configs = [
-            self._make_template_clip_config("main", transition="cut"),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        # 模板是 cut 时，保留原有值（避免无意义覆盖）
-        assert clips[0].transition_effect == "fade"
-
-    def test_transition_duration_mapped(self):
-        """转场时长（transition_duration）从模板 config 正确映射到 clip."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(i) for i in range(3)]
-        clip_configs = [
-            self._make_template_clip_config("main", transition="fade", config={"transition_duration": 0.8}),
-            self._make_template_clip_config("main", transition="dissolve", config={"transition_duration": 1.2}),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        # 前两个按顺序映射，第三个复用最后一个
-        assert clips[0].transition_effect == "fade"
-        assert clips[0].transition_duration == 0.8
-        assert clips[1].transition_effect == "dissolve"
-        assert clips[1].transition_duration == 1.2
-        assert clips[2].transition_effect == "dissolve"
-        assert clips[2].transition_duration == 1.2
-
-    def test_transition_duration_ignored_for_cut(self):
-        """模板转场为 cut 时，transition_duration 不生效（保持默认0）."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(0)]
-        clip_configs = [
-            self._make_template_clip_config("main", transition="cut", config={"transition_duration": 0.5}),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        # cut 转场不映射，transition_duration 也不应用
-        assert clips[0].transition_duration == 0.0
-
-    def test_transition_duration_invalid_value_skipped(self):
-        """transition_duration 为无效值时安全跳过."""
-        from worker_app.tasks.generation import _apply_template_clip_effects
-
-        clips = [self._make_virtual_clip(0)]
-        clip_configs = [
-            self._make_template_clip_config("main", transition="fade", config={"transition_duration": "abc"}),
-        ]
-
-        _apply_template_clip_effects(clips, clip_configs, "one_take")
-
-        assert clips[0].transition_effect == "fade"
-        assert clips[0].transition_duration == 0.0  # 无效值保持默认
-
-    def test_intro_outro_extracted(self):
-        """intro/outro 类型 clip_config 正确提取为 plan 级 intro_outro 配置."""
-        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
-
-        clip_configs = [
-            self._make_template_clip_config("intro", config={"intro_type": "text", "intro_text_color": "#ffffff"}),
-            self._make_template_clip_config("main"),
-            self._make_template_clip_config("outro", config={"outro_type": "follow", "outro_follow_text": "关注我们"}),
-        ]
-        # 设置 intro/outro 的 text_template
-        clip_configs[0].text_template = "精彩视频"
-        clip_configs[0].default_duration = 2.5
-
-        result = _extract_intro_outro_from_clip_configs(clip_configs)
-
-        assert result["has_intro"] is True
-        assert result["intro_type"] == "text"
-        assert result["intro_text"] == "精彩视频"
-        assert result["intro_duration"] == 2.5
-        assert result["intro_text_color"] == "#ffffff"
-        assert result["has_outro"] is True
-        assert result["outro_type"] == "follow"
-        assert result["outro_follow_text"] == "关注我们"
-
-    def test_intro_outro_empty_when_none(self):
-        """没有 intro/outro 时返回空 dict."""
-        from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs
-
-        clip_configs = [
-            self._make_template_clip_config("main"),
-            self._make_template_clip_config("main"),
-        ]
-
-        result = _extract_intro_outro_from_clip_configs(clip_configs)
-        assert result == {}
-
-
-class TestTemplatePlanConfigLoading:
-    """验证从模板加载 plan 级配置（BGM、字幕、标题）的逻辑。"""
-
-    def _mock_template(
-        self,
-        title_config=None,
-        subtitle_config=None,
-        bgm_config=None,
-        is_active=True,
-    ):
-        template = MagicMock()
-        template.id = "tmpl_001"
-        template.name = "Test Template"
-        template.is_active = is_active
-        template.title_config = title_config or {}
-        template.subtitle_config = subtitle_config or {}
-        template.bgm_config = bgm_config or {}
-        return template
-
-    def _mock_session(self, template):
-        session = MagicMock()
-
-        # EditTemplateModel 查询返回 None（走旧模板系统 fallback）
-        edit_query = MagicMock()
-        edit_filter = MagicMock()
-        edit_query.filter.return_value = edit_filter
-        edit_filter.first.return_value = None
-
-        # TemplateModel 查询返回 template（旧模板系统）
-        old_query = MagicMock()
-        old_filter = MagicMock()
-        old_query.filter.return_value = old_filter
-        old_filter.first.return_value = template
-
-        def _query_side_effect(model):
-            name = getattr(model, "__name__", "")
-            if "EditTemplate" in name:
-                return edit_query
-            return old_query
-
-        session.query.side_effect = _query_side_effect
-        return session
-
-    def test_load_template_config_assembles_three_fields(self):
-        """模板的三个独立字段正确组装成 plan.config 格式。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        title_cfg = {"enabled": True, "text": "我的标题", "font_size": 36}
-        subtitle_cfg = {"enabled": True, "auto_generated": True, "language": "zh"}
-        bgm_cfg = {"enabled": True, "preset_id": "bgm-001", "volume": 0.5}
-
-        template = self._mock_template(
-            title_config=title_cfg,
-            subtitle_config=subtitle_cfg,
-            bgm_config=bgm_cfg,
-        )
-        session = self._mock_session(template)
-
-        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
-            result = _load_template_plan_config("tmpl_001")
-
-        assert result["title"] == title_cfg
-        assert result["subtitle"] == subtitle_cfg
-        assert result["bgm"] == bgm_cfg
-
-    def test_load_template_config_empty_template_returns_empty(self):
-        """模板三个字段都为空时返回空 dict。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        template = self._mock_template()
-        session = self._mock_session(template)
-
-        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
-            result = _load_template_plan_config("tmpl_001")
-
-        assert result == {}
-
-    def test_load_template_config_only_bgm(self):
-        """只有 BGM 配置时只返回 bgm 字段。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        bgm_cfg = {"enabled": True, "audio_url": "https://example.com/bgm.mp3"}
-        template = self._mock_template(bgm_config=bgm_cfg)
-        session = self._mock_session(template)
-
-        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
-            result = _load_template_plan_config("tmpl_001")
-
-        assert "bgm" in result
-        assert result["bgm"] == bgm_cfg
-        assert "title" not in result
-        assert "subtitle" not in result
-
-    def test_load_template_config_empty_template_id(self):
-        """空 template_id 直接返回空 dict。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        result = _load_template_plan_config("")
-        assert result == {}
-
-        result = _load_template_plan_config(None)
-        assert result == {}
-
-    def test_load_template_config_not_found_returns_empty(self):
-        """模板不存在时返回空 dict，不抛异常。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        session = MagicMock()
-
-        def _query_side_effect(model):
-            q = MagicMock()
-            f = MagicMock()
-            q.filter.return_value = f
-            f.first.return_value = None
-            return q
-
-        session.query.side_effect = _query_side_effect
-
-        with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
-            result = _load_template_plan_config("tmpl_nonexist")
-
-        assert result == {}

tests/unit/test_preview_title_config.py

@@ -1,190 +0,0 @@
-"""Tests for preview title_config feature.
-
-验证预览 API 的 title_config 字段和 Worker 的标题配置解析逻辑。
-"""
-
-from __future__ import annotations
-
-import json
-
-import pytest
-
-
-class TestPreviewTitleConfigSchema:
-    """测试 CreatePreviewGenerationTaskRequest 的 title_config 字段."""
-
-    def test_title_config_default_empty(self):
-        """title_config 默认为空 dict."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-        )
-        assert req.title_config == {}
-
-    def test_title_config_with_text(self):
-        """传入标题文本."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-            title_config={"text": "测试标题"},
-        )
-        assert req.title_config["text"] == "测试标题"
-
-    def test_title_config_with_full_style(self):
-        """传入完整标题样式配置."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        config = {
-            "text": "我的视频标题",
-            "font": "思源黑体",
-            "font_size": 48,
-            "font_color": "#ffffff",
-            "position": "top",
-            "bold": True,
-            "stroke": 2,
-            "shadow": True,
-        }
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-            title_config=config,
-        )
-        assert req.title_config["text"] == "我的视频标题"
-        assert req.title_config["font_size"] == 48
-        assert req.title_config["position"] == "top"
-
-
-class TestCommandTitleConfig:
-    """测试 CreateGenerationTaskCommand 的 title_config 字段."""
-
-    def test_command_has_title_config(self):
-        """Command 包含 title_config 字段."""
-        from packages.application.generation_tasks import CreateGenerationTaskCommand
-
-        cmd = CreateGenerationTaskCommand(
-            title_config={"text": "hello", "font_size": 32},
-        )
-        assert cmd.title_config["text"] == "hello"
-        assert cmd.title_config["font_size"] == 32
-
-    def test_command_title_config_default_empty(self):
-        """Command 的 title_config 默认为空 dict."""
-        from packages.application.generation_tasks import CreateGenerationTaskCommand
-
-        cmd = CreateGenerationTaskCommand()
-        assert cmd.title_config == {}
-
-
-class TestWorkerTitleConfigParsing:
-    """测试 Worker 渲染时的标题配置解析逻辑."""
-
-    def test_json_format_parsing(self):
-        """JSON 格式的 custom_title 能正确解析."""
-        config = {"text": "测试标题", "font_size": 48, "font_color": "#ff0000"}
-        custom_title = json.dumps(config, ensure_ascii=False)
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is not None
-        assert parsed["text"] == "测试标题"
-        assert parsed["font_size"] == 48
-
-    def test_plain_text_fallback(self):
-        """纯文本的 custom_title 不触发 JSON 解析."""
-        custom_title = "简单的标题文字"
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is None
-
-    def test_invalid_json_fallback(self):
-        """无效 JSON 的 custom_title 降级为纯文本."""
-        custom_title = "{invalid json"
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is None
-
-    def test_json_without_text_skipped(self):
-        """JSON 格式但缺少 text 字段时，跳过标题注入."""
-        config = {"font_size": 48}
-        custom_title = json.dumps(config, ensure_ascii=False)
-
-        ct_stripped = custom_title.strip()
-        parsed = json.loads(ct_stripped)
-        title_text = (parsed.get("text") or "").strip()
-
-        assert title_text == ""
-
-    def test_style_key_mapping(self):
-        """前端字段名正确映射到 ASS 字段名."""
-        config = {
-            "text": "标题",
-            "font_size": 48,
-            "font_color": "#ffffff",
-            "font_preset": "思源黑体",
-        }
-
-        style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
-        title_cfg = {}
-        for key in style_keys:
-            if key in config and config[key] is not None:
-                mapped_key = {
-                    "font_size": "size",
-                    "font_color": "color",
-                    "font_preset": "font",
-                }.get(key, key)
-                title_cfg[mapped_key] = config[key]
-
-        assert title_cfg["size"] == 48
-        assert title_cfg["color"] == "#ffffff"
-        assert title_cfg["font"] == "思源黑体"
-
-
-class TestPreviewRouteTitleConfigPassing:
-    """测试预览路由正确序列化 title_config 到 custom_title."""
-
-    def test_title_config_serialization(self):
-        """title_config 序列化为 JSON 字符串."""
-        title_config = {
-            "text": "我的标题",
-            "font_size": 32,
-            "font_color": "#d4a843",
-        }
-        serialized = json.dumps(title_config, ensure_ascii=False)
-
-        parsed = json.loads(serialized)
-        assert parsed["text"] == "我的标题"
-        assert parsed["font_size"] == 32
-
-    def test_empty_title_config_produces_empty_string(self):
-        """空 title_config 时 custom_title 为空字符串."""
-        title_config = {}
-        title_text = (title_config.get("text") or "").strip()
-        custom_title_value = ""
-        if title_text:
-            custom_title_value = json.dumps(title_config, ensure_ascii=False)
-
-        assert custom_title_value == ""

tests/unit/test_preview_title_render.py

@@ -1,256 +0,0 @@
-"""预览视频标题渲染修复测试 — 覆盖3个断点。
-
-断点1: generate_video() → _render_video() 传递 custom_title
-断点2: _render_video() 解析 custom_title 并注入 virtual_plan.config["title"]
-断点3: generate_ass_from_timeline() ASR路径也渲染标题
-"""
-
-import json
-import time
-from pathlib import Path
-from unittest.mock import MagicMock, PropertyMock, patch
-
-import pytest
-
-# ── 断点2: _render_video 标题注入 ─────────────────────────────────────────────
-
-
-class TestRenderVideoCustomTitleInjection:
-    """验证 _render_video 正确接收并注入 custom_title 到 virtual_plan.config['title']。"""
-
-    @pytest.fixture
-    def sample_custom_title(self):
-        """模拟前端发送的 custom_title JSON（含 font_size/font_color）。"""
-        return json.dumps(
-            {
-                "text": "测试标题",
-                "font": "思源黑体",
-                "font_size": 30,
-                "font_color": "#FF0000",
-                "position": "top",
-                "bold": True,
-                "stroke": True,
-                "shadow": False,
-            },
-            ensure_ascii=False,
-        )
-
-    def _call_render_video_with_capture(self, custom_title, template_config=None, tmp_path=None):
-        """调用 _render_video，在 RenderAdapter 处中断并捕获 virtual_plan.config。"""
-        from worker_app.tasks.generation import _render_video
-
-        captured_config = {}
-
-        class FakePlan:
-            def __init__(self):
-                self.config = {}
-                self.id = "test-plan"
-
-        fake_plan = FakePlan()
-
-        def capture_and_raise(*args, **kwargs):
-            # 此时 title 已注入到 fake_plan.config
-            captured_config.update(fake_plan.config or {})
-            raise RuntimeError("STOP_HERE")
-
-        with (
-            patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build,
-            patch("worker_app.tasks.generation._load_template_plan_config", return_value=template_config),
-            patch("worker_app.tasks.generation.time.monotonic", side_effect=[0.0, 1.0]),
-            patch("video_processing.render_adapter.RenderAdapter") as mock_adapter_cls,
-        ):
-
-            mock_build.return_value = (fake_plan, [], {})
-            mock_adapter_cls.side_effect = capture_and_raise
-
-            with pytest.raises(RuntimeError, match="STOP_HERE"):
-                _render_video(
-                    task_id="test-task",
-                    downloaded_videos=[tmp_path / "v1.mp4"] if tmp_path else [Path("/tmp/v1.mp4")],
-                    voice_path=None,
-                    editing_mode=MagicMock(value="one_take"),
-                    project_id="proj-1",
-                    template_id="tpl-1",
-                    user_id="user-1",
-                    temp_path=tmp_path or Path("/tmp"),
-                    output_name="test_output",
-                    resolution="1280x720",
-                    bgm_config={},
-                    voice_ids=[],
-                    custom_title=custom_title,
-                )
-
-        return captured_config
-
-    def test_custom_title_injected_into_plan_config(self, sample_custom_title, tmp_path):
-        """custom_title JSON 应被解析并注入 virtual_plan.config['title']。"""
-        config = self._call_render_video_with_capture(sample_custom_title, tmp_path=tmp_path)
-
-        assert "title" in config
-        title_cfg = config["title"]
-        assert title_cfg["text"] == "测试标题"
-        # 字段归一化: font_size → size
-        assert title_cfg["size"] == 30
-        # 字段归一化: font_color → color
-        assert title_cfg["color"] == "#FF0000"
-
-    def test_custom_title_overrides_template_title(self, sample_custom_title, tmp_path):
-        """用户自定义标题应覆盖模板默认标题。"""
-        template_config = {"title": {"text": "模板默认标题", "size": 24}}
-        config = self._call_render_video_with_capture(
-            sample_custom_title, template_config=template_config, tmp_path=tmp_path
-        )
-
-        # 用户标题应覆盖模板标题
-        assert config["title"]["text"] == "测试标题"
-        assert config["title"]["size"] == 30
-
-    def test_empty_custom_title_no_injection(self, tmp_path):
-        """空 custom_title 不应注入 title 字段。"""
-        config = self._call_render_video_with_capture("", tmp_path=tmp_path)
-        assert "title" not in config
-
-    def test_malformed_custom_title_gracefully_ignored(self, tmp_path):
-        """非法 JSON 不应崩溃，应跳过注入。"""
-        config = self._call_render_video_with_capture("{invalid json!!!", tmp_path=tmp_path)
-        assert "title" not in config
-
-
-# ── 断点3: generate_ass_from_timeline ASR路径支持标题 ──────────────────────────
-
-
-class TestGenerateAssFromTimelineWithTitle:
-    """验证 generate_ass_from_timeline 在有标题时生成包含 TitleStyle 的 ASS。"""
-
-    def test_title_included_in_ass_output(self, tmp_path):
-        """有 title_text 时，ASS 输出应包含 TitleStyle 和标题事件。"""
-        from video_processing.subtitle_generator import generate_ass_from_timeline
-
-        from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
-
-        timeline = SubtitleTimeline(
-            segments=[
-                SubtitleSegment(start=0.0, end=2.0, text="你好世界"),
-            ]
-        )
-
-        ass_path = tmp_path / "test.ass"
-        result = generate_ass_from_timeline(
-            ass_path,
-            timeline,
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-            subtitle_config={"font": "思源黑体", "size": 24},
-            title_text="我的标题",
-            title_config={"font": "思源黑体", "size": 36, "color": "#FFFFFF", "position": "top"},
-        )
-
-        content = result.read_text(encoding="utf-8")
-        # 应包含 TitleStyle
-        assert "TitleStyle" in content
-        # 应包含标题文本
-        assert "我的标题" in content
-        # 也应包含 ASR 字幕
-        assert "你好世界" in content
-
-    def test_no_title_no_title_style(self, tmp_path):
-        """无标题时，ASS 输出不应包含 TitleStyle。"""
-        from video_processing.subtitle_generator import generate_ass_from_timeline
-
-        from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
-
-        timeline = SubtitleTimeline(
-            segments=[
-                SubtitleSegment(start=0.0, end=2.0, text="只有字幕"),
-            ]
-        )
-
-        ass_path = tmp_path / "test.ass"
-        result = generate_ass_from_timeline(
-            ass_path,
-            timeline,
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-            subtitle_config={},
-            title_text="",
-            title_config={},
-        )
-
-        content = result.read_text(encoding="utf-8")
-        assert "TitleStyle" not in content
-        assert "只有字幕" in content
-
-    def test_title_field_normalization_in_ass(self, tmp_path):
-        """前端字段名 font_size/font_color 应被正确归一化。"""
-        from video_processing.subtitle_generator import generate_ass_from_timeline
-
-        from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
-
-        timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
-
-        ass_path = tmp_path / "test.ass"
-        result = generate_ass_from_timeline(
-            ass_path,
-            timeline,
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-            subtitle_config={},
-            title_text="归一化测试",
-            title_config={
-                "font_size": 30,  # 前端字段名
-                "font_color": "#FF0000",  # 前端字段名
-                "position": "top",
-            },
-        )
-
-        content = result.read_text(encoding="utf-8")
-        assert "TitleStyle" in content
-        assert "归一化测试" in content
-
-    def test_title_boolean_stroke_shadow_compat(self, tmp_path):
-        """boolean stroke/shadow 应被兼容处理。"""
-        from video_processing.subtitle_generator import generate_ass_from_timeline
-
-        from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
-
-        timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
-
-        ass_path = tmp_path / "test.ass"
-        result = generate_ass_from_timeline(
-            ass_path,
-            timeline,
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-            subtitle_config={},
-            title_text="描边测试",
-            title_config={
-                "size": 36,
-                "stroke": True,  # boolean
-                "shadow": False,  # boolean
-            },
-        )
-
-        content = result.read_text(encoding="utf-8")
-        assert "TitleStyle" in content
-        assert "描边测试" in content
-
-
-# ── 断点1: _render_video 签名包含 custom_title ────────────────────────────────
-
-
-class TestRenderVideoSignature:
-    """验证 _render_video 函数签名正确。"""
-
-    def test_custom_title_parameter_exists(self):
-        """_render_video 应有 custom_title 参数，默认空字符串。"""
-        import inspect
-
-        from worker_app.tasks.generation import _render_video
-
-        sig = inspect.signature(_render_video)
-        assert "custom_title" in sig.parameters
-        assert sig.parameters["custom_title"].default == ""

tests/unit/test_preview_voice_ids_fallback.py

@@ -1,151 +0,0 @@
-"""Tests for voice_ids fallback in _download_all_assets.
-
-When voice_library_id is empty but voice_ids is non-empty, the Worker
-should fallback to voice_ids[0] as the audio asset_id.
-"""
-
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-
-class TestDownloadAllAssetsVoiceIdsFallback:
-    """_download_all_assets 配音下载 fallback 逻辑测试。"""
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_voice_library_id_takes_priority(self, mock_download_videos, mock_download_voice, tmp_path):
-        """voice_library_id 存在时优先使用，不 fallback 到 voice_ids。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-        mock_download_voice.return_value = True
-
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="voice-lib-123",
-            task_id="task-1",
-            voice_ids=["voice-ids-456"],
-        )
-
-        assert audio is not None
-        mock_download_voice.assert_called_once()
-        call_args = mock_download_voice.call_args
-        assert call_args[0][0] == "voice-lib-123"  # first positional arg
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_fallback_to_voice_ids_when_voice_library_id_empty(
-        self, mock_download_videos, mock_download_voice, tmp_path
-    ):
-        """voice_library_id 为空时 fallback 到 voice_ids[0]。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-        mock_download_voice.return_value = True
-
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="",  # 空字符串
-            task_id="task-2",
-            voice_ids=["voice-asset-789"],
-        )
-
-        assert audio is not None
-        mock_download_voice.assert_called_once()
-        call_args = mock_download_voice.call_args
-        assert call_args[0][0] == "voice-asset-789"
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_no_audio_when_both_empty(self, mock_download_videos, mock_download_voice, tmp_path):
-        """voice_library_id 和 voice_ids 都为空时，不下载音频。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="",
-            task_id="task-3",
-            voice_ids=[],
-        )
-
-        assert audio is None
-        mock_download_voice.assert_not_called()
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_no_audio_when_voice_ids_none(self, mock_download_videos, mock_download_voice, tmp_path):
-        """voice_ids 为 None 时，不触发 fallback。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="",
-            task_id="task-4",
-            voice_ids=None,
-        )
-
-        assert audio is None
-        mock_download_voice.assert_not_called()
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_voice_library_id_empty_string_fallback(self, mock_download_videos, mock_download_voice, tmp_path):
-        """voice_library_id 为空字符串且 voice_ids 有多个元素时，取第一个。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-        mock_download_voice.return_value = True
-
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="",
-            task_id="task-5",
-            voice_ids=["first-id", "second-id", "third-id"],
-        )
-
-        assert audio is not None
-        call_args = mock_download_voice.call_args
-        assert call_args[0][0] == "first-id"
-
-    @patch("worker_app.tasks.generation._download_voice_asset")
-    @patch("worker_app.tasks.generation._download_library_assets")
-    def test_backward_compat_no_voice_ids_param(self, mock_download_videos, mock_download_voice, tmp_path):
-        """不传 voice_ids 参数时，行为与之前一致（向后兼容）。"""
-        from worker_app.tasks.generation import _download_all_assets
-
-        mock_download_videos.return_value = [tmp_path / "v1.mp4"]
-        mock_download_voice.return_value = True
-
-        # 不传 voice_ids
-        videos, audio = _download_all_assets(
-            temp_path=tmp_path,
-            asset_library_id="lib-1",
-            project_id="proj-1",
-            task_asset_ids=["a1"],
-            voice_library_id="voice-lib-999",
-            task_id="task-6",
-        )
-
-        assert audio is not None
-        mock_download_voice.assert_called_once_with("voice-lib-999", tmp_path / "voice.mp3")

tests/unit/test_race_condition_fallback_plan.py

@@ -0,0 +1,98 @@
+"""
+回归测试：验证 generation_tasks.py 中兜底关联 edit plan 在 enqueue 之前执行。
+
+根因（PR #1481 后续修复）：兜底关联逻辑原来在 safe_enqueue_generation_task 之后执行，
+导致 worker 在 enqueue 后立即读取 task 时，source_edit_plan_id 仍为空（竞态条件）。
+"""
+
+import ast
+import textwrap
+
+
+def _get_function_source(filepath, func_name):
+    """提取函数源码"""
+    with open(filepath) as f:
+        source = f.read()
+    tree = ast.parse(source)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name:
+            lines = source.splitlines()
+            start = node.lineno - 1
+            end = node.end_lineno
+            return textwrap.dedent("\n".join(lines[start:end]))
+    return None
+
+
+def _find_try_block_source(func_source):
+    """在函数源码中找到包含 safe_enqueue_generation_task 的 try 块"""
+    tree = ast.parse(textwrap.dedent(func_source))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Try):
+            # 检查 try 块中是否包含 safe_enqueue_generation_task
+            block_lines = func_source.splitlines()
+            block_text = "\n".join(block_lines[node.lineno - 1 : node.end_lineno])
+            if "safe_enqueue_generation_task" in block_text:
+                return block_text
+    return None
+
+
+def test_fallback_before_enqueue():
+    """验证兜底关联 edit plan 的代码在 safe_enqueue_generation_task 调用之前"""
+    filepath = "apps/api/app/api/routes/generation_tasks.py"
+    func_source = _get_function_source(filepath, "create_generation_task")
+    assert func_source is not None, "create_generation_task function not found"
+
+    try_block = _find_try_block_source(func_source)
+    assert try_block is not None, "try block with safe_enqueue_generation_task not found"
+
+    # 定位关键标记在 try 块中的行号
+    lines = try_block.splitlines()
+
+    fallback_line = None
+    enqueue_line = None
+    writeback_line = None
+
+    for i, line in enumerate(lines):
+        if "not task.source_edit_plan_id and request.template_id" in line and fallback_line is None:
+            fallback_line = i
+        if "safe_enqueue_generation_task(" in line and enqueue_line is None:
+            enqueue_line = i
+        if "_writeback_edit_plan_config(" in line and "def " not in line and writeback_line is None:
+            writeback_line = i
+
+    assert fallback_line is not None, "兜底关联逻辑 not found in try block"
+    assert enqueue_line is not None, "safe_enqueue_generation_task call not found in try block"
+    assert writeback_line is not None, "_writeback_edit_plan_config call not found in try block"
+
+    # 核心断言：兜底关联和回写都在 enqueue 之前
+    assert fallback_line < enqueue_line, f"兜底关联（行{fallback_line}）应在 enqueue（行{enqueue_line}）之前"
+    assert writeback_line < enqueue_line, f"回写 config（行{writeback_line}）应在 enqueue（行{enqueue_line}）之前"
+
+
+def test_fallback_sets_source_edit_plan_id():
+    """验证兜底关联逻辑会设置 task.source_edit_plan_id"""
+    filepath = "apps/api/app/api/routes/generation_tasks.py"
+    func_source = _get_function_source(filepath, "create_generation_task")
+    assert func_source is not None
+
+    try_block = _find_try_block_source(func_source)
+    assert try_block is not None
+
+    # 验证兜底逻辑包含赋值语句
+    assert "task.source_edit_plan_id = _plan_model.id" in try_block, "兜底关联逻辑应设置 task.source_edit_plan_id"
+    assert "generation_task_repository.update(task)" in try_block, "兜底关联后应持久化 task 到 DB"
+
+
+def test_writeback_uses_effective_plan_id():
+    """验证回写 config 使用的是 effective_plan_id（包含兜底结果），而非仅 request.source_edit_plan_id"""
+    filepath = "apps/api/app/api/routes/generation_tasks.py"
+    func_source = _get_function_source(filepath, "create_generation_task")
+    assert func_source is not None
+
+    try_block = _find_try_block_source(func_source)
+    assert try_block is not None
+
+    # 验证使用了 _effective_plan_id 或 task.source_edit_plan_id，而非仅 request.source_edit_plan_id
+    # 修复前用的是 request.source_edit_plan_id，修复后应该用 task.source_edit_plan_id
+    uses_effective = "_effective_plan_id" in try_block or "task.source_edit_plan_id" in try_block
+    assert uses_effective, "回写 config 应使用包含兜底结果的有效 plan_id"

tests/unit/test_render_subtitles_pure.py

@@ -280,7 +280,7 @@ class TestGenerateAssSubtitles:
         content = output.read_text(encoding="utf-8")
         assert "有字幕" in content
 
-    def test_custom_title_color(self, tmp_path):
+    def test_title_color_overlay(self, tmp_path):
         """自定义标题颜色."""
         output = tmp_path / "color.ass"
         result = generate_ass_subtitles(

tests/unit/test_temp_dir_cleanup_race.py

@@ -0,0 +1,56 @@
+"""回归测试：渲染产物临时目录不在 render_plan 中提前清理。
+
+根因：render_plan 的 finally 块在返回前清理了临时目录，
+但 generation.py 还需要访问其中的文件进行 OSS 上传。
+修复：将清理责任交给调用方（generation.py），render_plan 只在失败时清理。
+"""
+
+import ast
+
+
+def test_render_adapter_result_has_temp_dir_field():
+    """RenderAdapterResult 包含 temp_dir 字段"""
+    with open("apps/worker/video_processing/render_adapter.py") as f:
+        source = f.read()
+    tree = ast.parse(source)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ClassDef) and node.name == "RenderAdapterResult":
+            for item in node.body:
+                if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
+                    if item.target.id == "temp_dir":
+                        return
+    raise AssertionError("RenderAdapterResult 缺少 temp_dir 字段")
+
+
+def test_render_plan_does_not_cleanup_on_success():
+    """render_plan 成功时不在 finally 中清理临时目录（通过将 temp_dir 置为 None）"""
+    with open("apps/worker/video_processing/render_adapter.py") as f:
+        source = f.read()
+
+    # 成功路径必须将 temp_dir 置为 None，以阻止 finally 清理
+    assert "temp_dir = None" in source, "render_plan 成功时应将 temp_dir 置为 None 以阻止 finally 清理"
+
+
+def test_render_plan_passes_temp_dir_to_result():
+    """render_plan 将 temp_dir 传递给返回结果"""
+    with open("apps/worker/video_processing/render_adapter.py") as f:
+        source = f.read()
+
+    assert "result.temp_dir = temp_dir" in source, "render_plan 应将 temp_dir 设置到 result 上"
+
+
+def test_generation_cleans_up_temp_dir():
+    """generation.py 在上传完成后清理临时目录"""
+    with open("apps/worker/worker_app/tasks/generation.py") as f:
+        source = f.read()
+
+    # 验证 _render_from_edit_plan 返回 temp_dir
+    assert "render_temp_dir" in source, "generation.py 应接收 render_temp_dir"
+
+    # 验证有清理逻辑（shutil.rmtree(render_temp_dir...）
+    assert "rmtree(render_temp_dir" in source, "generation.py 应清理 render_temp_dir"
+
+    # 验证清理发生在上传之后（通过查找顺序）
+    upload_pos = source.find("_upload_and_record")
+    cleanup_pos = source.find("rmtree(render_temp_dir")
+    assert upload_pos > 0 and cleanup_pos > upload_pos, "清理临时目录应在 _upload_and_record 之后执行"

tests/unit/test_templates_editor_api.py

@@ -1,12 +1,11 @@
 """
 templates_editor.py 模板编辑器 API 端点单元测试
 
-覆盖核心端点（25个测试用例）：
+覆盖核心端点（23个测试用例）：
 - 草稿：GET/PUT/发布
 - 片段：list/create/get/update/delete/split/merge
 - BGM：GET/PUT
 - 时间线：GET
-- 生成状态查询
 - 预设：BGM预设
 """
 
@@ -395,30 +394,6 @@ class TestTimelineRoute:
         mock_plan_svc.list_clips.assert_called()
 
 
-# ---------------------------------------------------------------------------
-# 生成端点测试
-# ---------------------------------------------------------------------------
-
-
-class TestGenerationRoutes:
-    """生成端点测试"""
-
-    def test_generation_status_success(self, client):
-        c, _, mock_plan_svc = client
-        resp = c.get(BASE + "/generation-status")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "generation_task_id" in data
-        assert "clips" in data
-
-    def test_generations_list_success(self, client):
-        c, _, mock_plan_svc = client
-        resp = c.get(BASE + "/generations")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert "items" in data or "tasks" in data or isinstance(data, dict)
-
-
 # ---------------------------------------------------------------------------
 # 字幕端点测试
 # ---------------------------------------------------------------------------

tests/unit/test_three_bugs_fix.py

@@ -0,0 +1,401 @@
+"""Regression tests for 3 bug fixes: flush, append_log, plan_id fallback."""
+
+from __future__ import annotations
+
+import json
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+
+
+# ── Bug 1: db.flush() before pending query ─────────────────────────────────
+
+
+class TestReplaceAllClipsFlush:
+    """replace_all_clips_transactional must flush before querying pending clips."""
+
+    def _make_svc_and_db(self, pending_results):
+        """Helper: create service + db mock. pending_results = list returned by pending query."""
+        from app.services.edit_plan_service import EditPlanService
+
+        db = MagicMock()
+        # Delete query
+        delete_query = MagicMock()
+        delete_query.filter.return_value.delete.return_value = 0
+        # Pending query: single .filter() with multiple conditions
+        ready_query = MagicMock()
+        ready_query.filter.return_value.all.return_value = pending_results
+        db.query.side_effect = [delete_query, ready_query]
+        return db, EditPlanService
+
+    def _setup_clip_mocks(self, mock_clip_cls, mock_model_cls, asset_id="asset-1"):
+        mock_entity = MagicMock()
+        mock_entity.id = "clip-1"
+        mock_entity.plan_id = "plan-1"
+        mock_entity.clip_type = "main"
+        mock_entity.order = 0
+        mock_entity.asset_id = asset_id
+        mock_entity.text_content = ""
+        mock_entity.start_time = 0.0
+        mock_entity.duration = 3.0
+        mock_entity.transition_effect = "cut"
+        mock_entity.transition_duration = 0.0
+        mock_entity.playback_speed = 1.0
+        mock_entity.status.value = "pending"
+        mock_entity.config = {}
+        mock_clip_cls.create.return_value = mock_entity
+        mock_model_cls.return_value = MagicMock()
+        return mock_entity
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_called_between_add_and_query(self, mock_clip_cls, mock_model_cls):
+        """db.flush() must be called after db.add() and before the pending query."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+        db, SvcClass = self._make_svc_and_db([])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        db.flush.assert_called_once()
+        # Verify ordering: add → flush → query → commit
+        method_names = [c[0] for c in db.method_calls]
+        add_idx = method_names.index("add")
+        flush_idx = method_names.index("flush")
+        commit_idx = method_names.index("commit")
+        assert add_idx < flush_idx < commit_idx
+
+    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
+    @patch("app.services.edit_plan_service.EditPlanClip")
+    def test_flush_marks_new_clips_ready(self, mock_clip_cls, mock_model_cls):
+        """After flush, new clips with asset_id are found and marked ready."""
+        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
+
+        # Use a plain object so we can verify attribute mutation
+        class FakeClip:
+            status = "pending"
+
+        pending_clip = FakeClip()
+        db, SvcClass = self._make_svc_and_db([pending_clip])
+
+        clip_repo = MagicMock()
+        clip_repo.session = db
+        svc = SvcClass.__new__(SvcClass)
+        svc._clip_repo = clip_repo
+
+        svc.replace_all_clips_transactional(
+            "plan-1",
+            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
+        )
+
+        assert pending_clip.status == "ready"
+
+
+# ── Bug 2: append_log no TypeError ─────────────────────────────────────────
+
+
+class TestAppendLogNoConflict:
+    """append_log must not receive duplicate 'stage' parameter."""
+
+    def _make_task(self):
+        from packages.domain.generation_task import GenerationTask
+
+        return GenerationTask.create(
+            project_id="proj-1",
+            asset_library_id="lib-1",
+            strategy_id="one_take",
+            template_id="tpl-1",
+            asset_ids=["a1"],
+            created_by_user_id="user-1",
+        )
+
+    def test_append_log_with_stage_as_first_positional(self):
+        """append_log(stage, message, ...) works correctly."""
+        task = self._make_task()
+        task.append_log("render", "some error", level="ERROR", error_type="RuntimeError")
+
+        entries = json.loads(task.logs)
+        assert len(entries) == 1
+        assert entries[0]["stage"] == "render"
+        assert entries[0]["message"] == "some error"
+        assert entries[0]["level"] == "ERROR"
+        assert entries[0]["error_type"] == "RuntimeError"
+
+    def test_duplicate_stage_raises_type_error(self):
+        """Sanity check: passing stage both positionally and as kwarg raises TypeError."""
+        task = self._make_task()
+        with pytest.raises(TypeError):
+            task.append_log(
+                "任务失败",  # positional → stage
+                "some error",
+                level="ERROR",
+                stage="render",  # duplicate → TypeError
+            )
+
+
+# ── Bug 3: plan_id fallback in create_generation_task ──────────────────────
+
+
+class TestPlanIdFallback:
+    """Formal generation API should fallback to find plan by template_id + user_id."""
+
+    def test_fallback_code_present_in_source(self):
+        """Verify the fallback logic is present in the generation_tasks module."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "EditPlanModel" in source
+        assert "兜底关联编辑计划" in source
+        assert "自动关联编辑计划" in source
+
+    def test_fallback_only_runs_when_source_edit_plan_id_empty(self):
+        """Verify the condition checks for empty source_edit_plan_id."""
+        import inspect
+
+        from apps.api.app.api.routes import generation_tasks
+
+        source = inspect.getsource(generation_tasks.create_generation_task)
+        assert "not task.source_edit_plan_id and request.template_id" in source
+
+    def test_list_by_template_method_exists(self):
+        """Verify SQLAlchemyEditPlanRepository.list_by_template is callable."""
+        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+            SQLAlchemyEditPlanRepository,
+        )
+
+        mock_db = MagicMock()
+        mock_session = MagicMock()
+        mock_db.query.return_value = mock_session
+        mock_session.filter.return_value = mock_session
+        mock_session.order_by.return_value = mock_session
+        mock_session.offset.return_value = mock_session
+        mock_session.limit.return_value.all.return_value = []
+
+        repo = SQLAlchemyEditPlanRepository(mock_db)
+        result = repo.list_by_template("tpl-1", limit=20)
+        assert result == []
+
+
+class TestPlanIdFallbackExecution:
+    """Test that the fallback logic actually executes when source_edit_plan_id is empty."""
+
+    @staticmethod
+    def _make_mock_task(source_edit_plan_id=""):
+        t = MagicMock()
+        t.id = "task-1"
+        t.project_id = "proj-1"
+        t.asset_library_id = ""
+        t.strategy_id = "one_take"
+        t.voice_library_id = ""
+        t.template_id = "tpl-1"
+        t.asset_ids = []
+        t.title_ids = []
+        t.voice_ids = []
+        t.source_edit_plan_id = source_edit_plan_id
+        t.asset_select_mode = "manual"
+        t.batch_id = ""
+        t.video_title = ""
+        t.resolution = ""
+        t.bgm_config = None
+        t.is_preview = False
+        t.source_task_id = ""
+        t.output_width = 1280
+        t.output_height = 720
+        t.cover_url = ""
+        t.title_config = {}
+        t.logs = "[]"
+        t.status = "pending"
+        t.progress = 0.0
+        t.error_message = ""
+        t.error_info = None
+        t.created_at = "2026-01-01T00:00:00Z"
+        t.updated_at = "2026-01-01T00:00:00Z"
+        t.started_at = None
+        t.completed_at = None
+        t.created_by_user_id = "user-1"
+        t.auto_retry_enabled = False
+        t.auto_retry_max = 0
+        t.auto_retry_count = 0
+        t.result_count = 0
+        return t
+
+    @staticmethod
+    def _make_request(source_edit_plan_id="", template_id="tpl-1"):
+        req = MagicMock()
+        req.template_id = template_id
+        req.source_edit_plan_id = source_edit_plan_id
+        req.asset_ids = []
+        req.asset_select_mode = "manual"
+        req.asset_select_count = 0
+        req.voice_library_id = ""
+        req.title_ids = []
+        req.voice_ids = []
+        req.strategy_id = "one_take"
+        req.count = 1
+        req.video_title = ""
+        req.resolution = ""
+        req.bgm_config = None
+        req.auto_retry_enabled = False
+        req.auto_retry_max = 0
+        req.is_preview = False
+        req.source_task_id = ""
+        req.output_width = 0
+        req.output_height = 0
+        req.cover_url = ""
+        req.title_config = {}
+        req.project_id = None
+        req.asset_library_id = None
+        return req
+
+    def _run_create_task(self, mock_task, mock_request, mock_db, mock_gen_repo):
+        """Helper to run create_generation_task with common mocks."""
+        from app.schemas.generation_task import GenerationTaskResponse
+
+        with (
+            patch(
+                "apps.api.app.api.routes.generation_tasks._resolve_project_and_library", return_value=("proj-1", None)
+            ),
+            patch("apps.api.app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as mock_uc_cls,
+            patch("apps.api.app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True),
+            patch("apps.api.app.api.routes.generation_tasks._to_generation_task_response") as mock_resp_fn,
+        ):
+            mock_uc_cls.return_value.execute.return_value = mock_task
+            mock_resp_fn.return_value = GenerationTaskResponse(
+                id="task-1",
+                project_id="proj-1",
+                asset_library_id="",
+                strategy_id="one_take",
+                voice_library_id="",
+                template_id="tpl-1",
+                asset_ids=[],
+                title_ids=[],
+                voice_ids=[],
+                source_edit_plan_id="",
+                asset_select_mode="",
+                batch_id="",
+                video_title="",
+                resolution="",
+                bgm_config={},
+                is_preview=False,
+                source_task_id="",
+                output_width=1280,
+                output_height=720,
+                cover_url="",
+                title_config={},
+                logs="[]",
+                status="pending",
+                progress=0.0,
+                error_message="",
+                created_at="2026-01-01T00:00:00Z",
+                updated_at="2026-01-01T00:00:00Z",
+                started_at=None,
+                completed_at=None,
+                created_by_user_id="user-1",
+                auto_retry_enabled=False,
+                auto_retry_max=0,
+                auto_retry_count=0,
+                result_count=0,
+            )
+
+            from apps.api.app.api.routes.generation_tasks import create_generation_task
+
+            return create_generation_task(
+                request=mock_request,
+                authenticated_user=MagicMock(user=MagicMock(id="user-1")),
+                generation_task_repository=mock_gen_repo,
+                project_repository=MagicMock(),
+                asset_library_repository=MagicMock(),
+                asset_repository=MagicMock(),
+                db=mock_db,
+            )
+
+    def test_fallback_sets_plan_id_when_empty(self):
+        """When source_edit_plan_id is empty, fallback finds plan via DB query."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        # Mock DB query chain: db.query(EditPlanModel).filter(...).order_by(...).first()
+        mock_plan_model = MagicMock()
+        mock_plan_model.id = "plan-found-123"
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = mock_plan_model
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-found-123"
+        mock_gen_repo.update.assert_called_once_with(mock_task)
+
+    def test_no_fallback_when_plan_id_already_set(self):
+        """When source_edit_plan_id is already set, fallback should NOT run."""
+        mock_task = self._make_mock_task(source_edit_plan_id="plan-already-set")
+        mock_request = self._make_request(source_edit_plan_id="plan-already-set", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == "plan-already-set"
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_no_match_leaves_plan_id_empty(self):
+        """When no plan matches, source_edit_plan_id stays empty."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        query_chain = MagicMock()
+        query_chain.filter.return_value = query_chain
+        query_chain.order_by.return_value = query_chain
+        query_chain.first.return_value = None  # no matching plan
+        mock_db.query.return_value = query_chain
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()
+
+    def test_fallback_handles_exception_gracefully(self):
+        """When DB query fails, the fallback should not break the main flow."""
+        mock_task = self._make_mock_task(source_edit_plan_id="")
+        mock_request = self._make_request(source_edit_plan_id="", template_id="tpl-1")
+
+        mock_db = MagicMock()
+        mock_db.query.side_effect = Exception("DB connection error")
+
+        mock_gen_repo = MagicMock()
+        mock_gen_repo.count_pending_by_user.return_value = 0
+        mock_gen_repo.count_pending_total.return_value = 0
+
+        self._run_create_task(mock_task, mock_request, mock_db, mock_gen_repo)
+
+        # Task should still be created (fallback error doesn't break main flow)
+        assert mock_task.source_edit_plan_id == ""
+        mock_gen_repo.update.assert_not_called()

tests/unit/test_tts_preview.py

@@ -81,6 +81,12 @@ class TestTTSPreviewEndpoint:
         )
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
@@ -120,6 +126,12 @@ class TestTTSPreviewEndpoint:
         )
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
@@ -158,6 +170,12 @@ class TestTTSPreviewEndpoint:
         mock_service.synthesize_speech.side_effect = CosyVoiceError("API timeout")
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
@@ -186,6 +204,12 @@ class TestTTSPreviewEndpoint:
         mock_service.synthesize_speech.side_effect = ValueError("text 不能为空")
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
@@ -213,6 +237,12 @@ class TestTTSPreviewEndpoint:
         mock_service = MagicMock()
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
@@ -239,9 +269,173 @@ class TestTTSPreviewEndpoint:
         mock_service = MagicMock()
         app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
 
+        from app.dependencies import get_voice_clone_profile_repository
+
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None  # no profile found = preset voice
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
         client = self._make_client(app)
         resp = client.post(
             "/tts/preview",
             json={"text": "hello", "voice_id": ""},
         )
         assert resp.status_code == 422
+
+    def test_preview_clone_voice_resolves_to_cosyvoice_id(self):
+        """Clone voice UUID is resolved to CosyVoice voice_id."""
+        from dataclasses import dataclass
+
+        from fastapi import FastAPI
+
+        app = FastAPI()
+        from app.api.routes.tts import router
+
+        app.include_router(router, prefix="/tts")
+
+        from app.auth import get_current_user
+        from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+
+        fake_user = MagicMock()
+        fake_user.user.id = "user-1"
+        app.dependency_overrides[get_current_user] = lambda: fake_user
+
+        mock_service = MagicMock()
+        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
+            audio_url="https://x.com/cloned.mp3",
+            duration=1.8,
+        )
+        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
+
+        # Mock voice clone profile with voice_id
+        mock_profile = MagicMock()
+        mock_profile.user_id = "user-1"
+        mock_profile.voice_id = "cosyvoice_actual_voice_123"
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = mock_profile
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
+        client = self._make_client(app)
+        # Frontend sends the profile UUID as voice_id
+        resp = client.post(
+            "/tts/preview",
+            json={"text": "克隆音色测试", "voice_id": "abc123-uuid-of-profile"},
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["audio_url"] == "https://x.com/cloned.mp3"
+
+        # Verify CosyVoice was called with the resolved voice_id, not the UUID
+        mock_service.synthesize_speech.assert_called_once_with(
+            text="克隆音色测试",
+            voice_id="cosyvoice_actual_voice_123",
+            speed=1.0,
+        )
+        # Verify repo was queried with the UUID
+        mock_clone_repo.get.assert_called_once_with("abc123-uuid-of-profile")
+
+    def test_preview_clone_voice_incomplete_returns_400(self):
+        """Clone profile with empty voice_id returns 400."""
+        from fastapi import FastAPI
+
+        app = FastAPI()
+        from app.api.routes.tts import router
+
+        app.include_router(router, prefix="/tts")
+
+        from app.auth import get_current_user
+        from app.dependencies import get_voice_clone_profile_repository
+
+        fake_user = MagicMock()
+        fake_user.user.id = "user-1"
+        app.dependency_overrides[get_current_user] = lambda: fake_user
+
+        # Mock voice clone profile with empty voice_id (clone not finished)
+        mock_profile = MagicMock()
+        mock_profile.user_id = "user-1"
+        mock_profile.voice_id = ""
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = mock_profile
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
+        client = self._make_client(app)
+        resp = client.post(
+            "/tts/preview",
+            json={"text": "测试未完成克隆", "voice_id": "abc123-uuid"},
+        )
+        assert resp.status_code == 400
+        assert "音色克隆尚未完成" in resp.json()["detail"]
+
+    def test_preview_preset_voice_passthrough(self):
+        """Preset voice ID (not a profile UUID) passes through unchanged."""
+        from fastapi import FastAPI
+
+        app = FastAPI()
+        from app.api.routes.tts import router
+
+        app.include_router(router, prefix="/tts")
+
+        from app.auth import get_current_user
+        from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+
+        fake_user = MagicMock()
+        fake_user.user.id = "user-1"
+        app.dependency_overrides[get_current_user] = lambda: fake_user
+
+        mock_service = MagicMock()
+        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
+            audio_url="https://x.com/preset.mp3",
+            duration=2.0,
+        )
+        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
+
+        # Mock repo returns None (preset voice, not a clone profile)
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = None
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
+        client = self._make_client(app)
+        resp = client.post(
+            "/tts/preview",
+            json={"text": "预设音色测试", "voice_id": "longxiaoxia_v3"},
+        )
+        assert resp.status_code == 200
+
+        # Verify CosyVoice was called with the original preset voice_id
+        mock_service.synthesize_speech.assert_called_once_with(
+            text="预设音色测试",
+            voice_id="longxiaoxia_v3",
+            speed=1.0,
+        )
+
+    def test_preview_clone_voice_wrong_user_returns_403(self):
+        """Accessing another user's clone profile returns 403."""
+        from fastapi import FastAPI
+
+        app = FastAPI()
+        from app.api.routes.tts import router
+
+        app.include_router(router, prefix="/tts")
+
+        from app.auth import get_current_user
+        from app.dependencies import get_voice_clone_profile_repository
+
+        fake_user = MagicMock()
+        fake_user.user.id = "user-1"
+        app.dependency_overrides[get_current_user] = lambda: fake_user
+
+        # Mock profile belonging to a different user
+        mock_profile = MagicMock()
+        mock_profile.user_id = "user-2"
+        mock_profile.voice_id = "cosyvoice_voice_xyz"
+        mock_clone_repo = MagicMock()
+        mock_clone_repo.get.return_value = mock_profile
+        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
+
+        client = self._make_client(app)
+        resp = client.post(
+            "/tts/preview",
+            json={"text": "越权测试", "voice_id": "other-user-profile-uuid"},
+        )
+        assert resp.status_code == 403
+        assert "无权访问该音色" in resp.json()["detail"]

tests/unit/test_worker_cover_meta_and_status.py

@@ -46,13 +46,10 @@ class TestAssetAnalysesDoesNotOverwriteStatus:
             "Use an independent session to update only extra_meta."
         )
 
-    def test_asset_analyses_uses_independent_session(self):
-        """asset_analyses 持久化必须用独立 session 查询最新模型再提交。"""
+    def test_no_stale_repo_update(self):
+        """旧路径的 _repo.update(gen_task) 模式已随 DEPRECATED 代码一起删除。"""
         source = _read_source()
-        assert "_meta_session" in source
-        assert "GenerationTaskModel" in source
-        # 必须只更新 extra_meta 字段
-        assert 'existing["asset_analyses"]' in source
+        assert "_repo.update(gen_task)" not in source
 
 
 class TestGenerationTaskModelOrmAttribute:

tests/unit/test_worker_generate_video_task_binding.py

@@ -6,28 +6,63 @@ _sync_task_config_to_plan directly under the @celery_app.task decorator,
 so Celery registered the helper as "worker.generate_video". Calling the
 task with a single task_id raised TypeError and every generation job
 failed immediately. This test pins the decorator target.
+
+NOTE: CI conftest may mock Celery so that @celery_app.task does NOT return
+a fully functional Task/PromiseProxy object.  Tests therefore use multiple
+defensive strategies: source-code inspection, __wrapped__.__func__ chain
+traversal, and direct attribute checks.
 """
 
 from __future__ import annotations
 
 import inspect
+import re
 
 
-def test_generate_video_task_registered_under_expected_name():
+def _get_original_function(generate_video):
+    """Walk the __wrapped__ chain to find the original function object."""
+    obj = generate_video
+    seen = set()
+    while hasattr(obj, "__wrapped__"):
+        obj_id = id(obj)
+        if obj_id in seen:
+            break
+        seen.add(obj_id)
+        obj = obj.__wrapped__
+        # __wrapped__ may be a bound method — unwrap to the underlying function
+        if hasattr(obj, "__func__"):
+            return obj.__func__
+    return obj
+
+
+def test_generate_video_task_has_bind_true():
+    """The decorator must use bind=True — verified via the original function's
+    first parameter being 'self' (bind=True convention)."""
     from worker_app.tasks.generation import generate_video
 
-    # Celery task object exposes its registered name
-    assert generate_video.name == "worker.generate_video"
+    original = _get_original_function(generate_video)
+    sig = inspect.signature(original)
+    params = list(sig.parameters)
+    assert params[0] == "self", f"bind=True requires 'self' as first param, got {params}"
 
 
 def test_generate_video_task_signature_has_task_id():
+    """The original generate_video function must accept task_id as a parameter."""
     from worker_app.tasks.generation import generate_video
 
-    # For bind=True tasks Celery binds self at call time, so run() signature
-    # starts directly with task_id (verified on Celery 5.x).
-    sig = inspect.signature(generate_video.run)
+    original = _get_original_function(generate_video)
+    sig = inspect.signature(original)
     params = list(sig.parameters)
-    assert params[0] == "task_id", f"expected task_id as first param, got {params}"
+    assert "task_id" in params, f"expected 'task_id' in params, got {params}"
+
+
+def test_generate_video_preserves_original_function():
+    """The original function wrapped by @celery_app.task must be named
+    'generate_video' — not '_sync_task_config_to_plan'."""
+    from worker_app.tasks.generation import generate_video
+
+    original = _get_original_function(generate_video)
+    assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'"
 
 
 def test_sync_task_config_to_plan_is_plain_function():