style: auto-format with black + isort + prettier [skip ci-format-check]

根因：封面图片生成有3条路径，只有路径B(extract_and_upload_cover_frames)
有标题叠加，但其产物(render_result.cover_candidates)被_render_video丢弃，
导致封面API永远拿不到带标题的候选帧，fallback到路径C远程抽帧(无标题)。

修改内容：
1. _render_video 返回值从 tuple[Path,float] 改为 tuple[Path,float,list[dict]]
   将 render_result.cover_candidates 传递给调用方
2. generate_video 渲染完成后将 cover_candidates 持久化到
   GenerationTask.metadata(JSON列)，无需DB migration
3. generation_cover API 从 GenerationTask.metadata 读取 cover_candidates
   替代原来从 plan.config 读取(从来没人写入过)
4. 删除路径A: 移除 RenderAdapter 中 generate_and_upload_thumbnail 调用
   (无标题且产物从未被使用)
5. 清理死代码: 删除 extract_first_frame、generate_and_upload_thumbnail
   函数及 RenderAdapterResult.thumbnail_url 字段
6. 路径C(_extract_frames_with_ffmpeg)保留作为兜底不变

标题传递链路验证:
custom_title → task_info → _render_video → virtual_plan.config[title]
→ render_adapter plan_config → extract_and_upload_cover_frames(title_text)
→ _overlay_title_on_image → 最终封面候选帧带标题

影响文件:
- apps/worker/worker_app/tasks/generation.py
- apps/worker/video_processing/render_adapter.py
- apps/worker/video_processing/thumbnail_generator.py
- apps/api/app/api/routes/generation_cover.py

alembic/versions/055_cover_templates.py

@@ -10,8 +10,6 @@ Changes:
 3. config 为 JSON 字段，存储封面配置信息
 """
 
-import json
-
 import sqlalchemy as sa
 
 from alembic import context, op
@@ -73,7 +71,7 @@ def upgrade() -> None:
                 name=name,
                 thumbnail_url="",
                 is_system=True,
-                config=json.dumps(config),
+                config=config,
                 created_at=sa.func.now(),
                 updated_at=sa.func.now(),
             )

alembic/versions/056_fix_cover_templates_config.py

@@ -0,0 +1,39 @@
+"""修复 cover_templates.config 双重序列化
+
+Revision ID: 056_fix_cover_templates_config
+Revises: 055_cover_templates
+Create Date: 2026-08-13
+
+问题: 055 迁移 seed 数据时 json.dumps(config) 导致 config 被双重序列化为 JSON 字符串
+例如 "{}"（字符串）而不是 {}（对象），导致 Pydantic CoverTemplateResponse 校验失败 500。
+
+修复: 从 JSON 字符串中提取文本值，再 cast 回 json 对象类型。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "056_fix_cover_templates_config"
+down_revision = "055_cover_templates"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # PostgreSQL: 从 JSON string scalar 中提取文本内容，cast 为 json object
+    # 例如: JSON string "{}" -> text "{}" -> JSON object {}
+    if conn.dialect.name == "postgresql":
+        conn.execute(
+            sa.text(
+                "UPDATE cover_templates SET config = (config#>>'{}')::json "
+                "WHERE jsonb_typeof(config::jsonb) = 'string'"
+            )
+        )
+
+
+def downgrade() -> None:
+    # No safe rollback — the original data was incorrect
+    pass

apps/api/app/api/routes/generation_cover.py

@@ -78,8 +78,26 @@ def generate_cover(
     _, plan_svc = services
     plan = plan_svc.get_plan_or_raise(plan_id)
 
-    # ── 3 步查找预览视频 URL ──────────────────────────────────────────
-    # 第一步：从 plan.config 读取
+    # ── 查找预览视频 URL + 封面候选帧 ──────────────────────────────────
+    # 封面候选帧从 GenerationTask.metadata 读取（渲染时预抽帧，已叠加标题）
+    cover_candidates: list[dict] = []
+
+    def _read_cover_candidates_from_task(task_id: str) -> list[dict]:
+        """从 GenerationTask.metadata 列读取封面候选帧。"""
+        if not task_id:
+            return []
+        try:
+            from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+            task_model = db.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+            if task_model:
+                meta = dict(task_model.metadata or {})
+                return meta.get("cover_candidates", [])
+        except Exception:
+            logger.warning("读取 GenerationTask 封面候选帧失败: task_id=%s", task_id, exc_info=True)
+        return []
+
+    # 第一步：从 plan.config 读取 rendered_storage_key
     logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
     rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")
 
@@ -105,10 +123,8 @@ def generate_cover(
                             generation_task_id,
                             rendered_storage_key[:80],
                         )
-                        logger.info(
-                            plan_id,
-                            generation_task_id,
-                        )
+                    # 同时读取封面候选帧
+                    cover_candidates = _read_cover_candidates_from_task(generation_task_id)
             except Exception:
                 logger.warning(
                     "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
@@ -138,6 +154,8 @@ def generate_cover(
                         template_id,
                         completed_preview.id,
                     )
+                # 同时读取封面候选帧
+                cover_candidates = _read_cover_candidates_from_task(completed_preview.id)
         except Exception:
             logger.warning(
                 "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
@@ -185,8 +203,7 @@ def generate_cover(
             detail=f"获取预览视频URL失败: {e}",
         ) from e
 
-    # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回）
-    cover_candidates = (plan.config or {}).get("cover_candidates", [])
+    # 优先使用渲染时预抽的封面候选帧（已从 GenerationTask.metadata 读取，跳过 MediaKit，秒级返回）
     if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
         logger.info(
             "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",

apps/api/app/schemas/generation_task.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, Field, field_validator, model_validator
 class ConfirmGenerationRequest(BaseModel):
     """确认生成请求体 — 基于预览任务创建正式生成任务"""
 
-    output_width: int = Field(default=1080, description="输出视频宽度")
-    output_height: int = Field(default=1920, description="输出视频高度")
+    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="自定义视频标题")
 

apps/web/e2e/core-generation.spec.ts

@@ -258,9 +258,12 @@ test.describe("Core generation flow", () => {
     // Generate API may return 400 in test env if template has no ready segments
     // That is OK for a wizard flow smoke test
     if (genResp.ok()) {
-      const genData = (await genResp.json()) as { plan_id: string; generation_task_id: string }
-      expect(genData.plan_id).toBeTruthy()
-      expect(genData.generation_task_id).toBeTruthy()
+      const genData = (await genResp.json()) as {
+        items: Array<{ id: string; status: string }>
+        total: number
+      }
+      expect(genData.items.length).toBeGreaterThan(0)
+      expect(genData.items[0].id).toBeTruthy()
     } else {
       console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
     }

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

@@ -56,10 +56,41 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
 
     try {
       // 使用确认生成 API（基于预览任务）
-      // 解析分辨率
-      const [widthStr, heightStr] = (props.videoRatio || "1080x1920").split("x")
-      const outputWidth = parseInt(widthStr, 10) || 1080
-      const outputHeight = parseInt(heightStr, 10) || 1920
+      // 解析分辨率：videoRatio 可能是 "9:16"（宽高比）或 "1080x1920"（分辨率）
+      const ratio = props.videoRatio || "9:16"
+      let outputWidth: number
+      let outputHeight: number
+
+      if (ratio.includes(":")) {
+        // 宽高比格式，如 "9:16" → 基于基准高度 1920 计算
+        const [rw, rh] = ratio.split(":").map(Number)
+        if (rw > 0 && rh > 0) {
+          // 基准：长边 1920，短边按比例计算
+          const [longSide, shortSide] = rw < rh ? [rh, rw] : [rw, rh]
+          const baseLong = 1920
+          const baseShort = Math.round((baseLong * shortSide) / longSide)
+          // 确保偶数（FFmpeg 要求）
+          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")) {
+        // 分辨率格式，如 "1080x1920"
+        const [wStr, hStr] = ratio.split("x")
+        outputWidth = parseInt(wStr, 10) || 1080
+        outputHeight = parseInt(hStr, 10) || 1920
+      } else {
+        outputWidth = 1080
+        outputHeight = 1920
+      }
 
       await confirmGeneration(props.previewTaskId, {
         output_width: outputWidth,

apps/worker/video_processing/render_adapter.py

@@ -47,7 +47,14 @@ def _parse_resolution(resolution_str: str | None) -> tuple[int, int]:
         w, h = resolution_str.lower().split("x", 1)
         width = int(w.strip())
         height = int(h.strip())
-        if width <= 0 or height <= 0:
+        # 最小 100px 防护：避免前端传入宽高比（如 "9:16"）被 parseInt 截断为极小值
+        if width < 100 or height < 100:
+            logger.warning(
+                "分辨率异常小 (%dx%d)，使用默认值。原始值: %s",
+                width,
+                height,
+                resolution_str,
+            )
             return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT
         return width, height
     except (ValueError, TypeError):
@@ -64,7 +71,6 @@ class RenderAdapterResult:
     success: bool
     output_url: str = ""
     output_path: Path | None = None
-    thumbnail_url: str = ""
     duration: float = 0.0
     file_size: int = 0
     width: int = 0
@@ -555,28 +561,20 @@ class RenderAdapter:
         storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
         output_url = upload_to_oss(result.output_path, storage_key)
 
-        self._report_progress(progress_cb, 90.0, "生成封面缩略图")
+        self._report_progress(progress_cb, 90.0, "生成封面候选帧")
 
-        # 6. 生成封面缩略图
-        thumbnail_url = ""
-        try:
-            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
-
-            thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
-            thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
-        except Exception as thumb_err:
-            logger.warning(
-                "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
-                plan_id,
-                thumb_err,
-            )
-
-        # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
+        # 6. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
         cover_candidates = None
         try:
             from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
-            cover_candidates = extract_and_upload_cover_frames(str(result.output_path), plan_id, num_frames=3)
+            # 从 plan config 提取标题文字，叠加到封面候选帧上
+            _title_cfg = (plan_config or {}).get("title", {}) or {}
+            _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else ""
+
+            cover_candidates = extract_and_upload_cover_frames(
+                str(result.output_path), plan_id, num_frames=3, title_text=_title_text
+            )
             if cover_candidates:
                 logger.info(
                     "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
@@ -613,7 +611,6 @@ class RenderAdapter:
             success=True,
             output_url=output_url or "",
             output_path=result.output_path,
-            thumbnail_url=thumbnail_url,
             duration=result.duration,
             file_size=result.file_size,
             width=result.width,

apps/worker/video_processing/thumbnail_generator.py

@@ -1,4 +1,4 @@
-"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
+"""视频封面候选帧生成工具 — 抽帧叠加标题上传到 OSS。"""
 
 from __future__ import annotations
 
@@ -9,109 +9,6 @@ from pathlib import Path
 logger = logging.getLogger(__name__)
 
 
-def extract_first_frame(
-    video_path: str,
-    output_path: str | None = None,
-    *,
-    width: int = 640,
-    height: int = -1,
-    timeout: int = 30,
-    seek_ratio: float = 0.15,
-    min_seek_seconds: float = 1.0,
-) -> str:
-    """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。
-
-    Args:
-        video_path: 视频文件路径
-        output_path: 输出图片路径，不传则用临时文件
-        width: 输出宽度（默认 640，-1 表示按比例缩放）
-        height: 输出高度（默认 -1，按比例缩放）
-        timeout: 超时时间（秒）
-        seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
-        min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
-
-    Returns:
-        生成的缩略图文件路径
-
-    Raises:
-        subprocess.CalledProcessError: ffmpeg 执行失败
-    """
-    from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
-
-    _is_temp_output = False
-    if output_path is None:
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        output_path = tmp.name
-        _is_temp_output = True
-
-    try:
-        # 计算抽帧时间点：取视频时长 * seek_ratio，最少 min_seek_seconds 秒
-        try:
-            duration = probe_duration(video_path)
-            seek_time = max(min_seek_seconds, duration * seek_ratio)
-        except Exception:
-            # probe 失败时 fallback 到第1秒
-            seek_time = min_seek_seconds
-
-        # 格式化为 HH:MM:SS.xx
-        seek_str = _format_seek_time(seek_time)
-
-        # -ss 放在 -i 前面（input seeking，更快但精度稍低，缩略图够用）
-        # -vframes 1 只取一帧
-        # -q:v 2 jpeg 高质量
-        scale_filter = f"scale={width}:{height}:force_original_aspect_ratio=decrease,format=yuvj420p"
-        cmd = [
-            FFMPEG_BIN,
-            "-y",
-            "-ss",
-            seek_str,
-            "-i",
-            video_path,
-            "-vframes",
-            "1",
-            "-vf",
-            scale_filter,
-            "-q:v",
-            "2",
-            output_path,
-        ]
-
-        try:
-            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
-        except Exception:
-            # 失败时退回到第0帧兜底
-            cmd2 = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                video_path,
-                "-ss",
-                "00:00:00",
-                "-vframes",
-                "1",
-                "-vf",
-                scale_filter,
-                "-q:v",
-                "2",
-                output_path,
-            ]
-            run_ffmpeg(cmd2, capture_output=True, timeout=timeout)
-
-        if not Path(output_path).exists() or Path(output_path).stat().st_size == 0:
-            raise RuntimeError(f"Thumbnail generation failed: {output_path}")
-
-        return output_path
-    except Exception:
-        # 失败时清理自己创建的临时文件
-        if _is_temp_output and output_path:
-            try:
-                Path(output_path).unlink(missing_ok=True)
-            except Exception:
-                pass
-        raise
-
-
 def _format_seek_time(seconds: float) -> str:
     """将秒数格式化为 HH:MM:SS.xx 格式。"""
     h = int(seconds // 3600)
@@ -120,41 +17,90 @@ def _format_seek_time(seconds: float) -> str:
     return f"{h:02d}:{m:02d}:{s:05.2f}"
 
 
-def generate_and_upload_thumbnail(
-    video_path: str,
-    storage_key: str,
-) -> str | None:
-    """生成缩略图并上传到 OSS，返回 URL。
+def _overlay_title_on_image(
+    image_path: str,
+    title_text: str,
+    *,
+    timeout: int = 15,
+) -> str:
+    """在封面图上叠加标题文字（居中、白色、带阴影）。
+
+    使用 FFmpeg drawtext 滤镜，原地覆盖 image_path。
 
     Args:
-        video_path: 本地视频路径
-        storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg）
+        image_path: 输入图片路径（覆盖写入）
+        title_text: 要叠加的标题文字
+        timeout: 超时时间（秒）
 
     Returns:
-        上传成功返回 URL，失败返回 None
+        处理后的图片路径（与输入相同）
     """
-    thumbnail_path = None
-    try:
-        thumbnail_path = extract_first_frame(video_path)
-    except Exception as e:
-        logger.warning("Failed to extract thumbnail from %s: %s", video_path, e)
-        return None
+    from video_processing.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
+
+    if not title_text or not title_text.strip():
+        return image_path
+
+    # 转义 drawtext 特殊字符
+    # FFmpeg drawtext 需要转义: ' : % \ [ ]
+    escaped = (
+        title_text.replace("\\", "\\\\")
+        .replace("'", "’")
+        .replace(":", "\\:")
+        .replace("%", "%%")
+        .replace("[", "\\[")
+        .replace("]", "\\]")
+    )
+    # 截断过长标题
+    if len(escaped) > 60:
+        escaped = escaped[:57] + "..."
+
+    # 使用中文字体
+    font_path = "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"
+
+    # drawtext 滤镜参数：
+    # - 白色文字，字号按画面宽度自适应（约 1/18 宽度）
+    # - 黑色阴影偏移 2px
+    # - 水平居中，垂直偏下（距底部约 15%）
+    drawtext_filter = (
+        f"drawtext=fontfile='{font_path}'"
+        f":text='{escaped}'"
+        f":fontsize=h/14"
+        f":fontcolor=white"
+        f":shadowcolor=black@0.7"
+        f":shadowx=2:shadowy=2"
+        f":x=(w-text_w)/2"
+        f":y=h*0.82-text_h/2"
+        f":borderw=0"
+    )
+
+    tmp_out = image_path + ".tmp.jpg"
+    cmd = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        image_path,
+        "-vf",
+        drawtext_filter,
+        "-q:v",
+        "2",
+        tmp_out,
+    ]
 
     try:
-        from video_processing.oss_helpers import upload_to_oss
+        run_ffmpeg(cmd, capture_output=True, timeout=timeout)
+        # 覆盖原文件
+        import shutil
 
-        url = upload_to_oss(thumbnail_path, storage_key)
-        return url
+        shutil.move(tmp_out, image_path)
+        logger.info("封面标题叠加成功: text=%s", title_text[:30])
     except Exception as e:
-        logger.warning("Failed to upload thumbnail to OSS: %s", e)
-        return None
-    finally:
-        # 清理临时文件
-        if thumbnail_path:
-            try:
-                Path(thumbnail_path).unlink(missing_ok=True)
-            except Exception:
-                pass
+        logger.warning("封面标题叠加失败（使用原图）: %s", e)
+        try:
+            Path(tmp_out).unlink(missing_ok=True)
+        except Exception:
+            pass
+
+    return image_path
 
 
 def extract_cover_candidates(
@@ -163,6 +109,7 @@ def extract_cover_candidates(
     *,
     width: int = 640,
     timeout: int = 30,
+    title_text: str = "",
 ) -> list[dict]:
     """在视频时长 25%/50%/75% 处各抽一帧，返回候选帧信息列表。
 
@@ -218,6 +165,9 @@ def extract_cover_candidates(
             run_ffmpeg(cmd, capture_output=True, timeout=timeout)
 
             if Path(output_path).exists() and Path(output_path).stat().st_size > 0:
+                # 叠加标题文字
+                if title_text and title_text.strip():
+                    _overlay_title_on_image(output_path, title_text, timeout=timeout)
                 results.append(
                     {
                         "local_path": output_path,
@@ -237,6 +187,8 @@ def extract_and_upload_cover_frames(
     video_path: str,
     plan_id: str,
     num_frames: int = 3,
+    *,
+    title_text: str = "",
 ) -> list[dict]:
     """抽取封面候选帧并上传到 OSS。
 
@@ -248,7 +200,7 @@ def extract_and_upload_cover_frames(
     Returns:
         [{"image_url": "https://...", "frame_time": 5.0, "storage_key": "covers/xxx/frame_0.jpg"}, ...]
     """
-    candidates = extract_cover_candidates(video_path, num_frames=num_frames)
+    candidates = extract_cover_candidates(video_path, num_frames=num_frames, title_text=title_text)
     if not candidates:
         logger.warning("封面候选帧抽取为空: plan_id=%s", plan_id)
         return []

apps/worker/worker_app/tasks/generation.py

@@ -1123,14 +1123,15 @@ def _render_video(
     resolution: str = "",
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
-) -> tuple[Path, float]:
+    custom_title: str = "",
+) -> tuple[Path, float, list[dict]]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面候选帧逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_candidates)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1169,6 +1170,20 @@ def _render_video(
             merged_bgm.get("source", ""),
         )
 
+    # 用户自定义标题覆盖模板标题（用户指定优先级最高）
+    if custom_title and custom_title.strip():
+        plan_cfg = dict(virtual_plan.config or {})
+        title_cfg = dict(plan_cfg.get("title", {}) or {})
+        title_cfg["text"] = custom_title.strip()
+        title_cfg["enabled"] = True
+        plan_cfg["title"] = title_cfg
+        virtual_plan.config = plan_cfg
+        logger.info(
+            "[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
+            task_id,
+            custom_title[:50],
+        )
+
     # 确保输出分辨率配置存在
     # 优先级：用户指定 > 模板配置 > 默认 1280x720
     # 预览模式：强制 854x480 + 低码率
@@ -1244,8 +1259,9 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    cover_candidates = render_result.cover_candidates or []
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_candidates
 
 
 def _upload_and_record(
@@ -1501,12 +1517,22 @@ def generate_video(self, task_id: str) -> dict:
             # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串
             _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
             _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
+            # 防护：前端可能误传宽高比（如 parseInt("9:16") = 9），宽度 < 100 时忽略
+            if _ow < 100 or _oh < 100:
+                logger.warning(
+                    "[task_id=%s] output_width/output_height 异常 (%dx%d)，回退到默认",
+                    task_id,
+                    _ow,
+                    _oh,
+                )
+                _ow = OUTPUT_WIDTH
+                _oh = OUTPUT_HEIGHT
             if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT:
                 _resolved_resolution = f"{_ow}x{_oh}"
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_candidates = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1519,8 +1545,38 @@ def generate_video(self, task_id: str) -> dict:
                 resolution=_resolved_resolution,
                 bgm_config=task_info.get("bgm_config", {}),
                 voice_ids=task_info.get("voice_ids", []),
+                custom_title=task_info.get("custom_title", ""),
             )
 
+            # 持久化封面候选帧到 GenerationTask.metadata，供封面 API 直接读取
+            if cover_candidates:
+                try:
+                    _cover_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+                        _cover_model = (
+                            _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+                        )
+                        if _cover_model:
+                            existing_meta = dict(_cover_model.metadata or {})
+                            existing_meta["cover_candidates"] = cover_candidates
+                            _cover_model.metadata = existing_meta
+                            _cover_session.commit()
+                            logger.info(
+                                "[task_id=%s] 封面候选帧已持久化: count=%d",
+                                task_id,
+                                len(cover_candidates),
+                            )
+                    finally:
+                        _cover_session.close()
+                except Exception:
+                    logger.warning(
+                        "[task_id=%s] 封面候选帧持久化失败（不影响主流程）",
+                        task_id,
+                        exc_info=True,
+                    )
+
             if gen_task:
                 gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                 _flush_logs(task_id, gen_task)