diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py
index c9bd2bf4d..3e0e99035 100644
--- a/apps/api/app/api/routes/generation_cover.py
+++ b/apps/api/app/api/routes/generation_cover.py
@@ -185,27 +185,39 @@ def generate_cover(
             detail=f"获取预览视频URL失败: {e}",
         ) from e
 
-    # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回）
-    cover_candidates = (plan.config or {}).get("cover_candidates", [])
-    if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
-        logger.info(
-            "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",
-            plan_id,
-            len(cover_candidates),
-        )
-        first_frame = cover_candidates[0]
-        cover_data = {
-            "type": "ai_frame",
-            "image_url": first_frame.get("image_url", ""),
-            "frame_time": first_frame.get("frame_time", 0.0),
-            "confidence": 0.9,
-        }
-        if cover_data["image_url"]:
-            current_config = dict(plan.config) if plan.config else {}
-            current_config["cover"] = cover_data
-            normalized = normalize_plan_config(current_config)
-            plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
-            return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
+    # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
+    if body.cover_type in ("ai_frame", "ai_regenerate"):
+        # 尝试从 GenerationTask 读取已持久化的封面 URL
+        generation_task_id = (plan.config or {}).get("generation_task_id", "")
+        if generation_task_id:
+            try:
+                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+                task = gen_task_repo.get(generation_task_id)
+                if task and getattr(task, "cover_url", ""):
+                    cover_data = {
+                        "type": "ai_frame",
+                        "image_url": task.cover_url,
+                        "frame_time": 0.0,
+                        "confidence": 0.95,
+                    }
+                    logger.info(
+                        "[封面生成] 使用统一管道封面: plan_id=%s task_id=%s url=%s",
+                        plan_id,
+                        generation_task_id,
+                        task.cover_url[:80],
+                    )
+                    current_config = dict(plan.config) if plan.config else {}
+                    current_config["cover"] = cover_data
+                    normalized = normalize_plan_config(current_config)
+                    plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
+                    return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
+            except Exception:
+                logger.warning(
+                    "[封面生成] 读取 GenerationTask.cover_url 失败: plan_id=%s task_id=%s",
+                    plan_id,
+                    generation_task_id,
+                    exc_info=True,
+                )
 
     from packages.shared.ai_service import run_generate_cover
 
diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py
index 4912031f1..d339f7144 100755
--- a/apps/worker/video_processing/dedup_helpers.py
+++ b/apps/worker/video_processing/dedup_helpers.py
@@ -85,19 +85,9 @@ def create_video_record_and_dedup(
         if thumbnail_url:
             generated_video.thumbnail_url = thumbnail_url
             video_repo.update_thumbnail(video_id, thumbnail_url)
-            logger.info("Thumbnail reused (pre-generated) for video %s", video_id)
+            logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80] if thumbnail_url else "")
         else:
-            thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg"
-            try:
-                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
-
-                _thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key)
-                if _thumbnail_url:
-                    generated_video.thumbnail_url = _thumbnail_url
-                    video_repo.update_thumbnail(video_id, _thumbnail_url)
-                    logger.info("Thumbnail generated for video %s: %s", video_id, _thumbnail_url)
-            except Exception as thumb_err:
-                logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err)
+            logger.debug("No thumbnail_url provided for video %s, skipping", video_id)
 
         # 计算视频指纹
         deduplicator = VideoDeduplicator()
diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py
index bd14d13a0..ce5907b61 100755
--- a/apps/worker/video_processing/render_adapter.py
+++ b/apps/worker/video_processing/render_adapter.py
@@ -81,9 +81,7 @@ class RenderAdapterResult:
     failed_clip_ids: list[str] = None  # 失败的 clip id 列表
     error_message: str = ""
     error_detail: str = ""  # 详细错误信息（如 ffmpeg stderr），用于排查
-    cover_candidates: list[dict] | None = (
-        None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
-    )
+    cover_url: str = ""  # 封面图片 URL（从渲染后视频抽帧，天然带标题）
 
     def __post_init__(self):
         if self.rendered_clip_ids is None:
@@ -562,43 +560,33 @@ 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 = ""
+        # 6. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题）
+        cover_url = ""
+        cover_frame_path = None
         try:
-            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+            from video_processing.thumbnail_generator import extract_first_frame
 
-            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（失败不阻断主流程）
-        cover_candidates = None
-        try:
-            from video_processing.thumbnail_generator import extract_and_upload_cover_frames
-
-            # 从 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:
+            cover_frame_path = extract_first_frame(str(result.output_path), width=640)
+            cover_storage_key = f"rendered/{plan_id}/cover.jpg"
+            try:
+                cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or ""
+            finally:
+                if cover_frame_path:
+                    try:
+                        Path(cover_frame_path).unlink(missing_ok=True)
+                    except Exception:
+                        pass
+            if cover_url:
                 logger.info(
-                    "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
+                    "[render-adapter] 封面帧提取成功: plan_id=%s url=%s",
                     plan_id,
-                    len(cover_candidates),
+                    cover_url[:80],
                 )
         except Exception as cover_err:
             logger.warning(
-                "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s",
+                "[render-adapter] 封面帧提取失败（不影响主流程）: plan_id=%s error=%s",
                 plan_id,
                 cover_err,
             )
@@ -626,7 +614,7 @@ class RenderAdapter:
             success=True,
             output_url=output_url or "",
             output_path=result.output_path,
-            thumbnail_url=thumbnail_url,
+            thumbnail_url=cover_url,
             duration=result.duration,
             file_size=result.file_size,
             width=result.width,
@@ -634,7 +622,7 @@ class RenderAdapter:
             clip_count=len(clips),
             rendered_clip_ids=final_rendered_ids,
             failed_clip_ids=final_failed_ids,
-            cover_candidates=cover_candidates,
+            cover_url=cover_url,
         )
 
     def render_from_memory(
diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py
index 55cdc28c9..86f4a32ac 100755
--- a/apps/worker/video_processing/thumbnail_generator.py
+++ b/apps/worker/video_processing/thumbnail_generator.py
@@ -1,4 +1,8 @@
-"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
+"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。
+
+统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
+渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
+"""
 
 from __future__ import annotations
 
@@ -13,28 +17,30 @@ def extract_first_frame(
     video_path: str,
     output_path: str | None = None,
     *,
-    width: int = 640,
+    width: int = -1,
     height: int = -1,
     timeout: int = 30,
     seek_ratio: float = 0.15,
     min_seek_seconds: float = 1.0,
 ) -> str:
-    """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。
+    """抽取视频封面帧（默认取视频时长 15% 处的帧，避开片头纯色画面）。
+
+    因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。
 
     Args:
         video_path: 视频文件路径
         output_path: 输出图片路径，不传则用临时文件
-        width: 输出宽度（默认 640，-1 表示按比例缩放）
-        height: 输出高度（默认 -1，按比例缩放）
+        width: 输出宽度（默认 -1，保持原始分辨率）
+        height: 输出高度（默认 -1，保持原始分辨率）
         timeout: 超时时间（秒）
         seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
         min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
 
     Returns:
-        生成的缩略图文件路径
+        生成的封面帧文件路径
 
     Raises:
-        subprocess.CalledProcessError: ffmpeg 执行失败
+        RuntimeError: ffmpeg 执行失败或输出文件为空
     """
     from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 
@@ -57,10 +63,20 @@ def extract_first_frame(
         # 格式化为 HH:MM:SS.xx
         seek_str = _format_seek_time(seek_time)
 
-        # -ss 放在 -i 前面（input seeking，更快但精度稍低，缩略图够用）
+        # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率。
+        # NOTE: scale_filter 在此处通过 if/else 分支赋值，之后不再被覆盖，
+        #       后续 cmd / cmd2 均复用同一变量，逻辑无变化。
+        if width > 0 or height > 0:
+            w_str = str(width) if width > 0 else "-1"
+            h_str = str(height) if height > 0 else "-1"
+            scale_filter = f"scale={w_str}:{h_str}:force_original_aspect_ratio=decrease,format=yuvj420p"
+        else:
+            # 保持原始分辨率，只确保格式兼容
+            scale_filter = "format=yuvj420p"
+
+        # -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",
@@ -99,7 +115,7 @@ def extract_first_frame(
             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}")
+            raise RuntimeError(f"Cover frame extraction failed: {output_path}")
 
         return output_path
     except Exception:
@@ -118,263 +134,3 @@ def _format_seek_time(seconds: float) -> str:
     m = int((seconds % 3600) // 60)
     s = seconds % 60
     return f"{h:02d}:{m:02d}:{s:05.2f}"
-
-
-def _overlay_title_on_image(
-    image_path: str,
-    title_text: str,
-    *,
-    timeout: int = 15,
-) -> str:
-    """在封面图上叠加标题文字（居中、白色、带阴影）。
-
-    使用 FFmpeg drawtext 滤镜，原地覆盖 image_path。
-
-    Args:
-        image_path: 输入图片路径（覆盖写入）
-        title_text: 要叠加的标题文字
-        timeout: 超时时间（秒）
-
-    Returns:
-        处理后的图片路径（与输入相同）
-    """
-    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:
-        run_ffmpeg(cmd, capture_output=True, timeout=timeout)
-        # 覆盖原文件
-        import shutil
-
-        shutil.move(tmp_out, image_path)
-        logger.info("封面标题叠加成功: text=%s", title_text[:30])
-    except Exception as e:
-        logger.warning("封面标题叠加失败（使用原图）: %s", e)
-        try:
-            Path(tmp_out).unlink(missing_ok=True)
-        except Exception:
-            pass
-
-    return image_path
-
-
-def generate_and_upload_thumbnail(
-    video_path: str,
-    storage_key: str,
-) -> str | None:
-    """生成缩略图并上传到 OSS，返回 URL。
-
-    Args:
-        video_path: 本地视频路径
-        storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg）
-
-    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
-
-    try:
-        from video_processing.oss_helpers import upload_to_oss
-
-        url = upload_to_oss(thumbnail_path, storage_key)
-        return url
-    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
-
-
-def extract_cover_candidates(
-    video_path: str,
-    num_frames: int = 3,
-    *,
-    width: int = 640,
-    timeout: int = 30,
-    title_text: str = "",
-) -> list[dict]:
-    """在视频时长 25%/50%/75% 处各抽一帧，返回候选帧信息列表。
-
-    Args:
-        video_path: 视频文件路径
-        num_frames: 抽帧数量（默认 3）
-        width: 输出宽度
-        timeout: 单帧超时（秒）
-
-    Returns:
-        [{"local_path": "...", "frame_time": 5.0}, ...]
-    """
-    from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
-
-    try:
-        duration = probe_duration(video_path)
-    except Exception:
-        duration = 0.0
-
-    if duration <= 0:
-        duration = 5.0  # fallback
-
-    # 计算抽帧时间点：25%, 50%, 75%
-    ratios = []
-    for i in range(1, num_frames + 1):
-        ratios.append(i / (num_frames + 1))
-
-    results = []
-    for _idx, ratio in enumerate(ratios):
-        frame_time = max(0.5, duration * ratio)
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        output_path = tmp.name
-
-        try:
-            seek_str = _format_seek_time(frame_time)
-            scale_filter = f"scale={width}:-1: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,
-            ]
-            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,
-                        "frame_time": round(frame_time, 2),
-                    }
-                )
-            else:
-                Path(output_path).unlink(missing_ok=True)
-        except Exception as e:
-            logger.warning("封面候选帧抽取失败 ratio=%.2f: %s", ratio, e)
-            Path(output_path).unlink(missing_ok=True)
-
-    return results
-
-
-def extract_and_upload_cover_frames(
-    video_path: str,
-    plan_id: str,
-    num_frames: int = 3,
-    *,
-    title_text: str = "",
-) -> list[dict]:
-    """抽取封面候选帧并上传到 OSS。
-
-    Args:
-        video_path: 本地视频路径
-        plan_id: 剪辑计划 ID（用于 OSS 路径）
-        num_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, title_text=title_text)
-    if not candidates:
-        logger.warning("封面候选帧抽取为空: plan_id=%s", plan_id)
-        return []
-
-    results = []
-    for idx, cand in enumerate(candidates):
-        local_path = cand["local_path"]
-        frame_time = cand["frame_time"]
-        storage_key = f"covers/{plan_id}/frame_{idx}.jpg"
-
-        try:
-            from video_processing.oss_helpers import upload_to_oss
-
-            url = upload_to_oss(local_path, storage_key)
-            if url:
-                results.append(
-                    {
-                        "image_url": url,
-                        "frame_time": frame_time,
-                        "storage_key": storage_key,
-                    }
-                )
-                logger.info(
-                    "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f",
-                    plan_id,
-                    idx,
-                    frame_time,
-                )
-        except Exception as e:
-            logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e)
-        finally:
-            try:
-                Path(local_path).unlink(missing_ok=True)
-            except Exception:
-                pass
-
-    return results
diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py
index e422739e9..33981b8ff 100644
--- a/apps/worker/worker_app/tasks/generation.py
+++ b/apps/worker/worker_app/tasks/generation.py
@@ -1124,14 +1124,14 @@ def _render_video(
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
     custom_title: str = "",
-) -> tuple[Path, float]:
+) -> tuple[Path, float, str]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_url)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1259,8 +1259,9 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    cover_url = getattr(render_result, "cover_url", "") or ""
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_url
 
 
 def _upload_and_record(
@@ -1271,6 +1272,7 @@ def _upload_and_record(
     editing_mode,
     user_id: str = "",
     video_name: str = "",
+    thumbnail_url: str = "",
 ) -> tuple[str, float, int, int]:
     """上传 OSS、创建视频记录并查重。
 
@@ -1531,7 +1533,7 @@ def generate_video(self, task_id: str) -> dict:
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_url = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1551,6 +1553,35 @@ def generate_video(self, task_id: str) -> dict:
                 gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                 _flush_logs(task_id, gen_task)
 
+            # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧）
+            if cover_url:
+                _cover_session = None
+                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:
+                            _cover_model.cover_url = cover_url
+                            _cover_session.commit()
+                            logger.info(
+                                "[task_id=%s] 封面URL已持久化: %s",
+                                task_id,
+                                cover_url[:80],
+                            )
+                    finally:
+                        if _cover_session:
+                            _cover_session.close()
+                except Exception as cover_err:
+                    logger.warning(
+                        "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s",
+                        task_id,
+                        cover_err,
+                    )
+
             _update_task_progress(task_id, 80, "渲染完成")
 
             # ── 4. 上传 OSS + 查重记录 ───────────────────────────────────────
@@ -1563,6 +1594,7 @@ def generate_video(self, task_id: str) -> dict:
                 editing_mode=editing_mode,
                 user_id=user_id,
                 video_name=task_info.get("video_title", ""),
+                thumbnail_url=cover_url,
             )
 
             if gen_task:
diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py
index 82ddb686b..6449b8d0f 100755
--- a/apps/worker/worker_app/tasks/ingest.py
+++ b/apps/worker/worker_app/tasks/ingest.py
@@ -206,11 +206,21 @@ def ingest_asset(job_id: str) -> dict:
             # 视频类型：生成缩略图（文件还在的时候生成）
             thumbnail_url = None
             if media_type == "video" and extract_success:
+                frame_path = None
                 try:
-                    from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+                    from video_processing.oss_helpers import upload_to_oss
+                    from video_processing.thumbnail_generator import extract_first_frame
 
+                    frame_path = extract_first_frame(str(local_file), width=640)
                     thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg"
-                    thumbnail_url = generate_and_upload_thumbnail(str(local_file), thumb_storage_key)
+                    try:
+                        thumbnail_url = upload_to_oss(frame_path, thumb_storage_key)
+                    finally:
+                        if frame_path:
+                            try:
+                                Path(frame_path).unlink(missing_ok=True)
+                            except Exception:
+                                pass
                     if thumbnail_url:
                         logger.info(
                             "素材缩略图生成成功: job_id=%s url=%s",
diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py
index 2ae55042e..390a32a68 100755
--- a/packages/shared/ai_service.py
+++ b/packages/shared/ai_service.py
@@ -13,8 +13,6 @@ import random
 import time
 from typing import Any, Dict, List, Optional
 
-import requests as http_requests
-
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
 from packages.shared.ai_client import get_doubao_client
 
@@ -354,93 +352,6 @@ def _transfer_cover_frame_to_storage(frame_url: str, plan_id: str) -> str:
         return frame_url
 
 
-def _extract_frames_with_ffmpeg(
-    video_url: str,
-    num_frames: int = 3,
-    timeout: int = 30,
-) -> list[dict]:
-    """用 FFmpeg 从远程视频 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）。
-
-    Args:
-        video_url: 视频 URL
-        num_frames: 抽帧数量
-        timeout: 单帧超时（秒）
-
-    Returns:
-        [{"local_path": "...", "frame_time": 5.0}, ...]
-    """
-    import re as _re
-    import tempfile
-    from pathlib import Path as _Path
-
-    from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
-
-    video_url = _re.sub(r"(?<!:)//", "/", video_url)
-
-    # 先用 ffprobe 获取视频时长
-    import subprocess as _subprocess
-
-    from packages.shared.ffmpeg_utils import FFPROBE_BIN
-
-    duration = 30.0  # 默认假设 30 秒
-    try:
-        probe_result = _subprocess.run(
-            [
-                FFPROBE_BIN,
-                "-v",
-                "error",
-                "-show_entries",
-                "format=duration",
-                "-of",
-                "default=noprint_wrappers=1:nokey=1",
-                video_url,
-            ],
-            capture_output=True,
-            text=True,
-            timeout=15,
-        )
-        if probe_result.returncode == 0 and probe_result.stdout.strip():
-            duration = float(probe_result.stdout.strip())
-    except Exception as e:
-        logger.warning("FFprobe 远程视频时长失败，使用默认值: %s", e)
-
-    ratios = [i / (num_frames + 1) for i in range(1, num_frames + 1)]
-    results = []
-
-    for _idx, ratio in enumerate(ratios):
-        frame_time = max(0.5, duration * ratio)
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        output_path = tmp.name
-
-        try:
-            seek_str = f"{int(frame_time // 3600):02d}:{int((frame_time % 3600) // 60):02d}:{frame_time % 60:05.2f}"
-            cmd = [
-                FFMPEG_BIN,
-                "-y",
-                "-ss",
-                seek_str,
-                "-i",
-                video_url,
-                "-vframes",
-                "1",
-                "-q:v",
-                "2",
-                output_path,
-            ]
-            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
-
-            if _Path(output_path).exists() and _Path(output_path).stat().st_size > 0:
-                results.append({"local_path": output_path, "frame_time": round(frame_time, 2)})
-            else:
-                _Path(output_path).unlink(missing_ok=True)
-        except Exception as e:
-            logger.warning("FFmpeg 远程抽帧失败 ratio=%.2f: %s", ratio, e)
-            _Path(output_path).unlink(missing_ok=True)
-
-    return results
-
-
 def _call_ai_cover_service(
     plan_id: str,
     asset_ids: List[str],
@@ -450,9 +361,9 @@ def _call_ai_cover_service(
 ) -> Dict[str, Any]:
     """调用 AI 封面生成服务.
 
-    优先级：
-    1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理
-    2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）
+    统一封面管道下，封面已由渲染后视频抽帧生成并持久化到 GenerationTask.cover_url。
+    此函数仅处理 manual/upload 等需要前端交互的类型，
+    ai_frame/ai_regenerate 类型应由调用方直接从持久化的封面 URL 读取。
 
     失败时抛出 RuntimeError。
 
@@ -484,88 +395,14 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧
-    if primary_video_url:
-        import re as _re
-
-        primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
-
-        # 先检查视频 URL 是否可访问
-        try:
-            head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
-            if head_resp.status_code != 200:
-                logger.error(
-                    "封面视频URL不可访问: plan_id=%s url=%s status=%d",
-                    plan_id,
-                    primary_video_url,
-                    head_resp.status_code,
-                )
-                raise RuntimeError(
-                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。" f"请重新生成预览视频后再试。"
-                )
-        except http_requests.RequestException as e:
-            logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
-            raise RuntimeError(
-                f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。请重新生成预览视频后再试。"
-            ) from e
-
-        # 使用 FFmpeg 从 URL 流式 seek 抽帧
-        try:
-            logger.info("FFmpeg 远程抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
-            frames = _extract_frames_with_ffmpeg(primary_video_url, num_frames=3)
-
-            if frames:
-                best_frame = frames[0]
-                local_path = best_frame["local_path"]
-                frame_time_val = best_frame["frame_time"]
-
-                # 上传到 OSS
-                try:
-                    import uuid
-                    from pathlib import Path
-
-                    from packages.shared.storage import get_shared_storage_service
-
-                    storage = get_shared_storage_service()
-                    cover_key = f"covers/{plan_id}/ffmpeg_frame_{uuid.uuid4().hex[:8]}.jpg"
-                    storage.upload_file(
-                        file_or_path=local_path,
-                        storage_key=cover_key,
-                        content_type="image/jpeg",
-                    )
-                    public_url = storage.get_url(cover_key)
-
-                    logger.info(
-                        "FFmpeg 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
-                        plan_id,
-                        frame_time_val,
-                        public_url[:80],
-                    )
-
-                    return {
-                        "type": "ai_frame",
-                        "image_url": public_url,
-                        "frame_time": round(frame_time_val, 1),
-                        "confidence": 0.85,
-                    }
-                finally:
-                    # 清理所有临时文件
-                    for frame in frames:
-                        try:
-                            Path(frame["local_path"]).unlink(missing_ok=True)
-                        except Exception:
-                            pass
-
-        except RuntimeError:
-            raise
-        except Exception as e:
-            logger.exception("FFmpeg 远程抽帧失败: %s", str(e))
-
-    # 封面生成失败
-    raise RuntimeError(f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。")
-
-
-# ── 公共入口 ────────────────────────────────────────────────────────────────
+    # ai_frame / ai_regenerate: 封面应由渲染后视频抽帧管道生成
+    # 如果调用方传入了持久化的封面 URL，直接使用
+    logger.warning(
+        "封面生成回退: plan_id=%s cover_type=%s — 统一管道应已生成封面，请检查 GenerationTask.cover_url",
+        plan_id,
+        cover_type,
+    )
+    raise RuntimeError(f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。")
 
 
 def run_ai_recommend(
diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py
index 7f3c75709..054c81640 100644
--- a/tests/unit/test_1294_preview_voice_injection.py
+++ b/tests/unit/test_1294_preview_voice_injection.py
@@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection:
 
             from packages.domain import EditingMode
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, _cover_url = _render_video(
                 task_id="test_task_123",
                 downloaded_videos=[Path("/tmp/video1.mp4")],
                 voice_path=None,
diff --git a/tests/unit/test_config_schemas_and_ai_endpoints.py b/tests/unit/test_config_schemas_and_ai_endpoints.py
index daf84e5eb..9c1ffbd5c 100644
--- a/tests/unit/test_config_schemas_and_ai_endpoints.py
+++ b/tests/unit/test_config_schemas_and_ai_endpoints.py
@@ -237,7 +237,7 @@ class TestAIRunTasks:
 
         from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover
 
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             run_generate_cover(
                 plan_id="plan-001",
                 asset_ids=["asset-1"],
diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py
index 3f462ec14..c8c2141bd 100644
--- a/tests/unit/test_cover_frame_pre_extract.py
+++ b/tests/unit/test_cover_frame_pre_extract.py
@@ -1,486 +1,149 @@
-"""Tests for cover frame pre-extraction during rendering.
+"""Tests for unified cover frame extraction pipeline.
 
-Tests:
-- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75%
-- extract_and_upload_cover_frames: extraction + OSS upload
-- RenderAdapterResult.cover_candidates field
-- generation_cover route uses pre-stored candidates
-- ai_service FFmpeg fallback
+统一封面管道测试：
+- extract_first_frame: 从已渲染视频抽取封面帧
+- 封面天然带标题（ASS 字幕已烧录到视频中）
 """
 
 from __future__ import annotations
 
+import tempfile
+import unittest
 from pathlib import Path
-from unittest.mock import MagicMock, Mock, call, patch
-
-import pytest
+from unittest.mock import patch
 
 
-class TestExtractCoverCandidates:
-    """extract_cover_candidates 测试."""
+class TestExtractFirstFrame(unittest.TestCase):
+    """extract_first_frame 单元测试."""
 
     @patch("video_processing.ffmpeg_utils.run_ffmpeg")
-    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
-    def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run):
-        """在 25%/50%/75% 处抽取 3 帧."""
-        import tempfile
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run):
+        """默认在视频 15% 处抽帧."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        # Mock run_ffmpeg 创建输出文件（ffmpeg 真实行为）
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        # Create temp files that look like they were created
-        def fake_run(cmd, **kwargs):
-            # Find the output path (last arg)
-            output_path = cmd[-1]
-            Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
-            return ("", "")
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name)
 
-        mock_run.side_effect = fake_run
-
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(b"fake video")
-            video_path = tmp.name
-
-        try:
-            results = extract_cover_candidates(video_path, num_frames=3)
-            assert len(results) == 3
-
-            # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0
-            assert results[0]["frame_time"] == 5.0
-            assert results[1]["frame_time"] == 10.0
-            assert results[2]["frame_time"] == 15.0
-
-            # Check local paths exist
-            for r in results:
-                assert Path(r["local_path"]).exists()
-
-            # Clean up
-            for r in results:
-                Path(r["local_path"]).unlink(missing_ok=True)
-        finally:
-            Path(video_path).unlink(missing_ok=True)
+            self.assertTrue(Path(result).exists())
+            # 验证 ffmpeg 被调用
+            mock_run.assert_called()
+            cmd = mock_run.call_args[0][0]
+            self.assertIn("-vframes", cmd)
+            self.assertIn("1", cmd)
+            Path(result).unlink(missing_ok=True)
 
     @patch("video_processing.ffmpeg_utils.run_ffmpeg")
-    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
-    def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run):
-        """FFmpeg 失败时跳过该帧，继续抽取其他帧."""
-        import tempfile
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_seek_ratio(self, mock_probe, mock_run):
+        """自定义抽帧位置."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        call_count = 0
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, seek_ratio=0.5)
 
-        def fake_run(cmd, **kwargs):
-            nonlocal call_count
-            call_count += 1
-            output_path = cmd[-1]
-            if call_count == 2:
-                # Second frame fails - don't create file
-                raise RuntimeError("ffmpeg error")
-            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-            return ("", "")
+            self.assertTrue(Path(result).exists())
+            # 50% of 10s = 5s
+            cmd = mock_run.call_args[0][0]
+            ss_idx = cmd.index("-ss") + 1
+            seek_val = cmd[ss_idx]
+            # Should be around 5 seconds
+            self.assertIn("05", seek_val)
+            Path(result).unlink(missing_ok=True)
 
-        mock_run.side_effect = fake_run
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_output_path_parameter(self, mock_probe, mock_run):
+        """指定输出路径."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(b"fake video")
-            video_path = tmp.name
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as out:
+                pass  # just get a path
 
-        try:
-            results = extract_cover_candidates(video_path, num_frames=3)
-            # Should get 2 frames (1st and 3rd), 2nd failed
-            assert len(results) == 2
-        finally:
-            Path(video_path).unlink(missing_ok=True)
-            for r in results:
-                Path(r["local_path"]).unlink(missing_ok=True)
+            # Create the file so ffmpeg "succeeds"
+            mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image")
 
-    @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed"))
-    def test_fallback_duration_when_probe_fails(self, mock_probe):
-        """probe 失败时使用默认时长."""
-        import tempfile
+            result = extract_first_frame(video.name, output_path=out.name)
+            self.assertEqual(result, out.name)
+            Path(out.name).unlink(missing_ok=True)
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_keeps_original_resolution_by_default(self, mock_probe, mock_run):
+        """默认保持原始分辨率（width=-1, height=-1）."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        # Mock run_ffmpeg to create output files
-        def fake_run(cmd, **kwargs):
-            output_path = cmd[-1]
-            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-            return ("", "")
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run):
-            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-                tmp.write(b"fake")
-                video_path = tmp.name
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name)
 
-            try:
-                results = extract_cover_candidates(video_path, num_frames=3)
-                assert len(results) == 3
-                # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75
-                assert results[0]["frame_time"] == 1.25
-                assert results[1]["frame_time"] == 2.5
-                assert results[2]["frame_time"] == 3.75
-            finally:
-                Path(video_path).unlink(missing_ok=True)
-                for r in results:
-                    Path(r["local_path"]).unlink(missing_ok=True)
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            # Should NOT have scale filter (only format)
+            self.assertNotIn("scale", vf_filter)
+            self.assertIn("format", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_width_triggers_scale(self, mock_probe, mock_run):
+        """指定宽度时添加 scale 滤镜."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, width=640)
+
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            self.assertIn("scale=640", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=RuntimeError("fail"))
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run):
+        """失败时清理临时文件."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+
+            with self.assertRaises(RuntimeError):
+                extract_first_frame(video.name)
 
 
-class TestExtractAndUploadCoverFrames:
-    """extract_and_upload_cover_frames 测试."""
+class TestRenderAdapterCoverUrl(unittest.TestCase):
+    """RenderAdapterResult.cover_url 字段测试."""
 
-    @patch("video_processing.oss_helpers.upload_to_oss")
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
-    def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload):
-        """上传帧到 OSS 并返回正确格式."""
-        import tempfile
+    def test_result_has_cover_url_field(self):
+        """RenderAdapterResult 包含 cover_url 字段."""
+        from video_processing.render_adapter import RenderAdapterResult
 
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+        result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg")
+        self.assertEqual(result.cover_url, "https://example.com/cover.jpg")
 
-        # Create actual temp files
-        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp1.close()
-        tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp2.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp2.close()
-
-        mock_extract.return_value = [
-            {"local_path": tmp1.name, "frame_time": 5.0},
-            {"local_path": tmp2.name, "frame_time": 10.0},
-        ]
-        mock_upload.side_effect = [
-            "https://oss.example.com/covers/plan1/frame_0.jpg",
-            "https://oss.example.com/covers/plan1/frame_1.jpg",
-        ]
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-
-        assert len(results) == 2
-        assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg"
-        assert results[0]["frame_time"] == 5.0
-        assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg"
-
-        assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg"
-        assert results[1]["frame_time"] == 10.0
-
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[])
-    def test_returns_empty_when_no_candidates(self, mock_extract):
-        """没有候选帧时返回空列表."""
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-        assert results == []
-
-    @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error"))
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
-    def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload):
-        """上传失败时跳过该帧."""
-        import tempfile
-
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
-
-        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp1.close()
-
-        mock_extract.return_value = [
-            {"local_path": tmp1.name, "frame_time": 5.0},
-        ]
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-        assert results == []
-
-
-class TestRenderAdapterResultCoverCandidates:
-    """RenderAdapterResult 的 cover_candidates 字段."""
-
-    def test_default_none(self):
-        """默认为 None."""
+    def test_result_cover_url_defaults_empty(self):
+        """cover_url 默认为空字符串."""
         from video_processing.render_adapter import RenderAdapterResult
 
         result = RenderAdapterResult(success=True)
-        assert result.cover_candidates is None
-
-    def test_can_set_candidates(self):
-        """可以设置候选帧列表."""
-        from video_processing.render_adapter import RenderAdapterResult
-
-        candidates = [
-            {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"},
-        ]
-        result = RenderAdapterResult(success=True, cover_candidates=candidates)
-        assert len(result.cover_candidates) == 1
-        assert result.cover_candidates[0]["frame_time"] == 5.0
+        self.assertEqual(result.cover_url, "")
 
 
-class TestAICoverServiceFFmpegFallback:
-    """AI 封面服务 FFmpeg 兜底测试."""
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head):
-        """FFmpeg 兜底抽帧成功."""
-        import tempfile
-
-        mock_head.return_value.status_code = 200
-
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp.close()
-
-        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}]
-
-        # Mock storage
-        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
-            mock_storage = Mock()
-            mock_storage.upload_file = Mock()
-            mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
-            mock_storage_fn.return_value = mock_storage
-
-            from packages.shared.ai_service import _call_ai_cover_service
-
-            result = _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-            assert result["type"] == "ai_frame"
-            assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
-            assert result["frame_time"] == 5.0
-            assert result["confidence"] == 0.85
-
-        Path(tmp.name).unlink(missing_ok=True)
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    def test_ffmpeg_no_video_url_raises(self, mock_head):
-        """没有视频 URL 时抛出 RuntimeError."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url=None,
-            )
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 抽帧为空时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-    def test_upload_type_returns_immediately(self):
-        """upload 类型直接返回."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        result = _call_ai_cover_service(
-            plan_id="plan1",
-            asset_ids=["a1"],
-            cover_type="upload",
-            primary_video_url="https://example.com/video.mp4",
-        )
-        assert result["type"] == "upload"
-
-    def test_manual_type_returns_immediately(self):
-        """manual 类型直接返回."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        result = _call_ai_cover_service(
-            plan_id="plan1",
-            asset_ids=["a1"],
-            cover_type="manual",
-            frame_time=5.0,
-            primary_video_url="https://example.com/video.mp4",
-        )
-        assert result["type"] == "manual"
-        assert result["frame_time"] == 5.0
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head):
-        """视频 URL 不可访问时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 404
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-
-class TestCoverTemplatesFix:
-    """CoverTemplateResponse config=None 修复测试."""
-
-    def test_config_none_becomes_empty_dict(self):
-        """config=None 时 CoverTemplateResponse 不报 ValidationError."""
-        from datetime import datetime
-
-        from app.schemas.cover_template import CoverTemplateResponse
-
-        # This should not raise
-        resp = CoverTemplateResponse(
-            id="1",
-            name="test",
-            thumbnail_url="",
-            is_system=True,
-            created_at=datetime.now(),
-            config={},
-        )
-        assert resp.config == {}
-
-
-class TestExtractFramesWithFFmpeg:
-    """_extract_frames_with_ffmpeg 单元测试."""
-
-    def test_extracts_frames_with_correct_seek_times(self):
-        """抽帧时间点正确计算."""
-        import subprocess
-        import tempfile
-
-        # Mock ffprobe to return duration
-        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="")
-        with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc:
-            # First call is ffprobe, rest are ffmpeg
-            call_count = 0
-
-            def side_effect(cmd, **kwargs):
-                nonlocal call_count
-                call_count += 1
-                if call_count == 1:
-                    # ffprobe call
-                    return mock_probe_result
-                else:
-                    # ffmpeg call - create output file
-                    output_path = cmd[-1]
-                    from pathlib import Path
-
-                    Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-                    return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
-
-            mock_subproc.side_effect = side_effect
-
-            from packages.shared.ai_service import _extract_frames_with_ffmpeg
-
-            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
-
-            assert len(results) == 3
-            # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0
-            assert results[0]["frame_time"] == 5.0
-            assert results[1]["frame_time"] == 10.0
-            assert results[2]["frame_time"] == 15.0
-
-            # Clean up
-            for r in results:
-                from pathlib import Path
-
-                Path(r["local_path"]).unlink(missing_ok=True)
-
-    def test_handles_ffmpeg_failure(self):
-        """FFmpeg 失败时跳过该帧."""
-        import subprocess
-        import tempfile
-        from pathlib import Path
-
-        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="")
-
-        call_count = 0
-
-        def side_effect(cmd, **kwargs):
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return mock_probe_result
-            output_path = cmd[-1]
-            if call_count == 2:
-                # First frame succeeds
-                Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-                return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
-            else:
-                # Other frames fail
-                raise subprocess.CalledProcessError(1, cmd)
-
-        with patch("subprocess.run", side_effect=side_effect):
-            from packages.shared.ai_service import _extract_frames_with_ffmpeg
-
-            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
-            assert len(results) == 1
-            Path(results[0]["local_path"]).unlink(missing_ok=True)
-
-
-class TestGenerationCoverPreStored:
-    """generation_cover.py 预存帧逻辑测试."""
-
-    def test_pre_stored_candidates_used_when_available(self):
-        """有预存帧时直接使用，不调用 AI 服务."""
-        from unittest.mock import patch
-
-        # Mock the dependencies
-        mock_plan = MagicMock()
-        mock_plan.config = {
-            "cover_candidates": [
-                {
-                    "image_url": "https://oss.example.com/covers/p1/frame_0.jpg",
-                    "frame_time": 5.0,
-                    "storage_key": "covers/p1/frame_0.jpg",
-                },
-                {
-                    "image_url": "https://oss.example.com/covers/p1/frame_1.jpg",
-                    "frame_time": 10.0,
-                    "storage_key": "covers/p1/frame_1.jpg",
-                },
-            ],
-            "rendered_storage_key": "rendered/p1/video.mp4",
-        }
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-        mock_body = MagicMock()
-        mock_body.asset_ids = ["a1"]
-        mock_body.cover_type = "ai_frame"
-        mock_body.frame_time = None
-
-        with (
-            patch("app.api.routes.generation_cover.get_editor_services") as mock_services,
-            patch("app.api.routes.generation_cover.get_db_session"),
-            patch("app.api.routes.generation_cover.get_current_user"),
-            patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"),
-            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-        ):
-
-            mock_services.return_value = (MagicMock(), mock_plan_svc)
-            mock_normalize.side_effect = lambda c: c
-
-            from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover
-
-            result = generate_cover(
-                body=mock_body,
-                template_id="t1",
-                plan_id="p1",
-                services=(MagicMock(), mock_plan_svc),
-                db=MagicMock(),
-                current_user=MagicMock(),
-            )
-
-            assert result.plan_id == "p1"
-            assert result.cover["type"] == "ai_frame"
-            assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg"
-            assert result.cover["frame_time"] == 5.0
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/unit/test_generated_video_creation_logic.py b/tests/unit/test_generated_video_creation_logic.py
index d715c4312..eab9b7133 100755
--- a/tests/unit/test_generated_video_creation_logic.py
+++ b/tests/unit/test_generated_video_creation_logic.py
@@ -310,9 +310,9 @@ class TestThumbnailInDedupHelpers:
         mock_dedup.VideoDeduplicator = MagicMock()
         sys.modules["video_processing.dedup"] = mock_dedup
 
-        # mock video_processing.thumbnail_generator
+        # mock video_processing.thumbnail_generator (统一封面管道: 仅保留 extract_first_frame)
         mock_thumb = MagicMock()
-        mock_thumb.generate_and_upload_thumbnail = MagicMock()
+        mock_thumb.extract_first_frame = MagicMock()
         sys.modules["video_processing.thumbnail_generator"] = mock_thumb
 
         # 关键：给 video_processing 包设置子模块属性，让 patch() 能通过属性访问找到
@@ -322,7 +322,7 @@ class TestThumbnailInDedupHelpers:
         video_processing.thumbnail_generator = mock_thumb
 
     def test_pre_generated_thumbnail_url_is_reused(self):
-        """传入 thumbnail_url 时直接复用，不调用 generate_and_upload_thumbnail。"""
+        """传入 thumbnail_url 时直接复用，统一封面管道不再自动生成缩略图。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -339,26 +339,23 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch("video_processing.thumbnail_generator.generate_and_upload_thumbnail") as mock_gen:
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-reuse",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                        thumbnail_url=pre_thumb_url,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-reuse",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                    thumbnail_url=pre_thumb_url,
+                )
 
             assert result == 1
-            # 预生成缩略图时不应调用 generate_and_upload_thumbnail
-            mock_gen.assert_not_called()
 
             from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
 
@@ -368,8 +365,8 @@ class TestThumbnailInDedupHelpers:
         finally:
             session.close()
 
-    def test_thumbnail_generated_when_not_provided(self):
-        """未传 thumbnail_url 时调用 generate_and_upload_thumbnail 生成。"""
+    def test_no_thumbnail_when_not_provided(self):
+        """未传 thumbnail_url 时不生成缩略图（统一封面管道已移除自动缩略图生成）。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -377,8 +374,6 @@ class TestThumbnailInDedupHelpers:
         Session = sessionmaker(bind=engine)
         session = Session()
 
-        generated_thumb_url = "https://oss.example.com/generated-thumb.jpg"
-
         try:
             with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls:
                 mock_dedup = mock_dedup_cls.return_value
@@ -386,43 +381,34 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch(
-                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-                    return_value=generated_thumb_url,
-                ) as mock_gen:
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-gen",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-gen",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                )
 
             assert result == 1
-            # 应调用一次缩略图生成
-            mock_gen.assert_called_once()
-            # 验证参数：video_path 和 storage_key
-            call_args = mock_gen.call_args
-            assert call_args[0][0] == "/tmp/fake.mp4"
-            assert "thumbnails" in call_args[0][1]
 
             from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
 
             video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-gen").first()
             assert video is not None
-            assert video.thumbnail_url == generated_thumb_url
+            # 统一封面管道下，不传 thumbnail_url 时不自动生成
+            assert not video.thumbnail_url
         finally:
             session.close()
 
-    def test_thumbnail_generation_failure_does_not_block(self):
-        """缩略图生成失败不影响主流程。"""
+    def test_no_thumbnail_does_not_block(self):
+        """统一封面管道下，缩略图不再在 dedup 阶段生成。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -437,24 +423,20 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch(
-                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-                    side_effect=RuntimeError("cv2 not found"),
-                ):
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-fail",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-fail",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                )
 
             assert result == 1  # 不阻断
 
diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py
index 1145d69bf..3edfd3ebd 100644
--- a/tests/unit/test_generation_cover.py
+++ b/tests/unit/test_generation_cover.py
@@ -92,3 +92,120 @@ def test_generation_cover_request_validation():
     req2 = GenerateCoverRequest(cover_type="upload", asset_ids=["a1", "a2"])
     assert req2.cover_type == "upload"
     assert req2.asset_ids == ["a1", "a2"]
+
+
+class TestUnifiedCoverPipelineEndpoint:
+    """测试统一封面管道在 generate_cover endpoint 中的逻辑 (lines 189-215)."""
+
+    def test_cover_url_from_generation_task(self):
+        """当 GenerationTask 有 cover_url 时，直接返回该 URL 作为封面。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse
+
+        # Mock plan with rendered_storage_key (so we skip the 3-step lookup)
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "generation_task_id": "task-123",
+            "rendered_storage_key": "rendered/plan-1/video.mp4",
+        }
+
+        # Mock plan_svc
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+        # Mock template_svc
+        mock_template_svc = MagicMock()
+
+        # Mock generation task with cover_url
+        mock_task = MagicMock()
+        mock_task.cover_url = "https://oss.example.com/rendered/plan-1/cover.jpg"
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_task
+            mock_repo_cls.return_value = mock_repo
+
+            # normalize_plan_config should return the config with cover
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/plan-1/cover.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-1",
+                plan_id="plan-1",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            # 验证返回的封面数据来自 GenerationTask.cover_url
+            assert result.plan_id == "plan-1"
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg"
+            assert result.cover["type"] == "ai_frame"
+
+    def test_cover_url_fallback_when_no_cover_url(self):
+        """当 GenerationTask 没有 cover_url 时，跳过统一管道走 run_generate_cover。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "generation_task_id": "task-456",
+            "rendered_storage_key": "rendered/plan-2/video.mp4",
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+        mock_template_svc = MagicMock()
+
+        # Task has no cover_url
+        mock_task = MagicMock()
+        mock_task.cover_url = ""
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_task
+            mock_repo_cls.return_value = mock_repo
+
+            # Mock storage service
+            mock_storage_svc = MagicMock()
+            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4"
+            mock_storage_getter.return_value = mock_storage_svc
+
+            mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"}
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-2",
+                plan_id="plan-2",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            # 应该走 run_generate_cover 回退
+            mock_run.assert_called_once()
+            assert result.cover["image_url"] == "https://fallback.com/cover.jpg"
diff --git a/tests/unit/test_mediakit_cover.py b/tests/unit/test_mediakit_cover.py
index 4b063e4f4..7c87bffbc 100755
--- a/tests/unit/test_mediakit_cover.py
+++ b/tests/unit/test_mediakit_cover.py
@@ -141,93 +141,13 @@ class TestMediaKitClient:
 
 
 class TestAICoverService:
-    """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。"""
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_with_ffmpeg_success(self, mock_ffmpeg, mock_head):
-        """FFmpeg 本地抽帧成功."""
-        import tempfile
-
-        mock_head.return_value.status_code = 200
-
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp.close()
-
-        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 3.5}]
-
-        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
-            mock_storage = Mock()
-            mock_storage.upload_file = Mock()
-            mock_storage.get_url.return_value = "https://example.com/frame.jpg"
-            mock_storage_fn.return_value = mock_storage
-
-            from packages.shared.ai_service import _call_ai_cover_service
-
-            result = _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-            assert result["type"] == "ai_frame"
-            assert result["image_url"] == "https://example.com/frame.jpg"
-            assert result["frame_time"] == 3.5
-            assert result["confidence"] == 0.85
-
-        Path(tmp.name).unlink(missing_ok=True)
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    def test_call_ai_cover_video_url_unreachable(self, mock_head):
-        """视频 URL 不可访问时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 404
+    """AI 封面服务测试（统一封面管道后）。"""
 
+    def test_call_ai_cover_ai_frame_raises(self):
+        """ai_frame type raises RuntimeError in unified pipeline."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/nonexistent.mp4",
-            )
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_url_double_slash_normalized(self, mock_ffmpeg, mock_head):
-        """URL 路径中的双斜杠应被规范化."""
-        dirty_url = "https://oss.example.com/generated/projects//tasks/abc123/rendered.mp4"
-        clean_url = "https://oss.example.com/generated/projects/tasks/abc123/rendered.mp4"
-
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError):
-            _call_ai_cover_service(
-                plan_id="plan-1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url=dirty_url,
-            )
-
-        # HEAD 请求使用规范化后的 URL
-        mock_head.assert_called_once()
-        assert mock_head.call_args[0][0] == clean_url
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_ffmpeg_failure_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 抽帧失败时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.side_effect = Exception("ffmpeg error")
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -235,11 +155,22 @@ class TestAICoverService:
                 primary_video_url="https://example.com/video.mp4",
             )
 
+    def test_call_ai_cover_ai_regenerate_raises(self):
+        """ai_regenerate type raises RuntimeError in unified pipeline."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_regenerate",
+            )
+
     def test_call_ai_cover_without_video_url_raises(self):
-        """没有视频 URL 时抛出 RuntimeError."""
+        """ai_frame without video URL still raises RuntimeError."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -276,23 +207,6 @@ class TestAICoverService:
         assert result["type"] == "manual"
         assert result["frame_time"] == 5.0
 
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_empty_frames_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 返回空帧列表时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
 
 class TestGenerateCover:
     """run_generate_cover 测试."""
diff --git a/tests/unit/test_render_adapter.py b/tests/unit/test_render_adapter.py
index c1ad355a1..39b76418a 100755
--- a/tests/unit/test_render_adapter.py
+++ b/tests/unit/test_render_adapter.py
@@ -348,24 +348,32 @@ class TestRenderPlan:
         mock_render_cls.return_value = mock_render
         mock_upload.return_value = "https://oss.example.com/out.mp4"
 
-        fake_thumb = "https://oss.example.com/rendered/plan_thumb/thumbnail.jpg"
-
         plan = FakePlan(id="plan_thumb")
         clips = [_make_clip("c1", order=0, duration=5.0)]
         asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
         adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
+        # Mock extract_first_frame to return a temp file path
+        import tempfile as _tf
+
+        _fake_frame = _tf.NamedTemporaryFile(suffix=".jpg", delete=False)
+        _fake_frame.write(b"fake frame")
+        _fake_frame.close()
         with patch(
-            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-            return_value=fake_thumb,
+            "video_processing.thumbnail_generator.extract_first_frame",
+            return_value=_fake_frame.name,
         ):
             result = adapter.render_plan(
                 "plan_thumb",
                 work_dir=tmp_path / "work",
             )
+        from pathlib import Path as _P
+
+        _P(_fake_frame.name).unlink(missing_ok=True)
 
         assert result.success
-        assert result.thumbnail_url == fake_thumb
+        # cover_url from upload_to_oss (mocked globally)
+        assert result.thumbnail_url == "https://oss.example.com/out.mp4"
 
     @patch("video_processing.render_adapter.upload_to_oss")
     @patch("video_processing.render_adapter.UnifiedRenderService")
@@ -397,8 +405,8 @@ class TestRenderPlan:
         adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         with patch(
-            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-            side_effect=RuntimeError("cv2 not available"),
+            "video_processing.thumbnail_generator.extract_first_frame",
+            side_effect=RuntimeError("ffmpeg not available"),
         ):
             result = adapter.render_plan(
                 "plan_thumb_fail",
diff --git a/tests/unit/test_shared_ai_service.py b/tests/unit/test_shared_ai_service.py
index f6b8ef35e..8cca936ce 100755
--- a/tests/unit/test_shared_ai_service.py
+++ b/tests/unit/test_shared_ai_service.py
@@ -436,12 +436,12 @@ class TestAiCoverService:
 
     def test_cover_type_ai_frame_raises_without_mediakit(self):
         """ai_frame mode raises RuntimeError when MediaKit is unavailable."""
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service("plan1", ["a1"], "ai_frame")
 
     def test_cover_type_ai_regenerate_raises_without_mediakit(self):
         """ai_regenerate mode raises RuntimeError when MediaKit is unavailable."""
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
 
     def test_cover_type_manual_still_works(self):