diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py
index 2ae55042e..1a6e5f90c 100755
--- a/packages/shared/ai_service.py
+++ b/packages/shared/ai_service.py
@@ -354,93 +354,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 +363,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,136 +397,16 @@ 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 是否可访问。")
-
-
-# ── 公共入口 ────────────────────────────────────────────────────────────────
-
-
-def run_ai_recommend(
-    plan_id: str,
-    template_id: str,
-    asset_ids: List[str],
-    editing_mode: str = "one_take",
-    target_duration: float = 30.0,
-    asset_analyses: Optional[Dict[str, str]] = None,
-) -> Dict[str, Any]:
-    """执行 AI 推荐片段方案
-
-    Args:
-        plan_id: 剪辑计划 ID
-        template_id: 模板 ID
-        asset_ids: 素材 ID 列表
-        editing_mode: 剪辑模式 (one_take / pip / voice_over / voice_pip)
-        target_duration: 目标时长（秒）
-        asset_analyses: 可选，{asset_id: 视频理解文本} 映射
-
-    Returns:
-        推荐方案 dict，包含 clips / config / total_duration / confidence
-    """
-    logger.info(
-        "AI 推荐片段方案: plan_id=%s template_id=%s assets=%d mode=%s duration=%.1f has_analysis=%s",
+    # ai_frame / ai_regenerate: 封面应由渲染后视频抽帧管道生成
+    # 如果调用方传入了持久化的封面 URL，直接使用
+    logger.warning(
+        "封面生成回退: plan_id=%s cover_type=%s — 统一管道应已生成封面，请检查 GenerationTask.cover_url",
         plan_id,
-        template_id,
-        len(asset_ids),
-        editing_mode,
-        target_duration,
-        bool(asset_analyses),
+        cover_type,
     )
-    result = _call_ai_recommend_service(
-        plan_id=plan_id,
-        template_id=template_id,
-        asset_ids=asset_ids,
-        editing_mode=editing_mode,
-        target_duration=target_duration,
-        asset_analyses=asset_analyses,
+    raise RuntimeError(
+        f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。"
     )
-    logger.info(
-        "AI 推荐完成: plan_id=%s clips=%d duration=%.1f confidence=%.2f",
-        plan_id,
-        len(result["clips"]),
-        result["total_duration"],
-        result["confidence"],
-    )
-    return result
 
 
 def run_generate_cover(