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）时回写。

alembic/versions/057_add_title_config_to_generation_tasks.py

@@ -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")

apps/api/app/api/routes/generation_tasks.py

@@ -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:

apps/api/app/api/routes/templates_editor/generation.py

@@ -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,

apps/api/app/schemas/generation_task.py

@@ -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

apps/web/src/api/template-editor/editPlans.ts

@@ -50,29 +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 || []
 }
-
-/** ── 草稿 clips 批量更新 ── */
-
-export interface EditPlanClipInput {
-  asset_id: string
-  start_time: number
-  duration: number
-  order: number
-}
-
-/**
- * 批量替换草稿的 clips（先全删再批量插入）
- * 后端路由：PUT /templates/{template_id}/editor/clips
- */
-export async function updateEditPlanClips(
-  templateId: string,
-  clips: EditPlanClipInput[],
-  signal?: AbortSignal,
-): Promise<{ count: number }> {
-  const response = await apiClient.put(
-    `/templates/${templateId}/editor/clips`,
-    { clips },
-    { signal },
-  )
-  return response.data
-}

apps/web/src/api/template-editor/index.ts

@@ -51,13 +51,11 @@ export {
 export {
   getEditPlan,
   updateEditPlan,
-  updateEditPlanClips,
   generateEditPlan,
   getGenerationStatus,
   getEditPlanGenerations,
   getGenerationTaskResults,
 } from "./editPlans"
-export type { EditPlanClipInput } from "./editPlans"
 
 // 片段 CRUD + 批量操作
 export {

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -123,10 +123,6 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
     presetVoices,
   } = props
 
-  /* 当前模板的 segments，传给 Step2 构建 clips */
-  const currentTemplate = userTemplates.find((t) => t.id === selectedTemplate)
-  const templateSegments = currentTemplate?.segments
-
   switch (currentStep) {
     case 1:
       return (
@@ -146,7 +142,6 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           smartSelectedIds={smartSelectedIds}
           onSmartSelectedIdsChange={onSmartSelectedIdsChange}
           selectedTemplate={selectedTemplate}
-          templateSegments={templateSegments}
         />
       )
     case 3:

apps/web/src/pages/generate/components/Step2MaterialSelect.tsx

@@ -2,7 +2,6 @@
  * Step 2 素材选择组件
  */
 import React from "react"
-import type { TemplateSegment } from "@/api/templates/types"
 import { useStep2Materials } from "../hooks/useStep2Materials"
 import MaterialModeTabs from "./material/MaterialModeTabs"
 import ManualMaterialList from "./material/ManualMaterialList"
@@ -18,8 +17,6 @@ interface Step2MaterialSelectProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
-  /** 当前模板的 segments（用于构建 clips duration） */
-  templateSegments?: TemplateSegment[]
 }
 
 const Step2MaterialSelect: React.FC<Step2MaterialSelectProps> = (props) => {

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

@@ -3,11 +3,7 @@
  * 组合素材库加载 + 智能匹配两个子 Hook
  */
 import { useCallback, useEffect, useRef } from "react"
-import type { AssetItem } from "@/api/assets"
-import type { TemplateSegment } from "@/api/templates/types"
-import { updateEditPlanClips } from "@/api/template-editor"
 import { formatDuration } from "../utils/formatDuration"
-import { buildClipsFromAssets } from "../utils/buildClipsFromAssets"
 import { useMaterialLibrary } from "./step2-materials/useMaterialLibrary"
 import { useSmartMatch } from "./step2-materials/useSmartMatch"
 import { useDraftAutoSave } from "./useDraftAutoSave"
@@ -21,8 +17,6 @@ interface UseStep2MaterialsProps {
   onSmartSelectedIdsChange: (ids: string[]) => void
   /** 当前选中的模板/草稿 ID，用于自动保存 */
   selectedTemplate?: string
-  /** 当前模板的 segments（用于构建 clips duration） */
-  templateSegments?: TemplateSegment[]
 }
 
 export function useStep2Materials({
@@ -33,7 +27,6 @@ export function useStep2Materials({
   smartSelectedIds,
   onSmartSelectedIdsChange,
   selectedTemplate,
-  templateSegments,
 }: UseStep2MaterialsProps) {
   const { libraries, selectedLibraryId, setSelectedLibraryId, materials, materialsLoading } =
     useMaterialLibrary()
@@ -64,7 +57,7 @@ export function useStep2Materials({
     handleSmartMatch()
   }, [selectedLibraryId, materialMode, materialsLoading, materials.items, handleSmartMatch])
 
-  /* ── Step2 选择素材后自动保存草稿 asset_ids（防抖 500ms，失败静默） ── */
+  /* ── Step2 选择素材后自动保存草稿（防抖 500ms，失败静默） ── */
   const { scheduleSave } = useDraftAutoSave(selectedTemplate)
   useEffect(() => {
     if (!selectedTemplate) return
@@ -72,61 +65,6 @@ export function useStep2Materials({
     scheduleSave({ asset_ids: ids }, 500)
   }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, scheduleSave])
 
-  /* ── Step2 选择素材后同步写入 edit_plan_clips（防抖 800ms，失败静默） ── */
-  const clipsTimerRef = useRef<ReturnType<typeof setTimeout>>()
-  const clipsAbortRef = useRef<AbortController | null>(null)
-  const templateSegmentsRef = useRef(templateSegments)
-  templateSegmentsRef.current = templateSegments
-  const selectedTemplateRef = useRef(selectedTemplate)
-  selectedTemplateRef.current = selectedTemplate
-  const materialsRef = useRef(materials)
-  materialsRef.current = materials
-  const smartMatchedRef = useRef<AssetItem[]>(smartMatch.smartMatchedResults)
-  smartMatchedRef.current = smartMatch.smartMatchedResults
-
-  useEffect(() => {
-    const tid = selectedTemplateRef.current
-    if (!tid) return
-    const ids = materialMode === "auto" ? smartSelectedIds : selectedMaterials
-    if (!ids.length) return
-
-    if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
-    clipsTimerRef.current = setTimeout(async () => {
-      // 取消上一次未完成的请求
-      if (clipsAbortRef.current) clipsAbortRef.current.abort()
-      const controller = new AbortController()
-      clipsAbortRef.current = controller
-
-      const clips = buildClipsFromAssets({
-        selectedIds: ids,
-        materials: materialsRef.current.items,
-        smartMatchedAssets: smartMatchedRef.current,
-        templateSegments: templateSegmentsRef.current || [],
-      })
-
-      try {
-        await updateEditPlanClips(tid, clips, controller.signal)
-      } catch (err) {
-        const name = (err as { name?: string })?.name
-        if (name !== "CanceledError" && name !== "AbortError") {
-          console.warn("[useStep2Materials] 写入 clips 失败:", err)
-        }
-      }
-    }, 800)
-
-    return () => {
-      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
-    }
-  }, [selectedTemplate, materialMode, selectedMaterials, smartSelectedIds, templateSegments])
-
-  // 组件卸载时取消未完成请求
-  useEffect(() => {
-    return () => {
-      if (clipsTimerRef.current) clearTimeout(clipsTimerRef.current)
-      if (clipsAbortRef.current) clipsAbortRef.current.abort()
-    }
-  }, [])
-
   /* ── 手动选择素材 ── */
   const handleToggleMaterial = useCallback(
     (materialId: string) => {

apps/web/src/pages/generate/utils/buildClipsFromAssets.ts

@@ -1,57 +0,0 @@
-/**
- * 将选中素材 + 模板 segments 构建为 edit_plan_clips 写入数据。
- *
- * 逻辑必须与 FrontendPreviewPlayer.tsx 中 buildPlaybackSegments 完全一致：
- *   assetDuration = asset.duration || asset.metadata?.duration || 30
- *   tplSeg = templateSegments[i] || lastSegment
- *   segDuration = clamp(assetDuration, tplSeg.duration_min, tplSeg.duration_max)
- *   start_time = 0
- *   duration = segDuration
- */
-import type { AssetItem } from "@/api/assets"
-import type { TemplateSegment } from "@/api/templates/types"
-import type { EditPlanClipInput } from "@/api/template-editor"
-
-interface BuildClipsOptions {
-  /** 选中的素材 ID 列表（按选择顺序） */
-  selectedIds: string[]
-  /** 已加载的素材列表（用于查 duration） */
-  materials: AssetItem[]
-  /** 智能匹配返回的素材（auto 模式下可能不在 materials 列表中） */
-  smartMatchedAssets?: AssetItem[]
-  /** 模板 segments */
-  templateSegments?: TemplateSegment[]
-}
-
-export function buildClipsFromAssets({
-  selectedIds,
-  materials,
-  smartMatchedAssets = [],
-  templateSegments = [],
-}: BuildClipsOptions): EditPlanClipInput[] {
-  if (!selectedIds.length) return []
-
-  // 合并两个素材来源，建立 id → asset 索引
-  const assetMap = new Map<string, AssetItem>()
-  for (const a of materials) assetMap.set(a.id, a)
-  for (const a of smartMatchedAssets) assetMap.set(a.id, a)
-
-  const lastSeg = templateSegments[templateSegments.length - 1]
-
-  return selectedIds.map((assetId, i) => {
-    const asset = assetMap.get(assetId)
-    const assetDuration = asset?.duration || asset?.metadata?.duration || 30
-
-    const tplSeg = templateSegments[i] || lastSeg
-    const segDuration = tplSeg
-      ? Math.min(tplSeg.duration_max, Math.max(tplSeg.duration_min, assetDuration))
-      : Math.min(assetDuration, 10)
-
-    return {
-      asset_id: assetId,
-      start_time: 0,
-      duration: segDuration,
-      order: i,
-    }
-  })
-}

apps/worker/worker_app/tasks/generation.py

@@ -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,6 +968,7 @@ def _render_video(
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
     custom_title: str = "",
+    title_config: dict | None = None,
 ) -> tuple[Path, float, list[dict] | None]:
     """渲染视频（含配音混音）。
 
@@ -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 {}
@@ -1401,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:

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -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

packages/adapters/sqlalchemy_impl/models.py

@@ -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="[]")

packages/application/generation_tasks.py

@@ -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)
 

packages/domain/generation_task.py

@@ -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 {},
         )
 

tests/unit/test_writeback_edit_plan_config.py

@@ -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"