diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py
index 73130bb73..4b904cfc8 100644
--- a/apps/api/app/api/routes/generation_cover.py
+++ b/apps/api/app/api/routes/generation_cover.py
@@ -65,6 +65,45 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
+
+def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
+    """下载 MediaKit 返回的临时帧图并转存到 OSS covers/ 路径。"""
+    import tempfile
+    import uuid
+    from pathlib import Path
+
+    tmp_path: str | None = None
+    try:
+        import httpx
+
+        resp = httpx.get(frame_url, timeout=30, follow_redirects=True)
+        resp.raise_for_status()
+        if not resp.content:
+            return frame_url
+
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
+            tmp.write(resp.content)
+            tmp_path = tmp.name
+
+        from packages.shared.storage import get_shared_storage_service
+
+        storage = get_shared_storage_service()
+        cover_key = f"covers/{plan_id}/cover_{uuid.uuid4().hex[:8]}.jpg"
+        storage.upload_file(
+            file_or_path=tmp_path,
+            storage_key=cover_key,
+            content_type="image/jpeg",
+        )
+        public_url = storage.get_url(cover_key)
+        return public_url or frame_url
+    except Exception:
+        logger.warning("封面帧转存失败，返回原始 URL: plan_id=%s", plan_id, exc_info=True)
+        return frame_url
+    finally:
+        if tmp_path:
+            Path(tmp_path).unlink(missing_ok=True)
+
+
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
 def generate_cover(
     body: GenerateCoverRequest,
@@ -198,44 +237,30 @@ def generate_cover(
                 exc_info=True,
             )
 
-    # 仍然找不到才报 400
-    if not rendered_storage_key:
-        logger.error("[封面生成] ❌ 找不到预览视频: plan_id=%s", plan_id)
-        raise HTTPException(
-            status_code=400,
-            detail="请先生成预览视频，再生成封面",
-        )
-
-    # 回写到 plan.config
-    plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
-
-    # 使用裸 URL（rendered/* 已配置公开读）
+    # 使用裸 URL（rendered/* 已配置公开读）；找不到渲染视频时不立即报错，
+    # 因为步骤 E 可以直接从源素材抽帧（前端纯 Canvas/WebCodecs 预览无后端渲染产物）
     primary_video_url = None
-    try:
-        if rendered_storage_key.startswith("http"):
-            primary_video_url = rendered_storage_key
-        else:
-            from packages.shared.storage import get_shared_storage_service
+    if rendered_storage_key:
+        plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
+        try:
+            if rendered_storage_key.startswith("http"):
+                primary_video_url = rendered_storage_key
+            else:
+                from packages.shared.storage import get_shared_storage_service
 
-            storage_svc = get_shared_storage_service()
-            primary_video_url = storage_svc.get_url(rendered_storage_key)
-        # 防御性规范化：合并路径中的双斜杠（// -> /），但保留协议头的 ://
-        # 历史数据中 project_id 为空时会产生 projects//tasks/ 路径，
-        # MediaKit 的 HTTP 客户端会规范化 URL 导致 404
-        if primary_video_url:
-            import re as _re
-
-            primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
-        logger.info(
-            "获取预览视频URL用于封面生成: plan_id=%s url=%s",
-            plan_id,
-            primary_video_url[:80] if primary_video_url else "",
-        )
-    except Exception as e:
-        raise HTTPException(
-            status_code=500,
-            detail=f"获取预览视频URL失败: {e}",
-        ) from e
+                storage_svc = get_shared_storage_service()
+                primary_video_url = storage_svc.get_url(rendered_storage_key)
+            if primary_video_url:
+                import re as _re
+                primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
+            logger.info(
+                "获取预览视频URL用于封面生成: plan_id=%s url=%s",
+                plan_id,
+                primary_video_url[:80] if primary_video_url else "",
+            )
+        except Exception as e:
+            logger.warning("获取预览视频URL失败: plan_id=%s err=%s", plan_id, e)
+            primary_video_url = None
 
     # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
     # 多步查找 cover_url，和查找视频 URL 一样的 fallback 逻辑
@@ -326,6 +351,93 @@ def generate_cover(
                             cover_url_from_task[:80],
                         )
 
+        # 步骤 E1：如果有已渲染的预览视频 URL 但 cover_url 未持久化（历史数据），
+        # 直接从渲染视频抽帧
+        if not cover_url_from_task and primary_video_url:
+            try:
+                from packages.shared.mediakit_client import get_mediakit_client
+
+                mk_client = get_mediakit_client()
+                if mk_client.is_available:
+                    logger.info(
+                        "[封面生成] 步骤E1-从渲染视频抽帧: plan_id=%s url=%s",
+                        plan_id,
+                        primary_video_url[:80],
+                    )
+                    snapshots = mk_client.extract_frames(
+                        video_url=primary_video_url,
+                        strategy="SpecifiedFrames",
+                        max_frames=1,
+                    )
+                    if snapshots:
+                        raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+                        if raw:
+                            cover_url_from_task = _persist_cover_frame(raw, plan_id)
+                            logger.info(
+                                "[封面生成] 统一管道封面(步骤E1-rendered-video): plan_id=%s url=%s",
+                                plan_id,
+                                cover_url_from_task[:80],
+                            )
+            except Exception:
+                logger.warning(
+                    "[封面生成] 步骤E1从渲染视频抽帧失败: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
+                )
+
+        # 步骤 E2：前端纯 Canvas/WebCodecs 预览（无后端渲染任务）时，
+        # 直接从用户选择的第一个视频素材中抽取封面帧。
+        if not cover_url_from_task and body.asset_ids:
+            from packages.adapters.sqlalchemy_impl.asset_repository import (
+                SQLAlchemyAssetRepository,
+            )
+            from packages.shared.mediakit_client import get_mediakit_client
+            from packages.shared.storage import get_shared_storage_service
+
+            asset_repo = SQLAlchemyAssetRepository(db)
+            storage_svc = get_shared_storage_service()
+            mk_client = get_mediakit_client()
+            if mk_client.is_available:
+                for aid in body.asset_ids:
+                    try:
+                        asset = asset_repo.get(aid)
+                        if not asset or asset.file_type != "video":
+                            continue
+                        sk = asset.storage_key or ""
+                        if not sk:
+                            continue
+                        src_url = sk if sk.startswith("http") else storage_svc.get_url(sk)
+                        if not src_url:
+                            continue
+                        logger.info(
+                            "[封面生成] 步骤E-从素材抽帧: plan_id=%s asset_id=%s url=%s",
+                            plan_id,
+                            aid,
+                            src_url[:80],
+                        )
+                        snapshots = mk_client.extract_frames(
+                            video_url=src_url,
+                            strategy="SpecifiedFrames",
+                            max_frames=1,
+                        )
+                        if snapshots:
+                            raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+                            if raw:
+                                cover_url_from_task = _persist_cover_frame(raw, plan_id)
+                                logger.info(
+                                    "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s",
+                                    plan_id,
+                                    cover_url_from_task[:80],
+                                )
+                                break
+                    except Exception:
+                        logger.warning(
+                            "[封面生成] 步骤E从素材抽帧失败: plan_id=%s asset_id=%s",
+                            plan_id,
+                            aid,
+                            exc_info=True,
+                        )
+
         if cover_url_from_task:
             # 标题已在预览视频渲染时烧录（ASS字幕），封面帧自然包含标题
             cover_data = {
@@ -347,7 +459,7 @@ def generate_cover(
         # ai_frame/ai_regenerate 类型必须从渲染管道获取，不再回退到 AI 服务
         raise HTTPException(
             status_code=400,
-            detail="封面尚未生成，请先重新生成预览视频以触发封面自动提取",
+            detail="封面生成失败：未找到可抽帧的视频素材，请确认已上传视频素材后重试",
         )
 
     from packages.shared.ai_service import run_generate_cover
diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py
index 689878510..dc6a05747 100644
--- a/tests/unit/test_generation_cover.py
+++ b/tests/unit/test_generation_cover.py
@@ -182,6 +182,7 @@ class TestUnifiedCoverPipelineEndpoint:
         with (
             patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
             patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
         ):
             mock_repo = MagicMock()
             mock_repo.get.return_value = mock_task
@@ -194,6 +195,13 @@ class TestUnifiedCoverPipelineEndpoint:
             mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4"
             mock_storage_getter.return_value = mock_storage_svc
 
+            # MediaKit 抽帧也返回 None，模拟最终失败
+            mock_mk = MagicMock()
+            mock_mk.is_available = True
+            mock_mk.extract_frames.return_value = None
+            mock_mk_getter.return_value = mock_mk
+
+            # body 不传 asset_ids，步骤 E2 不会进入
             from app.api.routes.generation_cover import generate_cover
 
             with pytest.raises(HTTPException) as exc_info:
@@ -207,7 +215,6 @@ class TestUnifiedCoverPipelineEndpoint:
                 )
 
             assert exc_info.value.status_code == 400
-            assert "封面尚未生成" in exc_info.value.detail
 
     def test_cover_url_found_via_source_edit_plan(self):
         """步骤B：通过 source_edit_plan_id 找到预览任务的 cover_url。"""
@@ -838,3 +845,138 @@ class TestUploadCoverType:
             assert result.cover["image_url"] == "https://oss.example.com/uploaded/my-cover.png"
             # 验证没有调用任何预览视频查找逻辑
             # (normalize_plan_config 是唯一被调用的外部函数)
+
+    def test_cover_extracted_from_source_asset_when_no_preview(self):
+        """步骤E2：无后端渲染产物时，直接从用户选择的视频素材抽帧。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}  # 无 rendered_storage_key，无 generation_task_id
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+        mock_db = MagicMock()
+
+        # 模拟视频素材
+        mock_asset = MagicMock()
+        mock_asset.file_type = "video"
+        mock_asset.storage_key = "uploads/source-clip.mp4"
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = mock_asset
+
+        mock_mk = MagicMock()
+        mock_mk.is_available = True
+        mock_mk.extract_frames.return_value = [{"image_url": "https://mediakit.internal/frame-abc.jpg"}]
+
+        mock_storage = MagicMock()
+        mock_storage.get_url.return_value = "https://oss.example.com/uploads/source-clip.mp4"
+
+        body = GenerateCoverRequest(
+            cover_type="ai_frame",
+            asset_ids=["asset-video-1"],
+        )
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
+            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
+            patch(
+                "app.api.routes.generation_cover._persist_cover_frame",
+                return_value="https://oss.example.com/covers/final.jpg",
+            ),
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/final.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="tpl-source",
+                plan_id="plan-source",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/covers/final.jpg"
+            mock_mk.extract_frames.assert_called_once()
+            # 确保用的是源素材 URL
+            call_kwargs = mock_mk.extract_frames.call_args.kwargs
+            assert "source-clip.mp4" in call_kwargs["video_url"]
+
+    def test_step_e_skips_non_video_assets(self):
+        """步骤E2：asset_ids 里只有图片素材时，不调用 MediaKit 并返回 400。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+        from fastapi import HTTPException
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+        mock_db = MagicMock()
+
+        mock_image_asset = MagicMock()
+        mock_image_asset.file_type = "image"
+        mock_image_asset.storage_key = "uploads/photo.png"
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = mock_image_asset
+
+        mock_mk = MagicMock()
+        mock_mk.is_available = True
+
+        body = GenerateCoverRequest(
+            cover_type="ai_frame",
+            asset_ids=["asset-img-1"],
+        )
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+            mock_storage_getter.return_value = MagicMock()
+
+            from app.api.routes.generation_cover import generate_cover
+
+            with pytest.raises(HTTPException) as exc_info:
+                generate_cover(
+                    body=body,
+                    template_id="tpl-img",
+                    plan_id="plan-img",
+                    services=(mock_template_svc, mock_plan_svc),
+                    db=mock_db,
+                    current_user=MagicMock(),
+                )
+
+            assert exc_info.value.status_code == 400
+            mock_mk.extract_frames.assert_not_called()