diff --git a/alembic/versions/057_add_title_config_to_generation_tasks.py b/alembic/versions/057_add_title_config_to_generation_tasks.py
new file mode 100644
index 000000000..3bad59f88
--- /dev/null
+++ b/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")
diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py
index b290310b6..6547ec4b8 100755
--- a/apps/api/app/api/routes/generation_tasks.py
+++ b/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,13 @@ def create_generation_task(
                     log_task_status=True,
                 ):
                     created_tasks.append(task)
+                    if request.source_edit_plan_id:
+                        _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:
diff --git a/apps/api/app/api/routes/templates_editor/generation.py b/apps/api/app/api/routes/templates_editor/generation.py
index 8c49d946d..7c8009df5 100755
--- a/apps/api/app/api/routes/templates_editor/generation.py
+++ b/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,
diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py
index 5a6d84283..8d27abd86 100755
--- a/apps/api/app/schemas/generation_task.py
+++ b/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
diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py
index 49b0dd53d..e9b02bb75 100644
--- a/apps/worker/worker_app/tasks/generation.py
+++ b/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:
diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py
index 75f394edc..5f3d1d489 100755
--- a/packages/adapters/sqlalchemy_impl/generation_task_repository.py
+++ b/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
diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py
index ce9959931..8e0cdbd55 100755
--- a/packages/adapters/sqlalchemy_impl/models.py
+++ b/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="[]")
diff --git a/packages/application/generation_tasks.py b/packages/application/generation_tasks.py
index 1502208eb..d916314ec 100755
--- a/packages/application/generation_tasks.py
+++ b/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)
 
diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py
index b1f7d0eb0..34e891c18 100755
--- a/packages/domain/generation_task.py
+++ b/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 {},
         )
 
diff --git a/tests/unit/test_writeback_edit_plan_config.py b/tests/unit/test_writeback_edit_plan_config.py
new file mode 100644
index 000000000..e218f07c0
--- /dev/null
+++ b/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"