fix(edit-flow): 批量生成时 plan.config 只回写一次，避免 generation_task_id 被循环覆盖

PR #1464 AI Review 发现的阻塞问题：count > 1 批量生成时，循环内对同一 source_edit_plan_id 多次调用 _writeback_edit_plan_config，每次都会覆盖 config.generation_task_id，最终只保留最后一个 task id。修复：仅在首个成功任务（len(created_tasks)==1）时回写。
feat(generate): 编辑流程自动保存草稿+创建任务参数修复+轮询改新接口+清理幽灵API (#1465 )
2026-08-23 14:11:42 +08:00 · 2026-08-23 13:56:32 +08:00 · 2026-08-23 13:55:42 +08:00 · 2026-08-23 13:53:36 +08:00 · 2026-08-23 05:39:27 +00:00 · 2026-08-23 13:34:57 +08:00
53 changed files with 1401 additions and 325 deletions
@@ -0,0 +1,26 @@
+"""Add title_config to generation_tasks
+
+Revision ID: 057_title_config
+Revises: 056_fix_cover_templates_config
+Create Date: 2026-08-23
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "057_title_config"
+down_revision = "056_fix_cover_templates_config"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("title_config", sa.JSON(), nullable=False, server_default="{}"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generation_tasks", "title_config")
@@ -63,8 +63,15 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────


-def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
-    """下载 MediaKit 返回的临时帧图并转存到 OSS covers/ 路径。"""
+def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> str:
+    """下载 MediaKit 返回的临时帧图，可选叠加标题后转存到 OSS covers/ 路径。
+
+    Args:
+        frame_url: MediaKit 返回的临时帧图 URL
+        plan_id: 剪辑计划 ID（生成 OSS key）
+        title_text: 非空时用 Pillow 在帧上叠加标题（用于 E2 从源素材抽帧，
+            因为源素材本身没有烧录标题）
+    """
    import tempfile
    import uuid
    from pathlib import Path
@@ -82,6 +89,21 @@ def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
            tmp.write(resp.content)
            tmp_path = tmp.name

+        # E2 从源素材抽帧时，源素材无标题，叠加标题文字
+        if title_text and title_text.strip():
+            try:
+                from packages.shared.title_overlay import apply_title_to_image
+
+                applied = apply_title_to_image(tmp_path, title_text)
+                if applied:
+                    logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id)
+            except Exception:
+                logger.warning(
+                    "[封面生成] E2 标题叠加失败（返回无标题帧）: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
+                )
+
        from packages.shared.storage import get_shared_storage_service

        storage = get_shared_storage_service()
@@ -396,6 +418,11 @@ def generate_cover(
            asset_repo = SQLAlchemyAssetRepository(db)
            storage_svc = get_shared_storage_service()
            mk_client = get_mediakit_client()
+            # 从 plan.config 读取标题，E2 从源素材抽帧时叠加（源素材本身无标题）
+            _e2_title_cfg = (plan.config or {}).get("title", {}) or {}
+            if not isinstance(_e2_title_cfg, dict):
+                _e2_title_cfg = {}
+            _e2_title_text = (_e2_title_cfg.get("text", "") or "").strip() if _e2_title_cfg.get("enabled", True) else ""
            if mk_client.is_available:
                for aid in body.asset_ids:
                    try:
@@ -425,7 +452,7 @@ def generate_cover(
                        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)
+                                cover_url_from_task = _persist_cover_frame(raw, plan_id, title_text=_e2_title_text)
                                logger.info(
                                    "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s",
                                    plan_id,
@@ -16,6 +16,7 @@ from app.core.task_enqueue import (
 from app.dependencies import (
    get_asset_library_repository,
    get_asset_repository,
+    get_db_session,
    get_generated_video_repository,
    get_generation_task_repository,
    get_project_repository,
@@ -32,6 +33,7 @@ from app.schemas.generation_task import (
    ListGenerationTasksResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session

 from packages.application import (
    CreateGenerationTaskCommand,
@@ -69,6 +71,7 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
        output_height=getattr(task, "output_height", 720),
        cover_url=getattr(task, "cover_url", ""),
        custom_title=getattr(task, "custom_title", ""),
+        title_config=getattr(task, "title_config", {}) or {},
        logs=getattr(task, "logs", "[]"),
        status=task.status,
        progress=task.progress,
@@ -142,6 +145,54 @@ def _select_assets_from_library(
    return [a.id for a in ready_video_assets]


+
+def _writeback_edit_plan_config(
+    plan_id: str,
+    task_id: str,
+    title_config: dict | None,
+    db: Session,
+) -> None:
+    """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。
+
+    用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
+    失败只记日志，不影响任务创建。
+    """
+    if not plan_id:
+        return
+    try:
+        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+        plan_model = db.query(EditPlanModel).filter(EditPlanModel.id == plan_id).first()
+        if plan_model is None:
+            logger.warning("[生成任务] 回写plan.config失败: plan不存在 plan_id=%s", plan_id)
+            return
+
+        current_config = plan_model.config if isinstance(plan_model.config, dict) else {}
+        merged = dict(current_config)
+        merged["generation_task_id"] = task_id
+        if title_config:
+            merged["title_config"] = title_config
+        plan_model.config = merged
+        db.commit()
+        logger.info(
+            "[生成任务] 回写plan.config成功: plan_id=%s task_id=%s keys=%s",
+            plan_id,
+            task_id,
+            list(merged.keys()),
+        )
+    except Exception as e:
+        logger.warning(
+            "[生成任务] 回写plan.config异常（不影响任务创建）: plan_id=%s error=%s",
+            plan_id,
+            e,
+            exc_info=True,
+        )
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+
 def _resolve_project_and_library(
    request: CreateGenerationTaskRequest,
    project_repository: Any,
@@ -187,6 +238,7 @@ def create_generation_task(
    project_repository: Any = Depends(get_project_repository),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    asset_repository: Any = Depends(get_asset_repository),
+    db: Session = Depends(get_db_session),
 ) -> BatchGenerationTaskResponse:
    logger.info(
        "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, count=%d",
@@ -304,6 +356,7 @@ def create_generation_task(
                    output_height=request.output_height,
                    cover_url=request.cover_url,
                    custom_title=request.custom_title,
+                    title_config=request.title_config or {},
                )
            )
            try:
@@ -315,6 +368,15 @@ def create_generation_task(
                    log_task_status=True,
                ):
                    created_tasks.append(task)
+                    # 只在首个成功任务时回写一次 plan.config，
+                    # 避免批量生成时循环覆盖 generation_task_id
+                    if request.source_edit_plan_id and len(created_tasks) == 1:
+                        _writeback_edit_plan_config(
+                            plan_id=request.source_edit_plan_id,
+                            task_id=task.id,
+                            title_config=request.title_config,
+                            db=db,
+                        )
                else:
                    failed_tasks.append(task)
            except UserPendingLimitExceeded as _e:
@@ -56,6 +56,7 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=["Template Editor"])


+# DEPRECATED: 前端已改用 /generation/tasks 体系，此路由保留仅供旧版兼容，计划下线
@router.post("/generate", response_model=EditPlanGenerateResponse)
 def generate_editor_draft(
    template_id: str,
@@ -285,6 +286,7 @@ def _get_task_output_url(task, gen_task_repo, db) -> str:
    return ""


+# DEPRECATED: 前端已改用 /generation/tasks 体系，此路由保留仅供旧版兼容，计划下线
@router.get("/generation-status", response_model=EditPlanGenerationStatusResponse)
 def get_editor_generation_status(
    template_id: str,
@@ -33,6 +33,11 @@ class CreateGenerationTaskRequest(BaseModel):
    voice_ids: list[str] = Field(default_factory=list)
    # ── 来源剪辑计划 ──
    source_edit_plan_id: str = ""
+    # ── 标题配置（结构化，优先于 custom_title 纯文本）──
+    title_config: dict | None = Field(
+        default=None,
+        description="标题样式对象，包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。",
+    )
    # ── 视频标题 ──
    video_title: str = Field(default="", description="生成视频的标题/名称，为空则使用默认命名")
    # ── 批量生成 ──
@@ -109,6 +114,7 @@ class GenerationTaskResponse(BaseModel):
    output_height: int = 720
    cover_url: str = ""
    custom_title: str = ""
+    title_config: dict = Field(default_factory=dict)
    status: str
    progress: float
    result_count: int
@@ -4,11 +4,20 @@
 import apiClient from "../client"
 import type { BgmPreset, BgmPresetsQuery } from "./types"

-/** 获取 BGM 预设列表 */
-export const getBgmPresets = async (params?: BgmPresetsQuery): Promise<BgmPreset[]> => {
+/**
+ * 获取 BGM 预设列表
+ * @param templateId 模板/草稿 ID
+ * @param params 分类/关键词筛选
+ */
+export const getBgmPresets = async (
+  templateId: string,
+  params?: BgmPresetsQuery,
+): Promise<BgmPreset[]> => {
  const searchParams: Record<string, string> = {}
  if (params?.category) searchParams.category = params.category
  if (params?.keyword) searchParams.keyword = params.keyword
-  const res = await apiClient.get("/bgm/presets", { params: searchParams })
+  const res = await apiClient.get(`/templates/${templateId}/editor/bgm/presets`, {
+    params: searchParams,
+  })
  return res.data?.data ?? res.data ?? []
 }
@@ -1,9 +1,22 @@
 import apiClient from "../client"

+export interface GenerateCoverTitleConfig {
+  text?: string
+  font?: string
+  font_size?: number
+  font_color?: string
+  position?: string
+  bold?: boolean
+  stroke?: boolean
+  shadow?: boolean
+}
+
 export interface GenerateCoverRequest {
  asset_ids: string[]
  cover_type?: "ai_frame" | "manual" | "upload" | "ai_regenerate"
  frame_time?: number
+  /** 标题样式，用于在封面上叠加标题文字 */
+  title_config?: GenerateCoverTitleConfig
 }

 export interface GenerateCoverResponse {
@@ -6,6 +6,7 @@ import apiClient from "../client"
 import type {
  CreateGenerationTaskRequest,
  CreateGenerationTaskResponse,
+  GenerationTaskDetail,
  TaskItem,
  TaskListParams,
  TaskListResponse,
@@ -19,6 +20,12 @@ export const createGenerationTask = async (
  return data
 }

+/** 获取单个生成任务详情（轮询用） */
+export const getGenerationTask = async (taskId: string): Promise<GenerationTaskDetail> => {
+  const { data } = await apiClient.get<GenerationTaskDetail>(`/generation/tasks/${taskId}`)
+  return data
+}
+
 /** 获取任务列表（支持分页和筛选） */
 export const getTasks = async (params?: TaskListParams): Promise<TaskListResponse> => {
  const { data } = await apiClient.get<TaskListResponse>("/tasks", {
@@ -82,10 +82,12 @@ export interface CreateGenerationTaskRequest {
    stroke?: boolean
    shadow?: boolean
  }
+  /** 关联的草稿 ID（编辑流程数据链路用） */
+  source_edit_plan_id?: string
 }

-/** 创建生成任务响应（对齐后端 GenerationTaskResponse） */
-export interface CreateGenerationTaskResponse {
+/** 单个生成任务详情（对齐后端 GenerationTaskResponse） */
+export interface GenerationTaskDetail {
  id: string
  project_id: string
  asset_library_id: string
@@ -95,8 +97,18 @@ export interface CreateGenerationTaskResponse {
  asset_ids: string[]
  title_ids: string[]
  voice_ids: string[]
+  source_edit_plan_id?: string
  status: string
  progress: number
  result_count: number
  error_message: string
+  error_info?: TaskErrorInfo
+  created_at?: string | null
+  updated_at?: string | null
+}
+
+/** 创建生成任务响应（后端返回批量结构 {items, total}） */
+export interface CreateGenerationTaskResponse {
+  items: GenerationTaskDetail[]
+  total: number
 }
@@ -4,51 +4,29 @@
 import apiClient from "../client"
 import type {
  EditPlan,
-  EditPlanListParams,
-  EditPlanListResponse,
-  CreateEditPlanRequest,
  UpdateEditPlanRequest,
  GenerateResponse,
  GenerationStatusResponse,
  EditPlanGeneration,
  GeneratedVideo,
-  CopyEditPlanRequest,
 } from "./types"

-/** 获取模板草稿列表（支持分页和筛选） */
-export async function getEditPlans(params?: EditPlanListParams): Promise<EditPlanListResponse> {
-  const response = await apiClient.get<EditPlanListResponse>("/templates/drafts", {
-    params,
-  })
-  return response.data
-}
-
 /** 获取单个模板草稿 */
 export async function getEditPlan(templateId: string): Promise<EditPlan> {
  const response = await apiClient.get(`/templates/${templateId}/editor`)
  return response.data
 }

-/** 创建模板草稿 */
-export async function createEditPlan(data: CreateEditPlanRequest): Promise<EditPlan> {
-  const response = await apiClient.post("/templates/drafts", data)
-  return response.data
-}
-
-/** 更新模板草稿 */
+/** 更新模板草稿（支持传入 AbortSignal 用于自动保存竞态取消） */
 export async function updateEditPlan(
  templateId: string,
  data: UpdateEditPlanRequest,
+  signal?: AbortSignal,
 ): Promise<EditPlan> {
-  const response = await apiClient.put(`/templates/${templateId}/editor`, data)
+  const response = await apiClient.put(`/templates/${templateId}/editor`, data, { signal })
  return response.data
 }

-/** 删除模板草稿 */
-export async function deleteEditPlan(templateId: string): Promise<void> {
-  await apiClient.delete(`/templates/${templateId}/editor`)
-}
-
 /** 触发生成 */
 export async function generateEditPlan(templateId: string): Promise<GenerateResponse> {
  const response = await apiClient.post(`/templates/${templateId}/editor/generate`)
@@ -72,20 +50,3 @@ export async function getGenerationTaskResults(taskId: string): Promise<Generate
  const response = await apiClient.get(`/generation/tasks/${taskId}/results`)
  return response.data.items || response.data || []
 }
-
-/** 取消生成任务 */
-export async function cancelGeneration(templateId: string): Promise<void> {
-  await apiClient.post(`/templates/${templateId}/editor/cancel`)
-}
-
-/** 复制模板草稿（含所有片段配置） */
-export async function copyEditPlan(
-  templateId: string,
-  data?: CopyEditPlanRequest,
-): Promise<EditPlan> {
-  const response = await apiClient.post<EditPlan>(
-    `/templates/${templateId}/editor/copy`,
-    data || {},
-  )
-  return response.data
-}
@@ -15,10 +15,7 @@ export type {
  EditPlanSegment,
  EditPlanConfig,
  EditPlan,
-  CreateEditPlanRequest,
  UpdateEditPlanRequest,
-  EditPlanListParams,
-  EditPlanListResponse,
  GenerateResponse,
  EditPlanGeneration,
  ClipStatusItem,
@@ -37,7 +34,6 @@ export type {
  ClipReorderResponse,
  ClipBatchDeleteResponse,
  ClipsFromAssetsResponse,
-  CopyEditPlanRequest,
  TransitionEffect,
  MediaAsset,
 } from "./types"
@@ -53,17 +49,12 @@ export {

 // 模板草稿 CRUD + 生成
 export {
-  getEditPlans,
  getEditPlan,
-  createEditPlan,
  updateEditPlan,
-  deleteEditPlan,
  generateEditPlan,
  getGenerationStatus,
  getEditPlanGenerations,
  getGenerationTaskResults,
-  cancelGeneration,
-  copyEditPlan,
 } from "./editPlans"

 // 片段 CRUD + 批量操作
@@ -118,6 +118,17 @@ export interface EditPlanConfig {
  generate_count?: number
  /** 素材模式 */
  material_mode?: string
+  /** 前端标题设置（Step4 自动保存，与 title_config 字段分离，不影响后端渲染） */
+  title?: {
+    text?: string
+    font?: string
+    font_size?: number
+    color?: string
+    position?: string
+    bold?: boolean
+    stroke?: boolean
+    shadow?: boolean
+  }
  /** 预览视频 URL（封面生成用） */
  rendered_storage_key?: string
  /** 生成任务 ID */
@@ -9,8 +9,6 @@ export type {
  TemplateSegment,
  TemplateListParams,
  TemplateListResponse,
-  GenerateFromTemplateRequest,
-  GenerateFromTemplateResponse,
  CopyTemplateResponse,
 } from "./types"

@@ -24,5 +22,4 @@ export {
  getTemplate,
  toggleFavoriteTemplate,
  copyTemplate,
-  generateFromTemplate,
 } from "./templates"
@@ -5,8 +5,6 @@
 import apiClient from "../client"
 import type {
  CopyTemplateResponse,
-  GenerateFromTemplateRequest,
-  GenerateFromTemplateResponse,
  TemplateItem,
  TemplateListParams,
  TemplateListResponse,
@@ -45,15 +43,3 @@ export const copyTemplate = async (templateId: string): Promise<CopyTemplateResp
  const response = await apiClient.post<CopyTemplateResponse>(`/templates/${templateId}/copy`)
  return response.data
 }
-
-/** 从模板生成 */
-export const generateFromTemplate = async (
-  templateId: string,
-  data?: GenerateFromTemplateRequest,
-): Promise<GenerateFromTemplateResponse> => {
-  const response = await apiClient.post<GenerateFromTemplateResponse>(
-    `/templates/${templateId}/generate`,
-    data,
-  )
-  return response.data
-}
@@ -16,9 +16,17 @@ interface BgmSelectorProps {
  onClose: () => void
  config: BgmMixConfig
  onChange: (config: BgmMixConfig) => void
+  /** 模板/草稿 ID，用于请求 BGM 预设 */
+  templateId?: string
 }

-const BgmSelector: React.FC<BgmSelectorProps> = ({ open, onClose, config, onChange }) => {
+const BgmSelector: React.FC<BgmSelectorProps> = ({
+  open,
+  onClose,
+  config,
+  onChange,
+  templateId,
+}) => {
  const {
    presets,
    loading,
@@ -30,7 +38,7 @@ const BgmSelector: React.FC<BgmSelectorProps> = ({ open, onClose, config, onChan
    loadPresets,
    handlePreview,
    stopPreview,
-  } = useBgmSelector(open)
+  } = useBgmSelector(open, templateId)

  /* ── 选中 BGM ── */
  const handleSelect = useCallback(
@@ -19,7 +19,7 @@ export const CATEGORY_LIST: {
 * BGM 选择器数据与交互 Hook
 * 封装列表加载、搜索、分类筛选、试听播放逻辑
 */
-export function useBgmSelector(open: boolean) {
+export function useBgmSelector(open: boolean, templateId?: string) {
  const [presets, setPresets] = useState<BgmPreset[]>([])
  const [loading, setLoading] = useState(false)
  const [activeCategory, setActiveCategory] = useState<BgmCategory | "all">("all")
@@ -30,19 +30,23 @@ export function useBgmSelector(open: boolean) {

  /* ── 加载 BGM 列表 ── */
  const loadPresets = useCallback(async () => {
+    if (!templateId) {
+      setPresets([])
+      return
+    }
    setLoading(true)
    try {
      const params: { category?: string; keyword?: string } = {}
      if (activeCategory !== "all") params.category = activeCategory
      if (keyword.trim()) params.keyword = keyword.trim()
-      const data = await getBgmPresets(params)
+      const data = await getBgmPresets(templateId, params)
      setPresets(data)
    } catch {
      message.error("加载 BGM 列表失败")
    } finally {
      setLoading(false)
    }
-  }, [activeCategory, keyword])
+  }, [activeCategory, keyword, templateId])

  useEffect(() => {
    if (open) loadPresets()
@@ -171,6 +171,7 @@ const GeneratePage: React.FC = () => {
    autoSubtitles,
    bgm,
    generateCount,
+    sourceEditPlanId: editPlanId,
  })

  /* ================================================================
@@ -141,6 +141,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
          onSelectedMaterialsChange={onSelectedMaterialsChange}
          smartSelectedIds={smartSelectedIds}
          onSmartSelectedIdsChange={onSmartSelectedIdsChange}
+          selectedTemplate={selectedTemplate}
        />
      )
    case 3:
@@ -156,6 +157,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
        <Step4TitleSettings
          titleSettings={titleSettings}
          onTitleSettingsChange={onTitleSettingsChange}
+          selectedTemplate={selectedTemplate}
        />
      )
    case 5:
@@ -182,6 +184,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
          duration={duration}
          assetIds={materialMode === "auto" ? smartSelectedIds : selectedMaterials}
          selectedTemplate={selectedTemplate}
+          titleSettings={titleSettings}
        />
      )
    case 7:
@@ -15,6 +15,8 @@ interface Step2MaterialSelectProps {
  onSelectedMaterialsChange: (ids: string[]) => void
  smartSelectedIds: string[]
  onSmartSelectedIdsChange: (ids: string[]) => void
+  /** 当前选中的模板/草稿 ID，用于自动保存 */
+  selectedTemplate?: string
 }

 const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {
@@ -12,6 +12,8 @@ import AiTitleGenerator from "./title/AiTitleGenerator"
 interface Step4TitleSettingsProps {
  titleSettings: TitleSettings
  onTitleSettingsChange: (settings: TitleSettings) => void
+  /** 当前选中的模板/草稿 ID，用于自动保存 */
+  selectedTemplate?: string
 }

 const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
@@ -14,6 +14,8 @@ interface Step6CoverSettingsProps {
  assetIds?: string[]
  /** 当前选中的模板 ID */
  selectedTemplate?: string
+  /** Step4 标题设置，用于预览视频烧录标题 & 封面叠加标题 */
+  titleSettings?: import("../types").TitleSettings
 }

 const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
@@ -40,6 +42,7 @@ const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
    duration: props.duration,
    assetIds: props.assetIds,
    selectedTemplate: props.selectedTemplate,
+    titleSettings: props.titleSettings,
  })

  const handleAutoGenerate = () => {
@@ -19,6 +19,8 @@ export interface UseGenerateVideoProps {
  autoSubtitles: boolean
  bgm: boolean
  generateCount: number
+  /** 当前草稿 ID（URL 参数 edit_plan_id，用于后端回写任务关联） */
+  sourceEditPlanId?: string | null
 }

 /** 生成阶段 */
@@ -1,87 +1,146 @@
 import { useRef, useCallback } from "react"
 import { message } from "antd"
-import { getGenerationStatus, getGenerationTaskResults } from "@/api/template-editor"
+import axios from "axios"
+import { getGenerationTask } from "@/api/tasks/tasks"
+import { getGenerationTaskResults } from "@/api/template-editor"
 import { safeExtractError } from "./errorUtils"

 interface UseGenerationPollingOptions {
-  templateId: string
  onProgress: (progress: number) => void
  onComplete: (videos: unknown[]) => void
  onFailed: (errorMsg: string) => void
 }

+/** 最大连续错误次数（仅对可重试错误），超过后终止轮询 */
+const MAX_RETRYABLE_ERRORS = 10
+/** 获取结果的最大重试次数 */
+const MAX_RESULTS_RETRIES = 3
+
 /**
- * 生成状态轮询 Hook
- * 轮询生成状态，更新进度，处理完成/失败
+ * 生成状态轮询 Hook（v2 — 改用 /generation/tasks/{task_id}）
+ *
+ * 旧版轮询 GET /templates/{id}/editor/generation-status 依赖 plan 维度状态，
+ * 在编辑流程数据链路断裂时拿不到 task_id。新版直接使用 POST /generation/tasks
+ * 返回的 task_id 轮询任务详情，不再依赖 plan。
+ *
+ * 错误处理：
+ * - 4xx（尤其 404）视为不可恢复，立即 onFailed，不再重试
+ * - 5xx / 网络错误重试，最多连续 MAX_RETRYABLE_ERRORS 次
+ * - 任务完成后获取结果失败会重试 MAX_RESULTS_RETRIES 次，仍失败则 onFailed
 */
 export const useGenerationPolling = ({
-  templateId,
  onProgress,
  onComplete,
  onFailed,
 }: UseGenerationPollingOptions) => {
  const progressTimer = useRef<ReturnType<typeof setTimeout>>()
+  const cancelledRef = useRef(false)

  const clearTimer = useCallback(() => {
+    cancelledRef.current = true
    if (progressTimer.current) {
      clearTimeout(progressTimer.current)
      progressTimer.current = undefined
    }
  }, [])

-  const startPolling = useCallback(() => {
-    const poll = async () => {
+  /** 任务完成后拉取结果列表，带重试 */
+  const fetchResultsWithRetry = useCallback(
+    async (taskId: string, attempt = 0): Promise<unknown[] | null> => {
      try {
-        const data = await getGenerationStatus(templateId)
-
-        if (data.plan_status === "completed") {
-          onProgress(100)
-          // 获取生成的视频结果
-          let videos: unknown[] = []
-          if (data.generation_task_id) {
-            try {
-              videos = await getGenerationTaskResults(data.generation_task_id)
-            } catch (err) {
-              console.error("[获取生成结果失败]", err)
-            }
-          }
-          onComplete(videos)
-          message.success("视频生成完成！")
-          return
+        return await getGenerationTaskResults(taskId)
+      } catch (err) {
+        if (cancelledRef.current) return null
+        console.error(`[获取生成结果失败] 第 ${attempt + 1} 次`, err)
+        if (attempt < MAX_RESULTS_RETRIES - 1) {
+          await new Promise((resolve) => setTimeout(resolve, 1000 * (attempt + 1)))
+          return fetchResultsWithRetry(taskId, attempt + 1)
        }
-        if (data.plan_status === "failed") {
-          const dataAny = data as unknown as Record<string, unknown>
-          const rawMsg =
-            dataAny.error_message ||
-            dataAny.error ||
-            dataAny.message ||
-            (Array.isArray(data.clips)
-              ? (data.clips as { status: string; error_message?: string }[]).find(
-                  (c) => c.status === "failed",
-                )?.error_message
-              : undefined) ||
-            "视频生成失败，请联系管理员或重试"
-          const errorMsg = safeExtractError(rawMsg)
-          console.error("[生成失败] templateId:", templateId, "响应:", data)
-          onFailed(errorMsg)
-          message.error(errorMsg)
-          return
-        }
-
-        const clips = data.clips || []
-        const total = clips.length || 1
-        const done = (clips as { status: string }[]).filter((c) => c.status === "completed").length
-        onProgress(Math.round((done / total) * 100))
-
-        progressTimer.current = setTimeout(poll, 2000)
-      } catch (pollErr) {
-        console.error("[轮询出错] templateId:", templateId, pollErr)
-        progressTimer.current = setTimeout(poll, 3000)
+        return null
      }
-    }
+    },
+    [],
+  )

-    progressTimer.current = setTimeout(poll, 2000)
-  }, [templateId, onProgress, onComplete, onFailed])
+  const startPolling = useCallback(
+    (taskId: string) => {
+      cancelledRef.current = false
+      let consecutiveErrors = 0
+
+      const poll = async () => {
+        if (cancelledRef.current) return
+        try {
+          const task = await getGenerationTask(taskId)
+          consecutiveErrors = 0
+
+          if (task.status === "completed") {
+            onProgress(100)
+            const videos = await fetchResultsWithRetry(taskId)
+            if (cancelledRef.current) return
+            if (videos === null) {
+              const errorMsg = "视频已生成，但获取结果列表失败，请稍后在任务列表查看"
+              console.error("[生成结果获取失败] taskId:", taskId)
+              onFailed(errorMsg)
+              message.error(errorMsg)
+              return
+            }
+            onComplete(videos)
+            message.success("视频生成完成！")
+            return
+          }
+
+          if (task.status === "failed" || task.status === "cancelled") {
+            const rawMsg =
+              task.error_info?.error_message ||
+              task.error_message ||
+              (task.status === "cancelled" ? "任务已取消" : "视频生成失败，请联系管理员或重试")
+            const errorMsg = safeExtractError(rawMsg)
+            console.error("[生成失败] taskId:", taskId, "响应:", task)
+            onFailed(errorMsg)
+            message.error(errorMsg)
+            return
+          }
+
+          // pending / waiting / running — 继续轮询
+          const pct = Math.max(0, Math.min(99, Math.round(Number(task.progress) || 0)))
+          onProgress(pct)
+          progressTimer.current = setTimeout(poll, 2000)
+        } catch (pollErr) {
+          if (cancelledRef.current) return
+          console.error("[轮询出错] taskId:", taskId, pollErr)
+
+          // 4xx 不可恢复，立即失败
+          const status = axios.isAxiosError(pollErr) ? pollErr.response?.status : undefined
+          if (status && status >= 400 && status < 500) {
+            const msg =
+              (axios.isAxiosError(pollErr) &&
+                (pollErr.response?.data as { detail?: string; message?: string } | undefined)
+                  ?.detail) ||
+              (axios.isAxiosError(pollErr) &&
+                (pollErr.response?.data as { detail?: string; message?: string } | undefined)
+                  ?.message) ||
+              `查询任务失败 (${status})`
+            const errorMsg = safeExtractError(msg)
+            onFailed(errorMsg)
+            message.error(errorMsg)
+            return
+          }
+
+          consecutiveErrors += 1
+          if (consecutiveErrors >= MAX_RETRYABLE_ERRORS) {
+            const errorMsg = "任务状态查询连续失败，请稍后在任务列表查看结果"
+            onFailed(errorMsg)
+            message.error(errorMsg)
+            return
+          }
+          progressTimer.current = setTimeout(poll, 3000)
+        }
+      }
+
+      progressTimer.current = setTimeout(poll, 1500)
+    },
+    [onProgress, onComplete, onFailed, fetchResultsWithRetry],
+  )

  return { startPolling, clearTimer }
 }
@@ -0,0 +1,124 @@
+/**
+ * 草稿自动保存工具 Hook
+ *
+ * 背景：后端 PUT /templates/{id}/editor 的 config 是「整体替换」语义，
+ * 直接发送 { config: { asset_ids } } 会把 title 等其他字段覆盖掉。
+ * 本 Hook 统一执行「GET 当前 config → 浅合并新字段 → PUT 回去」，
+ * 并用串行队列 + AbortController 保证：
+ *   - 同一时刻只有一个保存请求在飞
+ *   - 快速连续变化时只提交最后一次
+ *   - 组件卸载时取消未完成请求
+ *   - 保存失败时保留补丁，自动重试（指数退避，最多 5 次）
+ *
+ * 保存失败只 console.warn，不弹窗、不阻塞。
+ */
+import { useCallback, useEffect, useRef } from "react"
+import { getEditPlan, updateEditPlan } from "@/api/template-editor"
+
+type ConfigPatch = Record<string, unknown>
+
+/** 最大自动重试次数 */
+const MAX_RETRIES = 5
+/** 初始重试延迟（ms），每次翻倍 */
+const BASE_RETRY_DELAY = 1000
+
+export function useDraftAutoSave(templateId?: string) {
+  const timerRef = useRef<ReturnType<typeof setTimeout>>()
+  const abortRef = useRef<AbortController | null>(null)
+  // 待合并的补丁队列（解决「保存进行中又来了新变化」）
+  const pendingPatchRef = useRef<ConfigPatch | null>(null)
+  const savingRef = useRef(false)
+  const templateIdRef = useRef(templateId)
+  templateIdRef.current = templateId
+
+  const flush = useCallback(async (retryCount = 0) => {
+    const tid = templateIdRef.current
+    if (!tid) return
+    // 已有保存在飞：把新补丁暂存，等当前请求结束后再合并一次
+    if (savingRef.current) return
+
+    // 快照当前补丁，但先不清空 —— 成功后才清除，失败时保留以便重试
+    const patchToSave = pendingPatchRef.current
+    if (!patchToSave) {
+      savingRef.current = false
+      return
+    }
+    savingRef.current = true
+
+    const controller = new AbortController()
+    abortRef.current = controller
+    try {
+      // 1. 读当前 config（拿最新，避免覆盖别人/别的步骤写入的字段）
+      const current = await getEditPlan(tid)
+      if (controller.signal.aborted) return
+      const merged = { ...(current.config || {}), ...patchToSave }
+      // 2. 写回完整合并后的 config
+      await updateEditPlan(tid, { config: merged }, controller.signal)
+      // 3. 保存成功才清除已保存的补丁
+      //    （保存期间可能有新补丁进来，只清除我们已经保存的部分）
+      pendingPatchRef.current = null
+    } catch (err) {
+      const name = (err as { name?: string })?.name
+      if (name === "CanceledError" || name === "AbortError") {
+        // 组件卸载或新请求取消，不重试
+        return
+      }
+      console.warn("[useDraftAutoSave] 自动保存草稿失败:", err)
+
+      // 保存失败：把本次尝试保存的补丁合并回 pendingPatchRef
+      // （保存期间可能有新补丁，新补丁优先）
+      pendingPatchRef.current = {
+        ...patchToSave,
+        ...(pendingPatchRef.current || {}),
+      }
+
+      // 指数退避重试
+      if (retryCount < MAX_RETRIES && !controller.signal.aborted) {
+        const delay = BASE_RETRY_DELAY * Math.pow(2, retryCount)
+        timerRef.current = setTimeout(() => {
+          void flush(retryCount + 1)
+        }, delay)
+      }
+      // 超过最大重试次数后，补丁仍保留在 pendingPatchRef 中，
+      // 下次 scheduleSave 触发时会一起带上
+    } finally {
+      savingRef.current = false
+      // 保存期间又积累了新变化（且不是在重试路径中），再触发一次
+      if (pendingPatchRef.current && !controller.signal.aborted && retryCount === 0) {
+        timerRef.current = setTimeout(() => {
+          void flush()
+        }, 50)
+      }
+    }
+  }, [])
+
+  /**
+   * 调度一次自动保存（防抖）
+   * @param patch 要合并进 config 的局部字段
+   * @param delay 防抖毫秒数
+   */
+  const scheduleSave = useCallback(
+    (patch: ConfigPatch, delay = 500) => {
+      const tid = templateIdRef.current
+      if (!tid) return
+      if (timerRef.current) clearTimeout(timerRef.current)
+      // 累计补丁（同一周期内多次变化合并成一次写入）
+      pendingPatchRef.current = { ...(pendingPatchRef.current || {}), ...patch }
+      timerRef.current = setTimeout(() => {
+        void flush()
+      }, delay)
+    },
+    [flush],
+  )
+
+  useEffect(() => {
+    return () => {
+      if (timerRef.current) clearTimeout(timerRef.current)
+      if (abortRef.current) abortRef.current.abort()
+    }
+  }, [])
+
+  return { scheduleSave }
+}
+
+export default useDraftAutoSave
@@ -34,7 +34,6 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
  }, [])

  const { startPolling, clearTimer } = useGenerationPolling({
-    templateId: selectedTemplate,
    onProgress: handleProgress,
    onComplete: handleComplete,
    onFailed: handleFailed,
@@ -90,16 +89,20 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
      const assetIds =
        props.materialMode === "auto" ? props.smartSelectedIds : props.selectedMaterials

+      // 封面 URL：优先 AI 生成缩略图，兜底用户上传
+      const coverUrl = props.coverSettings?.thumbnail_url || props.coverSettings?.upload_url || ""
+
      // 直接创建正式生成任务
-      await createGenerationTask({
+      const taskResp = await createGenerationTask({
        template_id: selectedTemplate,
        asset_ids: assetIds,
        output_width: outputWidth,
        output_height: outputHeight,
-        cover_url: props.coverSettings?.upload_url || "",
+        cover_url: coverUrl,
        custom_title: props.titleSettings?.title || "",
        duration: props.duration || undefined,
        video_ratio: props.videoRatio,
+        ...(props.sourceEditPlanId ? { source_edit_plan_id: props.sourceEditPlanId } : {}),
        ...(props.titleSettings?.title
          ? {
              title_config: {
@@ -116,7 +119,12 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
          : {}),
      })

-      startPolling()
+      // 从创建响应直接拿 task_id，改用新接口轮询
+      const taskId = taskResp.items?.[0]?.id
+      if (!taskId) {
+        throw new Error("创建任务成功但未返回任务 ID，请稍后在任务列表查看")
+      }
+      startPolling(taskId)
    } catch (err: unknown) {
      console.error("[handleGenerate] 生成失败:", err)
      setGenerating(false)
@@ -6,6 +6,7 @@ import { useCallback, useEffect, useRef } from "react"
 import { formatDuration } from "../utils/formatDuration"
 import { useMaterialLibrary } from "./step2-materials/useMaterialLibrary"
 import { useSmartMatch } from "./step2-materials/useSmartMatch"
+import { useDraftAutoSave } from "./useDraftAutoSave"

 interface UseStep2MaterialsProps {
  materialMode: "manual" | "auto"
@@ -14,6 +15,8 @@ interface UseStep2MaterialsProps {
  onSelectedMaterialsChange: (ids: string[]) => void
  smartSelectedIds: string[]
  onSmartSelectedIdsChange: (ids: string[]) => void
+  /** 当前选中的模板/草稿 ID，用于自动保存 */
+  selectedTemplate?: string
 }

 export function useStep2Materials({
@@ -23,6 +26,7 @@ export function useStep2Materials({
  onSelectedMaterialsChange,
  smartSelectedIds,
  onSmartSelectedIdsChange,
+  selectedTemplate,
 }: UseStep2MaterialsProps) {
  const { libraries, selectedLibraryId, setSelectedLibraryId, materials, materialsLoading } =
    useMaterialLibrary()
@@ -53,6 +57,14 @@ export function useStep2Materials({
    handleSmartMatch()
  }, [selectedLibraryId, materialMode, materialsLoading, materials.items, handleSmartMatch])

+  /* ── Step2 选择素材后自动保存草稿（防抖 500ms，失败静默） ── */
+  const { scheduleSave } = useDraftAutoSave(selectedTemplate)
+  useEffect(() => {
+    if (!selectedTemplate) return
+    const ids = materialMode === "auto" ? smartSelectedIds : selectedMaterials
+    scheduleSave({ asset_ids: ids }, 500)
+  }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, scheduleSave])
+
  /* ── 手动选择素材 ── */
  const handleToggleMaterial = useCallback(
    (materialId: string) => {
@@ -4,17 +4,24 @@ import { getTitles } from "@/api/titles"
 import type { TitleSettings } from "../../types"
 import { useAiTitleGenerator } from "./useAiTitleGenerator"
 import { useTitleStyleUpdaters } from "./useTitleStyleUpdaters"
+import { useDraftAutoSave } from "../useDraftAutoSave"

 interface UseStep4TitleProps {
  titleSettings: TitleSettings
  onTitleSettingsChange: (settings: TitleSettings) => void
+  /** 当前选中的模板/草稿 ID，用于自动保存 */
+  selectedTemplate?: string
 }

 /**
 * Step 4 标题设置 Hook
 * 封装 AI 标题生成、标题样式设置等逻辑
 */
-export function useStep4Title({ titleSettings, onTitleSettingsChange }: UseStep4TitleProps) {
+export function useStep4Title({
+  titleSettings,
+  onTitleSettingsChange,
+  selectedTemplate,
+}: UseStep4TitleProps) {
  // 标题库数据
  const { data: userTitles = [] } = useQuery({
    queryKey: ["titles"],
@@ -42,6 +49,38 @@ export function useStep4Title({ titleSettings, onTitleSettingsChange }: UseStep4
  const prevAiAutoSelect = useRef(titleSettings.aiAutoSelect)
  const isFirstMount = useRef(true)

+  /* ── Step4 标题内容/样式变化后自动保存草稿（防抖 800ms，失败静默） ── */
+  const { scheduleSave: scheduleTitleSave } = useDraftAutoSave(selectedTemplate)
+  useEffect(() => {
+    if (!selectedTemplate) return
+    scheduleTitleSave(
+      {
+        title: {
+          text: titleSettings.title,
+          font: titleSettings.font,
+          font_size: titleSettings.size,
+          color: titleSettings.color,
+          position: titleSettings.position,
+          bold: titleSettings.bold,
+          stroke: titleSettings.stroke,
+          shadow: titleSettings.shadow,
+        },
+      },
+      800,
+    )
+  }, [
+    selectedTemplate,
+    titleSettings.title,
+    titleSettings.font,
+    titleSettings.size,
+    titleSettings.color,
+    titleSettings.position,
+    titleSettings.bold,
+    titleSettings.stroke,
+    titleSettings.shadow,
+    scheduleTitleSave,
+  ])
+
  // 当 AI 自动选择开关打开时，自动生成/选择一个标题填入
  // 首次挂载时如果开关已经是 true 且无标题，也需要触发
  useEffect(() => {
@@ -7,6 +7,8 @@ import { message } from "antd"
 import type { CoverConfig, CoverTemplate } from "../types/cover"
 import { generateCover } from "@/api/generation"
 import { createPreview, getPreviewStatus } from "@/api/generation/preview"
+import { updateEditPlan } from "@/api/template-editor"
+import type { TitleSettings } from "../types"
 import {
  fetchCoverTemplates,
  createCoverTemplate,
@@ -22,6 +24,8 @@ interface UseStep6CoverProps {
  assetIds?: string[]
  /** 当前选中的模板 ID */
  selectedTemplate?: string
+  /** Step4 标题设置，用于预览视频烧录标题 & 封面叠加标题 */
+  titleSettings?: TitleSettings
 }

 export function useStep6Cover({
@@ -30,6 +34,7 @@ export function useStep6Cover({
  duration,
  assetIds = [],
  selectedTemplate = "",
+  titleSettings,
 }: UseStep6CoverProps) {
  const [generating, setGenerating] = useState(false)

@@ -92,6 +97,20 @@ export function useStep6Cover({
      const response = await generateCover(selectedTemplate, {
        asset_ids: assetIds,
        cover_type: "ai_frame",
+        ...(titleSettings?.title
+          ? {
+              title_config: {
+                text: titleSettings.title,
+                font: titleSettings.font,
+                font_size: titleSettings.size,
+                font_color: titleSettings.color,
+                position: titleSettings.position,
+                bold: titleSettings.bold,
+                stroke: titleSettings.stroke,
+                shadow: titleSettings.shadow,
+              },
+            }
+          : {}),
      })
      clearTimeout(timeoutId)
      const thumbnailUrl = response.cover?.image_url || ""
@@ -136,6 +155,20 @@ export function useStep6Cover({
            template_id: selectedTemplate,
            asset_ids: assetIds,
            duration: duration || 30,
+            ...(titleSettings?.title
+              ? {
+                  title_config: {
+                    text: titleSettings.title,
+                    font: titleSettings.font,
+                    font_size: titleSettings.size,
+                    font_color: titleSettings.color,
+                    position: titleSettings.position,
+                    bold: titleSettings.bold,
+                    stroke: titleSettings.stroke,
+                    shadow: titleSettings.shadow,
+                  },
+                }
+              : {}),
          })
          // 轮询等待预览渲染完成：递归 setTimeout 避免请求重叠 + 120s 超时兜底
          await new Promise<void>((resolve, reject) => {
@@ -154,6 +187,21 @@ export function useStep6Cover({
              try {
                const status = await getPreviewStatus(previewResp.task_id)
                if (status.status === "completed") {
+                  // 保存预览视频地址到 plan.config.rendered_storage_key，
+                  // 供封面 API 的 E1 兜底路径定位渲染后的视频（含标题烧录）。
+                  // video_url 可能是完整 http(s) URL 或 OSS storage_key，两种格式后端都能处理。
+                  if (status.video_url) {
+                    try {
+                      await updateEditPlan(selectedTemplate, {
+                        config: { rendered_storage_key: status.video_url },
+                      })
+                    } catch (saveErr) {
+                      console.warn(
+                        "[Step6] 保存 rendered_storage_key 失败（不阻塞封面重试）:",
+                        saveErr,
+                      )
+                    }
+                  }
                  done(() => resolve())
                } else if (status.status === "failed") {
                  done(() => reject(new Error(status.error_message || "预览渲染失败")))
@@ -171,6 +219,20 @@ export function useStep6Cover({
          const retryResp = await generateCover(selectedTemplate, {
            asset_ids: assetIds,
            cover_type: "ai_frame",
+            ...(titleSettings?.title
+              ? {
+                  title_config: {
+                    text: titleSettings.title,
+                    font: titleSettings.font,
+                    font_size: titleSettings.size,
+                    font_color: titleSettings.color,
+                    position: titleSettings.position,
+                    bold: titleSettings.bold,
+                    stroke: titleSettings.stroke,
+                    shadow: titleSettings.shadow,
+                  },
+                }
+              : {}),
          })
          const retryUrl = retryResp.cover?.image_url || ""
          if (retryUrl) {
@@ -212,7 +274,15 @@ export function useStep6Cover({
      clearTimeout(timeoutId)
      setGenerating(false)
    }
-  }, [selectedTemplate, assetIds, coverSettings, onCoverSettingsChange, generating, duration])
+  }, [
+    selectedTemplate,
+    assetIds,
+    coverSettings,
+    onCoverSettingsChange,
+    generating,
+    duration,
+    titleSettings,
+  ])

  // ── 模板操作方法 ──
  const handleSelectTemplate = useCallback((id: string) => {
@@ -32,7 +32,7 @@ describe("bgm API", () => {

  describe("getBgmPresets", () => {
    it("should resolve successfully", async () => {
-      await expect(getBgmPresets("test-params?")).resolves.not.toThrow()
+      await expect(getBgmPresets("test-template", { category: "test" })).resolves.not.toThrow()
    })

    it("should reject on API error", async () => {
@@ -42,7 +42,7 @@ describe("bgm API", () => {
      mockDelete.mockRejectedValue(new Error("Network error"))
      mockPatch.mockRejectedValue(new Error("Network error"))

-      await expect(getBgmPresets("test-params?")).rejects.toThrow()
+      await expect(getBgmPresets("test-template", { category: "test" })).rejects.toThrow()
    })
  })
 })
@@ -1,16 +1,12 @@
 import { describe, expect, it, vi, beforeEach } from "vitest"
 import {
-  getEditPlans,
  getEditPlan,
-  createEditPlan,
  updateEditPlan,
-  deleteEditPlan,
  generateEditPlan,
  getGenerationStatus,
  aiRecommendClips,
  getEditPlanGenerations,
  getGenerationTaskResults,
-  cancelGeneration,
  getEditPlanClips,
  getEditPlanClip,
  createEditPlanClip,
@@ -19,7 +15,6 @@ import {
  reorderEditPlanClips,
  batchDeleteEditPlanClips,
  createClipsFromAssets,
-  copyEditPlan,
  getMediaAssets,
  getMediaAsset,
 } from "@/api/template-editor"
@@ -53,22 +48,6 @@ describe("editPlans API", () => {
    mockPatch.mockResolvedValue({ data: { success: true, items: [] } })
  })

-  describe("getEditPlans", () => {
-    it("should resolve successfully", async () => {
-      await expect(getEditPlans("test-params?")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(getEditPlans("test-params?")).rejects.toThrow()
-    })
-  })
-
  describe("getEditPlan", () => {
    it("should resolve successfully", async () => {
      await expect(getEditPlan("test-planId")).resolves.not.toThrow()
@@ -85,22 +64,6 @@ describe("editPlans API", () => {
    })
  })

-  describe("createEditPlan", () => {
-    it("should resolve successfully", async () => {
-      await expect(createEditPlan({ name: "test-item" })).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(createEditPlan({ name: "test-item" })).rejects.toThrow()
-    })
-  })
-
  describe("updateEditPlan", () => {
    it("should resolve successfully", async () => {
      await expect(updateEditPlan("test-planId")).resolves.not.toThrow()
@@ -117,22 +80,6 @@ describe("editPlans API", () => {
    })
  })

-  describe("deleteEditPlan", () => {
-    it("should resolve successfully", async () => {
-      await expect(deleteEditPlan("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(deleteEditPlan("test-planId")).rejects.toThrow()
-    })
-  })
-
  describe("generateEditPlan", () => {
    it("should resolve successfully", async () => {
      await expect(generateEditPlan("test-planId")).resolves.not.toThrow()
@@ -213,22 +160,6 @@ describe("editPlans API", () => {
    })
  })

-  describe("cancelGeneration", () => {
-    it("should resolve successfully", async () => {
-      await expect(cancelGeneration("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(cancelGeneration("test-planId")).rejects.toThrow()
-    })
-  })
-
  describe("getEditPlanClips", () => {
    it("should resolve successfully", async () => {
      await expect(getEditPlanClips("test-planId")).resolves.not.toThrow()
@@ -357,22 +288,6 @@ describe("editPlans API", () => {
    })
  })

-  describe("copyEditPlan", () => {
-    it("should resolve successfully", async () => {
-      await expect(copyEditPlan("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(copyEditPlan("test-planId")).rejects.toThrow()
-    })
-  })
-
  describe("getMediaAssets", () => {
    it("should resolve successfully", async () => {
      await expect(getMediaAssets("test-libraryId?")).resolves.not.toThrow()
@@ -5,7 +5,6 @@ import {
  getTemplate,
  toggleFavoriteTemplate,
  copyTemplate,
-  generateFromTemplate,
 } from "@/api/templates"

 const mockGet = vi.fn()
@@ -116,20 +115,4 @@ describe("templates API", () => {
      await expect(copyTemplate("test-templateId")).rejects.toThrow()
    })
  })
-
-  describe("generateFromTemplate", () => {
-    it("should resolve successfully", async () => {
-      await expect(generateFromTemplate("test-templateId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(generateFromTemplate("test-templateId")).rejects.toThrow()
-    })
-  })
 })
@@ -150,12 +150,10 @@ vi.mock("@/api/template-editor", () => ({
  getMediaAssets: vi.fn().mockResolvedValue({ items: [] }),
  getEditPlanGenerations: vi.fn().mockResolvedValue({ items: [] }),
  getEditPlan: vi.fn().mockResolvedValue({}),
-  createEditPlan: vi.fn().mockResolvedValue({ id: "test-plan" }),
  updateEditPlan: vi.fn().mockResolvedValue({}),
  generateEditPlan: vi.fn().mockResolvedValue({ task_id: "test-task" }),
  getGenerationStatus: vi.fn().mockResolvedValue({ status: "completed" }),
  getGenerationTaskResults: vi.fn().mockResolvedValue({ items: [] }),
-  cancelGeneration: vi.fn().mockResolvedValue({}),
  getEditPlanClips: vi.fn().mockResolvedValue({ items: [] }),
  createEditPlanClip: vi.fn().mockResolvedValue({}),
  batchDeleteEditPlanClips: vi.fn().mockResolvedValue({}),
@@ -109,7 +109,6 @@ vi.mock("@/api/templates", () => ({
  getTemplate: vi.fn().mockResolvedValue({ items: [], total: 0, success: true }),
  toggleFavoriteTemplate: vi.fn().mockResolvedValue({ items: [], total: 0, success: true }),
  copyTemplate: vi.fn().mockResolvedValue({ items: [], total: 0, success: true }),
-  generateFromTemplate: vi.fn().mockResolvedValue({ items: [], total: 0, success: true }),
 }))

 vi.mock("@/pages/templates/TemplateLibrary.css", () => ({}))
@@ -585,14 +585,11 @@ class RenderAdapter:
        try:
            from video_processing.thumbnail_generator import extract_and_upload_cover_frames

-            # 从 plan config 提取标题文字，叠加到封面候选帧上
-            _title_cfg = (plan_config or {}).get("title", {}) or {}
-            if not isinstance(_title_cfg, dict):
-                _title_cfg = {}
-            _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else ""
-
+            # 已渲染视频在统一渲染阶段已通过 ASS 字幕把标题烧录进画面，
+            # 抽帧天然带标题，因此这里传空字符串，避免 Pillow 二次叠加导致重影。
+            # Pillow 叠加仅用于 API 从源素材抽帧（源素材本身无标题）的兜底场景。
            cover_candidates = extract_and_upload_cover_frames(
-                str(result.output_path), plan_id, num_frames=3, title_text=_title_text
+                str(result.output_path), plan_id, num_frames=3, title_text=""
            )
            if cover_candidates:
                logger.info(
@@ -1,7 +1,8 @@
-"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。
+"""视频封面抽帧工具 — 从视频中抽取帧作为封面，支持标题文字叠加。

-统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
-渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
+统一封面管道：
+- 从已渲染视频抽帧：标题已通过 ASS 字幕烧进视频，帧天然带标题，无需再叠加。
+- 从源素材抽帧（API E2 兜底）：源素材无标题，通过 Pillow 在帧上绘制标题文字。
 """

 from __future__ import annotations
@@ -12,6 +13,38 @@ from pathlib import Path

 logger = logging.getLogger(__name__)

+# ── 标题叠加（Pillow）──────────────────────────────────────────────────────
+# 实现统一放在 packages/shared/title_overlay.py，API 和 Worker 共用。
+
+
+def apply_title_overlay(
+    image_path: str,
+    title_text: str,
+    *,
+    position: str = "bottom",
+    font_size: int | None = None,
+    margin_ratio: float = 0.06,
+    stroke_width_ratio: float = 0.04,
+) -> str:
+    """在图片上绘制标题文字（白色 + 黑色描边/阴影）。
+
+    委托给 packages.shared.title_overlay.apply_title_to_image，
+    保持 Worker 内调用方式不变。title_text 为空时直接返回原路径。
+    """
+    from packages.shared.title_overlay import apply_title_to_image
+
+    if not title_text or not title_text.strip():
+        return image_path
+    result = apply_title_to_image(
+        image_path,
+        title_text,
+        position=position,
+        font_size=font_size,
+        margin_ratio=margin_ratio,
+        stroke_width_ratio=stroke_width_ratio,
+    )
+    return result or image_path
+

 def extract_first_frame(
    video_path: str,
@@ -182,7 +215,8 @@ def extract_and_upload_cover_frames(
        video_path: 视频文件路径
        plan_id: 编辑计划 ID（用于生成 storage key）
        num_frames: 抽取帧数（默认 3）
-        title_text: 标题文字（当前版本未叠加，预留参数）
+        title_text: 标题文字；非空时用 Pillow 叠加到每帧（白色 + 黑色描边）。
+            从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。

    Returns:
        封面候选列表，每项包含 {"url": str, "position": float}
@@ -208,6 +242,9 @@ def extract_and_upload_cover_frames(
                seek_ratio=ratio,
                min_seek_seconds=0.5,
            )
+            # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
+            if title_text and title_text.strip():
+                apply_title_overlay(frame_path, title_text)
            storage_key = f"covers/{plan_id}/frame_{i}.jpg"
            url = upload_to_oss(frame_path, storage_key)
            if url:
@@ -19,4 +19,14 @@ celery_app.conf.imports = (
    "worker_app.tasks.batch_download",
    "worker_app.tasks._startup",
    "apps.worker.video_processing.dedup",
+    "worker_app.tasks.cleanup",
 )
+
+# Celery Beat 定时任务调度
+celery_app.conf.beat_schedule = {
+    "cleanup-stale-pending-tasks": {
+        "task": "worker.cleanup_stale_pending_tasks",
+        "schedule": 600.0,  # 每 10 分钟（秒）
+        "options": {"expires": 300},  # 5 分钟过期，避免堆积
+    },
+}
@@ -10,6 +10,9 @@ logger = logging.getLogger(__name__)
 # 孤儿任务超时阈值：渲染任务超过此时间未更新则视为卡死
 ORPHAN_TASK_TIMEOUT_MINUTES = 10

+# Pending 任务超时阈值：pending 任务在队列中等待超过此时间则自动清理
+PENDING_TASK_TIMEOUT_MINUTES = 30
+

 def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:  # pragma: no cover
    """清理数据库中超时未更新的 running GenerationTask（孤儿任务）。
@@ -83,6 +86,38 @@ def cleanup_stale_jobs(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> in
        return 0


+def cleanup_stale_pending_tasks(timeout_minutes: int = PENDING_TASK_TIMEOUT_MINUTES) -> int:  # pragma: no cover
+    """清理数据库中卡在 pending 状态超时的 GenerationTask。
+
+    全局任务队列有 pending 数量上限，长期卡在 pending 的任务会占满队列，
+    导致新用户无法创建任务。通过 created_at 超时判断并标记为 failed。
+
+    Args:
+        timeout_minutes: 超时时间（分钟），默认 PENDING_TASK_TIMEOUT_MINUTES
+
+    Returns:
+        清理的任务数量
+    """
+    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+        SQLAlchemyGenerationTaskRepository,
+    )
+
+    session = SessionLocal()
+    try:
+        repo = SQLAlchemyGenerationTaskRepository(session)
+        count = repo.cleanup_stale_pending(timeout_minutes)
+        if count > 0:
+            logger.warning("清理了 %d 个超时的 pending GenerationTask（超过 %d 分钟未处理）", count, timeout_minutes)
+        else:
+            logger.info("无超时 pending GenerationTask 需要清理")
+        return count
+    except Exception as e:
+        logger.error("清理超时 pending GenerationTask 失败: %s", e, exc_info=True)
+        return 0
+    finally:
+        session.close()
+
+
 def cleanup_all_stale_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> dict:  # pragma: no cover
    """统一清理所有超时的孤儿任务。

@@ -93,15 +128,17 @@ def cleanup_all_stale_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES)
    """
    gen_count = cleanup_orphan_tasks(timeout_minutes)
    job_count = cleanup_stale_jobs(timeout_minutes)
-    total = gen_count + job_count
+    pending_count = cleanup_stale_pending_tasks(PENDING_TASK_TIMEOUT_MINUTES)
+    total = gen_count + job_count + pending_count
    if total > 0:
        logger.warning(
-            "孤儿任务清理完成: GenerationTask=%d, Job=%d, 总计=%d",
+            "任务清理完成: 孤儿 GenerationTask=%d, 孤儿 Job=%d, 超时 pending=%d, 总计=%d",
            gen_count,
            job_count,
+            pending_count,
            total,
        )
-    return {"generation_tasks": gen_count, "jobs": job_count}
+    return {"generation_tasks": gen_count, "jobs": job_count, "pending": pending_count}


@worker_ready.connect
@@ -0,0 +1,35 @@
+"""定期清理任务 — Celery Beat 调度。
+
+包含：
+- cleanup_stale_pending_tasks: 定期清理卡在 pending 超时的 generation_tasks
+"""
+
+import logging
+
+from celery import shared_task
+from worker_app.tasks._startup import (
+    PENDING_TASK_TIMEOUT_MINUTES,
+    cleanup_stale_pending_tasks,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(name="worker.cleanup_stale_pending_tasks")
+def scheduled_cleanup_stale_pending(timeout_minutes: int = PENDING_TASK_TIMEOUT_MINUTES) -> dict:
+    """Celery Beat 调度的定期任务：清理超时的 pending 任务。
+
+    每 10 分钟执行一次（由 celery_app.py 的 beat_schedule 配置），
+    查找所有 status='pending' 且 created_at < NOW() - timeout_minutes
+    的 generation_tasks，批量更新为 failed。
+
+    Args:
+        timeout_minutes: 超时时间（分钟），默认 30 分钟
+
+    Returns:
+        {"cleaned": int}
+    """
+    count = cleanup_stale_pending_tasks(timeout_minutes)
+    if count > 0:
+        logger.info("[Beat] 清理了 %d 个超时 pending 任务（超时阈值 %d 分钟）", count, timeout_minutes)
+    return {"cleaned": count}
@@ -887,6 +887,7 @@ def _load_task_info(task_id: str) -> dict | None:
            "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT,
            "cover_url": getattr(gen_task, "cover_url", "") or "",
            "custom_title": getattr(gen_task, "custom_title", "") or "",
+            "title_config": dict(getattr(gen_task, "title_config", {}) or {}),
            "voice_ids": list(getattr(gen_task, "voice_ids", []) or []),
        }
    finally:
@@ -967,14 +968,15 @@ def _render_video(
    bgm_config: dict | None = None,
    voice_ids: list[str] | None = None,
    custom_title: str = "",
-) -> tuple[Path, float]:
+    title_config: dict | None = None,
+) -> tuple[Path, float, list[dict] | None]:
    """渲染视频（含配音混音）。

    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。

    Args:
    Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_candidates)
    """
    if not downloaded_videos:
        raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -999,27 +1001,34 @@ def _render_video(
                list(template_config.keys()),
            )

-    # ── 用户自定义标题覆盖模板标题配置 ──────────────────────────────────
-    if custom_title:
+    # ── 用户自定义标题：title_config 优先，custom_title 兜底 ─────────────
+    effective_title_cfg: dict | None = None
+    if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip():
+        effective_title_cfg = dict(title_config)
+    elif custom_title:
        try:
-            user_title_cfg = json.loads(custom_title) if isinstance(custom_title, str) else custom_title
-            if isinstance(user_title_cfg, dict) and user_title_cfg.get("text", "").strip():
-                # 字段名归一化: 前端 font_size/font_color → 后端 size/color
-                if "font_size" in user_title_cfg and "size" not in user_title_cfg:
-                    user_title_cfg["size"] = user_title_cfg["font_size"]
-                if "font_color" in user_title_cfg and "color" not in user_title_cfg:
-                    user_title_cfg["color"] = user_title_cfg["font_color"]
-                plan_cfg = dict(virtual_plan.config or {})
-                plan_cfg["title"] = user_title_cfg
-                virtual_plan.config = plan_cfg
-                logger.info(
-                    "[task_id=%s] [渲染] 用户自定义标题已注入: text=%s",
-                    task_id,
-                    user_title_cfg.get("text", "")[:30],
-                )
+            parsed = json.loads(custom_title) if isinstance(custom_title, str) else custom_title
+            if isinstance(parsed, dict) and parsed.get("text", "").strip():
+                effective_title_cfg = parsed
        except (json.JSONDecodeError, TypeError):
            logger.warning("[task_id=%s] custom_title JSON解析失败: %s", task_id, custom_title[:100])

+    if effective_title_cfg:
+        # 字段名归一化: 前端 font_size/font_color → 后端 size/color
+        if "font_size" in effective_title_cfg and "size" not in effective_title_cfg:
+            effective_title_cfg["size"] = effective_title_cfg["font_size"]
+        if "font_color" in effective_title_cfg and "color" not in effective_title_cfg:
+            effective_title_cfg["color"] = effective_title_cfg["font_color"]
+        plan_cfg = dict(virtual_plan.config or {})
+        plan_cfg["title"] = effective_title_cfg
+        virtual_plan.config = plan_cfg
+        logger.info(
+            "[task_id=%s] [渲染] 标题配置已注入(source=%s): text=%s",
+            task_id,
+            "title_config" if title_config else "custom_title",
+            effective_title_cfg.get("text", "")[:30],
+        )
+
    # 用户自定义 BGM 覆盖模板 BGM（用户指定优先级最高）
    if bgm_config:
        plan_cfg = virtual_plan.config or {}
@@ -1113,8 +1122,10 @@ def _render_video(

    # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
    output_path = render_output_path
+    # RenderAdapter 在渲染完成后用本地 ffmpeg 抽取的封面候选帧（已上传 OSS）
+    cover_candidates = getattr(render_result, "cover_candidates", None)

-    return output_path, render_duration
+    return output_path, render_duration, cover_candidates


 def _upload_and_record(
@@ -1385,7 +1396,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_candidates = _render_video(
                task_id=task_id,
                downloaded_videos=downloaded_videos,
                voice_path=audio_path,
@@ -1399,6 +1410,7 @@ def generate_video(self, task_id: str) -> dict:
                bgm_config=task_info.get("bgm_config", {}),
                voice_ids=task_info.get("voice_ids", []),
                custom_title=task_info.get("custom_title", ""),
+                title_config=task_info.get("title_config", {}),
            )

            if gen_task:
@@ -1430,51 +1442,47 @@ def generate_video(self, task_id: str) -> dict:

            _update_task_progress(task_id, 95, "上传完成")

-            # ── 4.5 封面抽帧 ────────────────────────────────────────────────
-            # 预览视频上传完成后，提取封面帧写入 gen_task.cover_url
-            # 这样封面路由（generation_cover.py 步骤A）可以通过 generation_task_id 直接找到
+            # ── 4.5 封面帧持久化 ────────────────────────────────────────────
+            # RenderAdapter 在渲染完成后已用本地 ffmpeg 从 output_path 抽帧
+            # （标题通过 ASS 烧录，帧天然带标题），并上传 OSS 返回 cover_candidates。
+            # 这里把第一帧写入 gen_task.cover_url，完整列表写入 metadata，
+            # 封面路由（generation_cover.py）的 A/B/C/D 步骤即可直接命中。
            try:
-                from packages.shared.mediakit_client import get_mediakit_client
+                if cover_candidates:
+                    first = cover_candidates[0]
+                    # 候选帧字段兼容：RenderAdapter 用 image_url，thumbnail_generator 用 url
+                    cover_frame_url = first.get("image_url") or first.get("url") or ""
+                    if cover_frame_url:
+                        _cover_session = SessionLocal()
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import (
+                                GenerationTaskModel,
+                            )

-                mk_client = get_mediakit_client()
-                if mk_client.is_available:
-                    _update_task_progress(task_id, 96, "提取封面帧")
-                    snapshots = mk_client.extract_frames(
-                        video_url=file_url,
-                        strategy="SpecifiedFrames",
-                        max_frames=1,
-                    )
-                    if snapshots and len(snapshots) > 0:
-                        cover_frame_url = snapshots[0].get("image_url", "")
-                        if cover_frame_url and gen_task:
-                            # 通过独立 session 持久化 cover_url
-                            _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_frame_url
+                                # 持久化完整候选列表到 metadata
+                                meta = dict(_cover_model.metadata or {})
+                                meta["cover_candidates"] = cover_candidates
+                                _cover_model.metadata = meta
+                                _cover_session.commit()
+                                logger.info(
+                                    "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d",
+                                    task_id,
+                                    cover_frame_url[:80],
+                                    len(cover_candidates),
                                )
-
-                                _cover_model = (
-                                    _cover_session.query(GenerationTaskModel)
-                                    .filter(GenerationTaskModel.id == task_id)
-                                    .first()
-                                )
-                                if _cover_model:
-                                    _cover_model.cover_url = cover_frame_url
-                                    _cover_session.commit()
-                                    logger.info(
-                                        "[task_id=%s] 封面帧提取成功: %s",
-                                        task_id,
-                                        cover_frame_url[:80],
-                                    )
-                            finally:
-                                _cover_session.close()
-                    else:
-                        logger.warning("[task_id=%s] 封面帧提取返回空结果", task_id)
+                        finally:
+                            _cover_session.close()
                else:
-                    logger.warning("[task_id=%s] MediaKit 未配置，跳过封面帧提取", task_id)
+                    logger.warning("[task_id=%s] 渲染未产出 cover_candidates，封面将依赖 API 兜底", task_id)
            except Exception:
-                logger.warning("[task_id=%s] 封面帧提取失败（不影响主流程）", task_id, exc_info=True)
+                logger.warning("[task_id=%s] 封面帧持久化失败（不影响主流程）", task_id, exc_info=True)

            # ── 5. 标记完成 ──────────────────────────────────────────────────
            _update_task_status(task_id, "mark_completed", result_count=video_count)
@@ -17,6 +17,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq-dev \
    libpq5 \
    ffmpeg \
+    fonts-noto-cjk \
+    fontconfig \
    && rm -rf /var/lib/apt/lists/*

 # 创建虚拟环境
@@ -6,8 +6,13 @@ set -e

 CONCURRENCY="${WORKER_CONCURRENCY:-2}"

+# ⚠️ 部署约束：此 Worker 必须且只能运行单实例（replicas=1）
+# -B 标志嵌入 celery beat，beat 负责定期触发 pending 超时清理等定时任务
+# 多实例部署会导致每个 Worker 独立运行 Beat，造成定时任务重复执行
+# 若需横向扩展 Worker，必须将 Beat 拆分为独立服务（celery beat -A worker_app.celery_app）
 exec celery \
  -A worker_app.celery_app \
  worker \
  --loglevel=info \
+  "-B" \
  "--concurrency=${CONCURRENCY}"
@@ -42,6 +42,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
        output_height=getattr(model, "output_height", 720) or 720,
        cover_url=getattr(model, "cover_url", "") or "",
        custom_title=getattr(model, "custom_title", "") or "",
+        title_config=dict(getattr(model, "title_config", {}) or {}),
        logs=model.logs or "[]",
        created_at=model.created_at,
        updated_at=model.updated_at,
@@ -86,6 +87,7 @@ class SQLAlchemyGenerationTaskRepository:
            output_height=task.output_height,
            cover_url=task.cover_url or "",
            custom_title=task.custom_title or "",
+            title_config=dict(task.title_config) if task.title_config else {},
            logs=task.logs,
            created_at=task.created_at,
            updated_at=task.updated_at,
@@ -273,6 +275,7 @@ class SQLAlchemyGenerationTaskRepository:
        model.output_height = task.output_height
        model.cover_url = task.cover_url or ""
        model.custom_title = task.custom_title or ""
+        model.title_config = dict(task.title_config) if task.title_config else {}
        model.logs = task.logs
        self.session.commit()
        return task
@@ -310,3 +313,42 @@ class SQLAlchemyGenerationTaskRepository:
            model.completed_at = datetime.now(timezone.utc)
        self.session.commit()
        return len(models)
+
+    def cleanup_stale_pending(self, timeout_minutes: int = 30) -> int:
+        """清理超时的 pending 任务（未被 Worker 拉取的任务）。
+
+        全局任务队列有 pending 数量上限，长期卡在 pending 的任务会占满队列，
+        导致新用户无法创建任务。将超时的 pending 任务标记为 failed。
+
+        Args:
+            timeout_minutes: 超时时间（分钟），默认 30 分钟
+
+        Returns:
+            清理的任务数量
+        """
+        from datetime import timedelta
+
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+        error_info = {
+            "error_type": "PendingTimeout",
+            "message": f"任务在 pending 状态停留超过 {timeout_minutes} 分钟，自动清理",
+            "failed_at": datetime.now(timezone.utc).isoformat(),
+        }
+        count = (
+            self.session.query(GenerationTaskModel)
+            .filter(
+                GenerationTaskModel.status == GenerationTaskStatus.PENDING.value,
+                GenerationTaskModel.created_at < cutoff,
+            )
+            .update(
+                {
+                    GenerationTaskModel.status: GenerationTaskStatus.FAILED.value,
+                    GenerationTaskModel.error_message: "pending timeout: auto cleanup",
+                    GenerationTaskModel.error_info: error_info,
+                    GenerationTaskModel.completed_at: datetime.now(timezone.utc),
+                },
+                synchronize_session=False,
+            )
+        )
+        self.session.commit()
+        return count
@@ -298,6 +298,7 @@ class GenerationTaskModel(Base):
    output_height = Column(Integer, nullable=False, default=720)
    cover_url = Column(String(1000), nullable=False, default="")
    custom_title = Column(String(500), nullable=False, default="")
+    title_config = Column(JSON, nullable=False, default=dict)
    bgm_config = Column(JSON, nullable=False, default=dict)
    extra_meta = Column("metadata", JSON, nullable=False, default=dict)
    logs = Column(Text, nullable=False, default="[]", server_default="[]")
@@ -69,6 +69,7 @@ class CreateGenerationTaskUseCase:
            output_height=command.output_height,
            cover_url=command.cover_url,
            custom_title=command.custom_title,
+            title_config=command.title_config,
        )
        return self.generation_task_repository.create(task)

@@ -121,6 +121,7 @@ class GenerationTask:
    output_height: int = 720
    cover_url: str = ""
    custom_title: str = ""
+    title_config: dict = field(default_factory=dict)
    extra_meta: dict = field(default_factory=dict)
    logs: str = "[]"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -153,6 +154,7 @@ class GenerationTask:
        output_height: int = 720,
        cover_url: str = "",
        custom_title: str = "",
+        title_config: dict | None = None,
        extra_meta: dict | None = None,
    ) -> "GenerationTask":
        if not project_id.strip() and not template_id.strip():
@@ -184,6 +186,7 @@ class GenerationTask:
            output_height=output_height,
            cover_url=cover_url,
            custom_title=custom_title,
+            title_config=dict(title_config) if title_config else {},
            extra_meta=dict(extra_meta) if extra_meta else {},
        )

@@ -364,7 +364,7 @@ class MediaKitClient:
                data = response.json()

                status = data.get("status")
-                if status == "success":
+                if status in ("completed", "success"):
                    result = data.get("result", {})
                    snapshots = result.get("snapshots", [])
                    logger.info(
@@ -0,0 +1,153 @@
+"""封面标题文字叠加（Pillow）— API / Worker 共用。
+
+在封面帧上绘制白色标题文字 + 黑色描边/阴影，支持 CJK 字体和自动换行。
+从已渲染视频抽帧时通常不需要调用（标题已烧录）；
+从源素材抽帧（API E2 兜底）时调用，保证封面带标题。
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# 按优先级查找 CJK 字体（Debian/Ubuntu fonts-noto-cjk 安装路径）
+_FONT_CANDIDATES = (
+    "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
+    "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
+    "/usr/share/fonts/truetype/noto/NotoSansCJK-Bold.ttc",
+    "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
+    "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc",
+)
+
+
+def find_title_font(size: int):
+    """查找可用的 CJK 字体并返回 PIL ImageFont，找不到返回 None。"""
+    try:
+        from PIL import ImageFont
+    except ImportError:
+        return None
+
+    for fp in _FONT_CANDIDATES:
+        if Path(fp).exists():
+            try:
+                return ImageFont.truetype(fp, size=size)
+            except Exception:
+                continue
+    logger.warning("未找到 CJK 字体，标题叠加将使用 PIL 默认字体（中文可能显示为方块）")
+    return ImageFont.load_default()
+
+
+def wrap_title_text(text: str, font, max_width: int) -> list[str]:
+    """按像素宽度对中英文混合文本自动换行，支持显式 \\n。"""
+    lines: list[str] = []
+    current = ""
+    for ch in text:
+        if ch == "\n":
+            if current:
+                lines.append(current)
+            current = ""
+            continue
+        trial = current + ch
+        try:
+            bbox = font.getbbox(trial)
+            width = bbox[2] - bbox[0]
+        except Exception:
+            width = len(trial) * (font.size // 2)
+        if width <= max_width:
+            current = trial
+        else:
+            if current:
+                lines.append(current)
+            current = ch
+    if current:
+        lines.append(current)
+    return lines
+
+
+def apply_title_to_image(
+    image_path: str,
+    title_text: str,
+    *,
+    position: str = "bottom",
+    font_size: Optional[int] = None,
+    margin_ratio: float = 0.06,
+    stroke_width_ratio: float = 0.04,
+) -> Optional[str]:
+    """在图片上绘制标题文字并覆盖保存。
+
+    Args:
+        image_path: 图片路径（处理结果覆盖写回）
+        title_text: 标题文字；为空直接返回 None 表示跳过
+        position: top / center / bottom
+        font_size: 字号，None 时按图片宽度自动计算
+        margin_ratio: 边缘留白占短边比例
+        stroke_width_ratio: 描边宽度占字号比例
+
+    Returns:
+        成功返回 image_path；标题为空或 PIL 不可用返回 None。
+    """
+    if not title_text or not title_text.strip():
+        return None
+
+    try:
+        from PIL import Image, ImageDraw
+    except ImportError:
+        logger.warning("Pillow 未安装，跳过标题叠加: image=%s", image_path)
+        return None
+
+    img = Image.open(image_path).convert("RGB")
+    draw = ImageDraw.Draw(img)
+    img_w, img_h = img.size
+
+    if font_size is None:
+        font_size = max(28, min(72, img_w // 16))
+
+    font = find_title_font(font_size)
+    if font is None:
+        return None
+
+    stroke_width = max(2, int(font_size * stroke_width_ratio))
+    margin = int(min(img_w, img_h) * margin_ratio)
+    max_text_width = img_w - 2 * margin
+
+    lines = wrap_title_text(title_text.strip(), font, max_text_width)
+    if not lines:
+        return None
+
+    line_heights = []
+    for ln in lines:
+        bbox = font.getbbox(ln)
+        line_heights.append(bbox[3] - bbox[1])
+    line_height = max(line_heights) if line_heights else font_size
+    line_gap = int(line_height * 0.3)
+    total_height = len(lines) * line_height + (len(lines) - 1) * line_gap
+
+    if position == "top":
+        y_start = margin
+    elif position == "center":
+        y_start = (img_h - total_height) // 2
+    else:
+        y_start = img_h - total_height - margin
+
+    for i, ln in enumerate(lines):
+        bbox = font.getbbox(ln)
+        line_w = bbox[2] - bbox[0]
+        x = (img_w - line_w) // 2
+        y = y_start + i * (line_height + line_gap)
+        # 阴影
+        draw.text((x + 2, y + 2), ln, font=font, fill=(0, 0, 0))
+        # 白色文字 + 黑色描边
+        draw.text(
+            (x, y),
+            ln,
+            font=font,
+            fill=(255, 255, 255),
+            stroke_width=stroke_width,
+            stroke_fill=(0, 0, 0),
+        )
+
+    img.save(image_path, "JPEG", quality=92)
+    return image_path
@@ -29,3 +29,4 @@ httpx==0.27.2

 # Prometheus monitoring
 prometheus-client==0.21.1
+Pillow==10.4.0
@@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection:

            from packages.domain import EditingMode

-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_candidates = _render_video(
                task_id="test_task_123",
                downloaded_videos=[Path("/tmp/video1.mp4")],
                voice_path=None,
@@ -920,6 +920,73 @@ class TestUploadCoverType:
            call_kwargs = mock_mk.extract_frames.call_args.kwargs
            assert "source-clip.mp4" in call_kwargs["video_url"]

+    def test_e2_passes_plan_title_to_persist_for_overlay(self):
+        """步骤E2：plan.config.title.text 存在时，作为 title_text 传给 _persist_cover_frame 叠加标题。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {"title": {"enabled": True, "text": "我的视频标题"}}
+
+        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/src.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://mk/frame.jpg"}]
+
+        mock_storage = MagicMock()
+        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"
+
+        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])
+
+        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",
+            ) as mock_persist,
+            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",
+                plan_id="plan-title",
+                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"
+            # 标题文字必须透传给持久化函数（用于源素材帧叠加标题）
+            assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题"
+
    def test_step_e_skips_non_video_assets(self):
        """步骤E2：asset_ids 里只有图片素材时，不调用 MediaKit 并返回 400。"""
        from unittest.mock import MagicMock, patch
@@ -0,0 +1,151 @@
+"""GenerationTaskRepository - cleanup_stale_pending 超时 pending 清理单元测试。"""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+    SQLAlchemyGenerationTaskRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.domain import GenerationTask, GenerationTaskStatus
+
+
+def _repository():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return SQLAlchemyGenerationTaskRepository(session), session, engine
+
+
+def _make_task(**kwargs) -> GenerationTask:
+    defaults = dict(
+        project_id="proj-1",
+        asset_library_id="lib-1",
+        created_by_user_id="user-1",
+    )
+    defaults.update(kwargs)
+    return GenerationTask.create(**defaults)
+
+
+# ---------------------------------------------------------------------------
+# cleanup_stale_pending 基本测试
+# ---------------------------------------------------------------------------
+
+
+def test_cleanup_stale_pending_no_tasks_returns_zero():
+    """没有任务时返回 0。"""
+    repo, _, _ = _repository()
+    count = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count == 0
+
+
+def test_cleanup_stale_pending_recent_pending_not_cleaned():
+    """30 分钟内的 pending 任务不被清理。"""
+    repo, _, _ = _repository()
+    task = _make_task()
+    repo.create(task)
+    # 刚创建的 pending 任务不应被清理
+    count = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count == 0
+    assert repo.get(task.id).status == GenerationTaskStatus.PENDING
+
+
+def test_cleanup_stale_pending_old_pending_marked_failed():
+    """超过 30 分钟的 pending 任务被标记为 failed。"""
+    repo, _, engine = _repository()
+    task = _make_task()
+    repo.create(task)
+
+    # 手动把 created_at 改到 1 小时前
+    with engine.connect() as conn:
+        conn.execute(
+            text("UPDATE generation_tasks SET created_at = :ts WHERE id = :id"),
+            {"ts": datetime.now(timezone.utc) - timedelta(hours=1), "id": task.id},
+        )
+        conn.commit()
+
+    count = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count == 1
+
+    saved = repo.get(task.id)
+    assert saved.status == GenerationTaskStatus.FAILED
+    assert saved.error_message == "pending timeout: auto cleanup"
+    assert saved.error_info.get("error_type") == "PendingTimeout"
+    assert "30" in saved.error_info["message"]
+    assert "failed_at" in saved.error_info
+    assert saved.completed_at is not None
+
+
+def test_cleanup_stale_pending_running_not_touched():
+    """running 任务不受影响，只清理 pending。"""
+    repo, _, engine = _repository()
+    task = _make_task()
+    repo.create(task)
+    task.mark_processing()
+    repo.update(task)
+
+    # 回写 created_at 到 1 小时前
+    with engine.connect() as conn:
+        conn.execute(
+            text("UPDATE generation_tasks SET created_at = :ts WHERE id = :id"),
+            {"ts": datetime.now(timezone.utc) - timedelta(hours=1), "id": task.id},
+        )
+        conn.commit()
+
+    count = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count == 0
+    assert repo.get(task.id).status == GenerationTaskStatus.RUNNING
+
+
+def test_cleanup_stale_pending_custom_timeout():
+    """自定义超时时间生效。"""
+    repo, _, engine = _repository()
+    task = _make_task()
+    repo.create(task)
+
+    # 回写 created_at 到 20 分钟前
+    with engine.connect() as conn:
+        conn.execute(
+            text("UPDATE generation_tasks SET created_at = :ts WHERE id = :id"),
+            {"ts": datetime.now(timezone.utc) - timedelta(minutes=20), "id": task.id},
+        )
+        conn.commit()
+
+    # 30 分钟超时：不清理
+    count_30 = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count_30 == 0
+    # 15 分钟超时：清理
+    count_15 = repo.cleanup_stale_pending(timeout_minutes=15)
+    assert count_15 == 1
+    assert repo.get(task.id).status == GenerationTaskStatus.FAILED
+
+
+def test_cleanup_stale_pending_multiple():
+    """批量清理多个超时的 pending 任务。"""
+    repo, _, engine = _repository()
+
+    tasks = []
+    for i in range(5):
+        t = _make_task(project_id=f"proj-{i}")
+        repo.create(t)
+        tasks.append(t)
+
+    # 全部回写 created_at 到 2 小时前
+    with engine.connect() as conn:
+        for t in tasks:
+            conn.execute(
+                text("UPDATE generation_tasks SET created_at = :ts WHERE id = :id"),
+                {"ts": datetime.now(timezone.utc) - timedelta(hours=2), "id": t.id},
+            )
+        conn.commit()
+
+    count = repo.cleanup_stale_pending(timeout_minutes=30)
+    assert count == 5
+    for t in tasks:
+        assert repo.get(t.id).status == GenerationTaskStatus.FAILED
@@ -0,0 +1,58 @@
+"""packages.shared.title_overlay 单元测试。"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from packages.shared.title_overlay import apply_title_to_image, wrap_title_text
+
+
+@pytest.fixture
+def sample_image():
+    """生成一张 640x360 的纯黑测试图片。"""
+    from PIL import Image
+
+    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+    tmp.close()
+    img = Image.new("RGB", (640, 360), color=(0, 0, 0))
+    img.save(tmp.name, "JPEG")
+    yield tmp.name
+    Path(tmp.name).unlink(missing_ok=True)
+
+
+def test_apply_title_to_image_empty_text_returns_none(sample_image):
+    assert apply_title_to_image(sample_image, "") is None
+    assert apply_title_to_image(sample_image, "   ") is None
+
+
+def test_apply_title_to_image_draws_title(sample_image):
+    result = apply_title_to_image(sample_image, "测试标题")
+    assert result == sample_image
+    assert Path(sample_image).exists()
+    assert Path(sample_image).stat().st_size > 0
+
+
+def test_apply_title_to_image_respects_position(sample_image):
+    for pos in ("top", "center", "bottom"):
+        result = apply_title_to_image(sample_image, "位置测试", position=pos)
+        assert result == sample_image
+
+
+def test_wrap_title_text_supports_long_text():
+    from PIL import ImageFont
+
+    font = ImageFont.load_default()
+    lines = wrap_title_text("这是一个比较长的标题需要自动换行处理ABC", font, max_width=40)
+    assert isinstance(lines, list)
+    assert len(lines) >= 1
+
+
+def test_wrap_title_text_respects_explicit_newline():
+    from PIL import ImageFont
+
+    font = ImageFont.load_default()
+    lines = wrap_title_text("第一行\n第二行", font, max_width=10000)
+    assert lines == ["第一行", "第二行"]
@@ -0,0 +1,126 @@
+"""Tests for _writeback_edit_plan_config in generation_tasks route.
+
+覆盖 CI 增量覆盖率不足的代码：
+- generation_tasks.py 行 160-193 (_writeback_edit_plan_config 函数体)
+- generation_tasks.py 行 371-372 (路由中调用该函数)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from app.api.routes.generation_tasks import _writeback_edit_plan_config
+
+
+@pytest.fixture
+def mock_db():
+    """Mock SQLAlchemy Session."""
+    db = MagicMock()
+    db.query.return_value = db
+    db.filter.return_value = db
+    return db
+
+
+@pytest.fixture
+def mock_plan():
+    """Mock EditPlanModel instance."""
+    plan = MagicMock()
+    plan.config = {"existing_key": "existing_value"}
+    return plan
+
+
+class TestWritebackEditPlanConfig:
+    """_writeback_edit_plan_config 全分支覆盖"""
+
+    # ---- 行 160-161: plan_id 为空直接返回 ----
+    def test_empty_plan_id_returns_immediately(self, mock_db):
+        _writeback_edit_plan_config(plan_id="", task_id="task_1", title_config={"text": "hi"}, db=mock_db)
+        mock_db.query.assert_not_called()
+        mock_db.commit.assert_not_called()
+
+    def test_none_plan_id_returns_immediately(self, mock_db):
+        _writeback_edit_plan_config(plan_id=None, task_id="task_1", title_config=None, db=mock_db)
+        mock_db.query.assert_not_called()
+
+    # ---- 行 165-168: plan 不存在 → warning + 不 commit ----
+    def test_plan_not_found_no_commit(self, mock_db):
+        mock_db.first.return_value = None
+
+        _writeback_edit_plan_config(plan_id="plan_999", task_id="task_1", title_config=None, db=mock_db)
+
+        mock_db.query.assert_called_once()
+        mock_db.commit.assert_not_called()
+
+    # ---- 行 170-182: 正常写入 + title_config ----
+    def test_success_with_title_config(self, mock_db, mock_plan):
+        mock_db.first.return_value = mock_plan
+
+        _writeback_edit_plan_config(
+            plan_id="plan_123",
+            task_id="task_456",
+            title_config={"text": "标题", "font_size": 36},
+            db=mock_db,
+        )
+
+        assert mock_plan.config["generation_task_id"] == "task_456"
+        assert mock_plan.config["title_config"] == {"text": "标题", "font_size": 36}
+        assert mock_plan.config["existing_key"] == "existing_value"
+        mock_db.commit.assert_called_once()
+
+    # ---- 行 170-175: 正常写入、无 title_config ----
+    def test_success_without_title_config(self, mock_db, mock_plan):
+        mock_db.first.return_value = mock_plan
+
+        _writeback_edit_plan_config(plan_id="plan_123", task_id="task_789", title_config=None, db=mock_db)
+
+        assert mock_plan.config["generation_task_id"] == "task_789"
+        assert "title_config" not in mock_plan.config
+        mock_db.commit.assert_called_once()
+
+    # ---- 行 170: config 不是 dict → 兜底空 dict ----
+    def test_config_not_dict_uses_empty_dict(self, mock_db):
+        bad_plan = MagicMock()
+        bad_plan.config = "not_a_dict"
+        mock_db.first.return_value = bad_plan
+
+        _writeback_edit_plan_config(plan_id="plan_123", task_id="task_1", title_config=None, db=mock_db)
+
+        assert isinstance(bad_plan.config, dict)
+        assert bad_plan.config["generation_task_id"] == "task_1"
+        mock_db.commit.assert_called_once()
+
+    # ---- 行 183-189: DB 异常 → warning + rollback ----
+    def test_db_exception_triggers_rollback(self, mock_db, mock_plan):
+        mock_db.first.return_value = mock_plan
+        mock_db.commit.side_effect = RuntimeError("DB connection lost")
+
+        # 不应抛异常
+        _writeback_edit_plan_config(plan_id="plan_123", task_id="task_1", title_config=None, db=mock_db)
+
+        mock_db.rollback.assert_called_once()
+
+    # ---- 行 190-193: rollback 也失败 → 静默 ----
+    def test_rollback_failure_silent(self, mock_db, mock_plan):
+        mock_db.first.return_value = mock_plan
+        mock_db.commit.side_effect = RuntimeError("commit failed")
+        mock_db.rollback.side_effect = RuntimeError("rollback also failed")
+
+        # 两个异常都不应抛出
+        _writeback_edit_plan_config(plan_id="plan_123", task_id="task_1", title_config=None, db=mock_db)
+        mock_db.rollback.assert_called_once()
+
+    # ---- 行 173: title_config 为空 dict → 不写入 title_config ----
+    def test_empty_title_config_not_written(self, mock_db, mock_plan):
+        mock_db.first.return_value = mock_plan
+
+        _writeback_edit_plan_config(plan_id="plan_123", task_id="task_1", title_config={}, db=mock_db)
+
+        # 空 dict 为 falsy，不写入
+        assert "title_config" not in mock_plan.config
+        assert mock_plan.config["generation_task_id"] == "task_1"