feat(#2039 ): 爆款视频 DB 模型 + Alembic migration（PR1/2） (#2098 )

ci: retrigger pipeline (previous run stalled on runner)
style: auto-format with black + isort + ruff + prettier [skip ci-format-check]
2026-09-29 22:21:53 +08:00 · 2026-09-29 20:26:21 +08:00 · 2026-09-29 11:54:05 +00:00 · 2026-09-29 19:51:39 +08:00 · 2026-09-29 19:28:12 +08:00 · 2026-09-29 11:17:23 +00:00
21 changed files with 1003 additions and 1073 deletions
@@ -0,0 +1,100 @@
+"""add viral video tables
+
+Revision ID: 086_add_viral_video_tables
+Revises: 085_atom_clip_caption_embedding
+Create Date: 2026-09-28
+
+新增爆款视频相关表：
+- viral_video_jobs: 爆款视频任务
+- viral_video_style_templates: 风格模板配置
+- viral_video_prompt_templates: Prompt 模板（由 #2040 seed）
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "086_add_viral_video_tables"
+down_revision = "085_atom_clip_caption_embedding"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # viral_video_jobs
+    op.create_table(
+        "viral_video_jobs",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("user_id", sa.String(36), nullable=False, index=True),
+        sa.Column("images", sa.JSON(), nullable=False, server_default="[]"),
+        sa.Column("industry", sa.String(100), nullable=False, server_default=""),
+        sa.Column("target_customer", sa.String(500), nullable=False, server_default=""),
+        sa.Column("persona_id", sa.String(36), nullable=False, server_default=""),
+        sa.Column("viral_structure", sa.String(50), nullable=False, server_default=""),
+        sa.Column("marketing_purpose", sa.String(100), nullable=False, server_default=""),
+        sa.Column("bgm_preference", sa.String(50), nullable=False, server_default=""),
+        sa.Column("duration", sa.Integer(), nullable=False, server_default="30"),
+        sa.Column("user_copy_text", sa.Text(), nullable=False, server_default=""),
+        sa.Column("fusion_level", sa.String(20), nullable=False, server_default="ai_polish"),
+        sa.Column("reference_audio_path", sa.String(1000), nullable=False, server_default=""),
+        # v1.3 新增
+        sa.Column("reference_video_url", sa.String(1000), nullable=False, server_default=""),
+        sa.Column("style_strength", sa.String(20), nullable=False, server_default="medium"),
+        sa.Column("style_guide", sa.JSON(), nullable=True),
+        sa.Column("style_template_id", sa.String(36), nullable=False, server_default="", index=True),
+        # 状态与结果
+        sa.Column("status", sa.String(30), nullable=False, server_default="pending", index=True),
+        sa.Column("intent_result", sa.JSON(), nullable=True),
+        sa.Column("result_video_url", sa.String(1000), nullable=False, server_default=""),
+        sa.Column("credits_cost", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("error_msg", sa.Text(), nullable=False, server_default=""),
+        sa.Column("retry_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+
+    # viral_video_style_templates
+    op.create_table(
+        "viral_video_style_templates",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("name", sa.String(200), nullable=False),
+        sa.Column("description", sa.Text(), nullable=False, server_default=""),
+        sa.Column("thumbnail_url", sa.String(1000), nullable=False, server_default=""),
+        sa.Column("style_config", sa.JSON(), nullable=False, server_default="{}"),
+        sa.Column("is_system", sa.Boolean(), nullable=False, server_default=sa.text("true"), index=True),
+        sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+
+    # viral_video_prompt_templates
+    op.create_table(
+        "viral_video_prompt_templates",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("prompt_type", sa.String(50), nullable=False, index=True),
+        sa.Column("name", sa.String(200), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False, server_default=""),
+        sa.Column("variables", sa.JSON(), nullable=False, server_default="[]"),
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("true"), index=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+
+    # Seed 默认风格模板
+    op.execute("""
+        INSERT INTO viral_video_style_templates (id, name, description, style_config, is_system, sort_order)
+        VALUES
+            ('style-tpl-001', '快节奏冲击', '高频切镜+动感BGM，适合食品饮料等快消品', '{"cut_speed": "fast", "transition": "jump_cut", "energy": "high"}', true, 1),
+            ('style-tpl-002', '质感慢镜', '慢节奏+电影感调色，适合美妆护肤珠宝', '{"cut_speed": "slow", "transition": "dissolve", "energy": "low", "color_grade": "cinematic"}', true, 2),
+            ('style-tpl-003', '口播种草', '数字人口播+产品特写穿插', '{"cut_speed": "medium", "transition": "cross_dissolve", "has_talking_head": true}', true, 3),
+            ('style-tpl-004', '场景叙事', '多场景切换+故事线叙述', '{"cut_speed": "medium", "transition": "wipe", "narrative": true}', true, 4)
+        """)
+
+
+def downgrade() -> None:
+    op.drop_table("viral_video_prompt_templates")
+    op.drop_table("viral_video_style_templates")
+    op.drop_table("viral_video_jobs")
@@ -11,9 +11,7 @@ from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
 from app.core.task_enqueue import (
    GLOBAL_PENDING_LIMIT,
-    USER_PENDING_LIMIT,
    GlobalQueueFull,
-    UserPendingLimitExceeded,
    build_rate_limit_detail,
    safe_enqueue_generation_task,
 )
@@ -302,26 +300,17 @@ def create_preview_generation_task(
        count,
    )

-    # 预检查队列限流（按变体总数计）
-    try:
-        user_pending = generation_task_repository.count_pending_by_user(user_id)
-        global_pending = generation_task_repository.count_pending_total()
-        if user_pending + count > USER_PENDING_LIMIT:
-            raise UserPendingLimitExceeded(
-                user_id=user_id, pending_count=user_pending + count, limit=USER_PENDING_LIMIT
-            )
-        if global_pending + count > GLOBAL_PENDING_LIMIT:
-            raise GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT)
-    except UserPendingLimitExceeded as e:
-        raise HTTPException(
-            status_code=429,
-            detail=build_rate_limit_detail(e, generation_task_repository, scope="user"),
-        ) from e
-    except GlobalQueueFull as e:
+    # 预检查队列限流（按变体总数计）——仅保留全局硬上限，用户上限改为软 warning 在 safe_enqueue 内处理（#2098）
+    global_pending = generation_task_repository.count_pending_total()
+    if global_pending + count > GLOBAL_PENDING_LIMIT:
        raise HTTPException(
            status_code=503,
-            detail=build_rate_limit_detail(e, generation_task_repository, scope="global"),
-        ) from e
+            detail=build_rate_limit_detail(
+                GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT),
+                generation_task_repository,
+                scope="global",
+            ),
+        )

    # 确定视频比例：优先前端传入，否则从模板 mode 推断
    video_ratio = request.video_ratio or ""
@@ -589,9 +578,6 @@ def create_preview_generation_task(
            if not enqueued:
                logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id)
                _mark_task_failed(generation_task_repository, task, "任务入队失败")
-        except UserPendingLimitExceeded as e:
-            _mark_task_failed(generation_task_repository, task, "待处理任务超限")
-            rate_limit_exc = rate_limit_exc or e
        except GlobalQueueFull as e:
            _mark_task_failed(generation_task_repository, task, "系统队列已满")
            rate_limit_exc = rate_limit_exc or e
@@ -603,11 +589,6 @@ def create_preview_generation_task(

    # 队列满/限流时若全部失败，返回结构化错误码（前端区分"排队"与"创建失败"）
    if all(r.status == "failed" for r in responses) and rate_limit_exc is not None:
-        if isinstance(rate_limit_exc, UserPendingLimitExceeded):
-            raise HTTPException(
-                status_code=429,
-                detail=build_rate_limit_detail(rate_limit_exc, generation_task_repository, scope="user"),
-            )
        raise HTTPException(
            status_code=503,
            detail=build_rate_limit_detail(rate_limit_exc, generation_task_repository, scope="global"),
@@ -7,7 +7,6 @@ from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import OSSStorageService, get_storage_service
 from app.core.task_enqueue import (
    GLOBAL_PENDING_LIMIT,
-    USER_PENDING_LIMIT,
    GlobalQueueFull,
    UserPendingLimitExceeded,
    build_rate_limit_detail,
@@ -50,12 +49,98 @@ from packages.domain.smart_match import smart_select_assets
 # #2035：文案关键词 → 素材分类 映射表（用于 smart_match category_match 维度）
 # AssetClassification 枚举: scenic / product / person / animal / food / tech / sport / music / other
 _CATEGORY_KEYWORDS: dict[str, set[str]] = {
-    "scenic": {"风景", "自然", "山水", "大海", "天空", "日落", "日出", "森林", "城市", "建筑", "夜景", "街道", "公园", "景区", "旅行", "旅游", "户外"},
-    "product": {"产品", "商品", "展示", "演示", "开箱", "评测", "好物", "推荐", "种草", "购物", "电商", "带货", "品牌", "广告", "包装"},
-    "person": {"人物", "人物采访", "对话", "说话", "讲解", "演讲", "采访", "聊天", "开会", "工作", "办公室", "团队", "员工", "老板", "女性", "男性", "美女", "帅哥"},
+    "scenic": {
+        "风景",
+        "自然",
+        "山水",
+        "大海",
+        "天空",
+        "日落",
+        "日出",
+        "森林",
+        "城市",
+        "建筑",
+        "夜景",
+        "街道",
+        "公园",
+        "景区",
+        "旅行",
+        "旅游",
+        "户外",
+    },
+    "product": {
+        "产品",
+        "商品",
+        "展示",
+        "演示",
+        "开箱",
+        "评测",
+        "好物",
+        "推荐",
+        "种草",
+        "购物",
+        "电商",
+        "带货",
+        "品牌",
+        "广告",
+        "包装",
+    },
+    "person": {
+        "人物",
+        "人物采访",
+        "对话",
+        "说话",
+        "讲解",
+        "演讲",
+        "采访",
+        "聊天",
+        "开会",
+        "工作",
+        "办公室",
+        "团队",
+        "员工",
+        "老板",
+        "女性",
+        "男性",
+        "美女",
+        "帅哥",
+    },
    "animal": {"动物", "宠物", "狗", "猫", "鸟", "鱼", "马", "牛", "羊", "野生动物", "动物园"},
-    "food": {"美食", "食物", "餐饮", "餐厅", "做饭", "烹饪", "厨房", "菜品", "饮料", "水果", "甜点", "蛋糕", "咖啡", "茶", "零食", "吃"},
-    "tech": {"科技", "数码", "电脑", "手机", "屏幕", "软件", "APP", "互联网", "AI", "人工智能", "机器人", "办公", "程序员", "代码", "屏幕录制"},
+    "food": {
+        "美食",
+        "食物",
+        "餐饮",
+        "餐厅",
+        "做饭",
+        "烹饪",
+        "厨房",
+        "菜品",
+        "饮料",
+        "水果",
+        "甜点",
+        "蛋糕",
+        "咖啡",
+        "茶",
+        "零食",
+        "吃",
+    },
+    "tech": {
+        "科技",
+        "数码",
+        "电脑",
+        "手机",
+        "屏幕",
+        "软件",
+        "APP",
+        "互联网",
+        "AI",
+        "人工智能",
+        "机器人",
+        "办公",
+        "程序员",
+        "代码",
+        "屏幕录制",
+    },
    "sport": {"运动", "健身", "跑步", "篮球", "足球", "游泳", "瑜伽", "户外", "锻炼", "体育", "比赛", "球场"},
    "music": {"音乐", "歌曲", "演唱会", "乐器", "唱歌", "跳舞", "舞蹈", "MV", "演出", "乐队", "钢琴", "吉他", "节奏"},
 }
@@ -77,6 +162,7 @@ def _infer_expected_categories(script_tags: set[str] | None) -> set[str] | None:
                break
    return matched or None

+
 from packages.middleware.points_gate import points_gate

 logger = logging.getLogger(__name__)
@@ -193,10 +279,13 @@ def _select_assets_from_library(
    # #2035：加载片段级 AI 标签，供叙事模式 AI 加权和 smart 模式语义匹配使用。
    # 失败降级为空（不影响选片主流程）。
    clip_ai_tags_by_asset: dict[str, list[dict]] = {}
-    ai_tags_by_asset: dict[str, dict] = {}  # asset_id → 聚合后的 ai_tags dict（取首个有 has_text 的片段；合并 scene/objects/action 去重）
+    ai_tags_by_asset: dict[
+        str, dict
+    ] = {}  # asset_id → 聚合后的 ai_tags dict（取首个有 has_text 的片段；合并 scene/objects/action 去重）
    try:
        if db is not None:
            from packages.adapters.sqlalchemy_impl.models import AssetAtomClipModel
+
            ready_ids = [a.id for a in ready_video_assets]
            clip_rows = (
                db.query(AssetAtomClipModel.asset_id, AssetAtomClipModel.ai_tags)
@@ -594,21 +683,13 @@ def create_generation_task(
    # 同批次任务共享 batch_id，用于视频查重时批次内比对
    batch_id = uuid.uuid4().hex if count > 1 else ""

-    # 预检查：批量提交前先看会不会超限，避免建一半才拒
+    # 预检查（Bug B #2098）：只保留全局 503 保护，用户级不再硬拒 429；
+    # 超额任务直接入队等待 worker 自然消费，前端展示排队位置而非阻止提交。
+    # USER_PENDING_LIMIT 作为软上限（safe_enqueue 兜底），提高到 20 支持批量提交。
    try:
-        user_pending = generation_task_repository.count_pending_by_user(user_id)
        global_pending = generation_task_repository.count_pending_total()
-        if user_pending + count > USER_PENDING_LIMIT:
-            raise UserPendingLimitExceeded(
-                user_id=user_id, pending_count=user_pending + count, limit=USER_PENDING_LIMIT
-            )
        if global_pending + count > GLOBAL_PENDING_LIMIT:
            raise GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT)
-    except UserPendingLimitExceeded as e:
-        raise HTTPException(
-            status_code=429,
-            detail=build_rate_limit_detail(e, generation_task_repository, scope="user"),
-        ) from e
    except GlobalQueueFull as e:
        raise HTTPException(
            status_code=503,
@@ -923,14 +1004,10 @@ def create_generation_task(
                else:
                    failed_tasks.append(task)
            except UserPendingLimitExceeded as _e:
-                # 兜底：如果预检查后又并发提交了，在这里也拦住
-                failed_tasks.append(task)
-                if not created_tasks:
-                    raise HTTPException(
-                        status_code=429,
-                        detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"),
-                    ) from _e
-                break
+                # Bug B #2098: 用户级限流已改为软限制，此分支理论上不再触发；
+                # 极端并发兜底仍入队（safe_enqueue 内部会打 warning 日志），不 429 拒绝
+                logger.warning("[生成任务] 用户 pending 超软限制，仍允许入队: task_id=%s", task.id)
+                created_tasks.append(task)
            except GlobalQueueFull as _e:
                failed_tasks.append(task)
                if not created_tasks:
@@ -1081,10 +1158,8 @@ def confirm_generation(
        ):
            logger.warning("[确认生成] 入队失败: task_id=%s", new_task.id)
    except UserPendingLimitExceeded as _e:
-        raise HTTPException(
-            status_code=429,
-            detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"),
-        ) from None
+        # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行
+        logger.warning("[任务] 用户 pending 超软限制，任务已入队")
    except GlobalQueueFull as _e:
        raise HTTPException(
            status_code=503,
@@ -1263,22 +1338,8 @@ def retry_generation_task(
        raise HTTPException(status_code=409, detail="Only failed tasks can be retried")

    user_id = authenticated_user.user.id
-    # 预检查：创建前判断，>= 上限就拒绝
-    user_pending = generation_task_repository.count_pending_by_user(user_id)
+    # 预检查（Bug B #2098）：只保留全局 503，用户级不再硬拒
    global_pending = generation_task_repository.count_pending_total()
-    if user_pending >= USER_PENDING_LIMIT:
-        raise HTTPException(
-            status_code=429,
-            detail=build_rate_limit_detail(
-                UserPendingLimitExceeded(
-                    user_id=user_id,
-                    pending_count=user_pending,
-                    limit=USER_PENDING_LIMIT,
-                ),
-                generation_task_repository,
-                scope="user",
-            ),
-        )
    if global_pending >= GLOBAL_PENDING_LIMIT:
        raise HTTPException(
            status_code=503,
@@ -1321,10 +1382,8 @@ def retry_generation_task(
        ):
            logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id)
    except UserPendingLimitExceeded as _e:
-        raise HTTPException(
-            status_code=429,
-            detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"),
-        ) from None
+        # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行
+        logger.warning("[任务] 用户 pending 超软限制，任务已入队")
    except GlobalQueueFull as _e:
        raise HTTPException(
            status_code=503,
@@ -6,7 +6,7 @@ from app.core.celery_app import celery_app
 logger = logging.getLogger(__name__)

 # ── 限流阈值常量（全系统统一管理，不要在业务代码里硬编码） ──
-USER_PENDING_LIMIT = 3  # 单用户 pending 上限
+USER_PENDING_LIMIT = 20  # 单用户 pending 上限（#2098: 从 3 提到 20，支持批量任务自动排队）
 GLOBAL_PENDING_LIMIT = 20  # 全局 pending 上限
 WORKER_CONCURRENCY = 4  # worker 渲染并发数（infra/docker/compose.yml WORKER_CONCURRENCY 默认值）

@@ -154,19 +154,18 @@ def check_queue_limits(
    user_pending_limit: int = USER_PENDING_LIMIT,
    global_pending_limit: int = GLOBAL_PENDING_LIMIT,
 ) -> None:
-    """检查队列限流（预检查用，任务创建前调用），超限抛对应异常。
+    """检查队列限流（预检查用，任务创建前调用）。

-    边界语义：>= 上限即拒绝（达到上限就不能再加新任务）。
+    #2098 语义变更：用户级限流改为软提示，不再抛异常拒绝；仅全局硬上限抛 GlobalQueueFull。

    Args:
-        user_id: 用户 ID
+        user_id: 用户 ID（保留参数，当前不做用户级硬拒）
        generation_task_repository: 任务仓储
-        user_pending_limit: 单用户 pending 上限，默认 USER_PENDING_LIMIT
+        user_pending_limit: 单用户 pending 上限（保留，当前未硬拒）
        global_pending_limit: 全局 pending 上限，默认 GLOBAL_PENDING_LIMIT

    Raises:
-        GlobalQueueFull: 全局超限时抛出（优先级更高，先查全局）
-        UserPendingLimitExceeded: 用户超限时抛出
+        GlobalQueueFull: 全局超限时抛出
    """
    # 先查全局（系统级保护优先级更高）
    global_pending = generation_task_repository.count_pending_total()
@@ -179,17 +178,9 @@ def check_queue_limits(
        )
        raise GlobalQueueFull(pending_count=global_pending, limit=global_pending_limit)

-    # 再查用户级
-    if user_id:
-        user_pending = generation_task_repository.count_pending_by_user(user_id)
-        if user_pending >= user_pending_limit:
-            logger.warning(
-                "[队列限流] 用户 pending 任务数超限: user_id=%s, count=%d/%d",
-                user_id,
-                user_pending,
-                user_pending_limit,
-            )
-            raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit)
+    # #2098: 用户级限流改为软提示，不在预检查阶段拒绝（超额任务仍入队排队）。
+    # 真正的系统保护由全局 GLOBAL_PENDING_LIMIT 硬上限承担。
+    # UserPendingLimitExceeded 保留以兼容历史 import/except，但预检查与 safe_enqueue 均不再 raise。


 def _mark_task_failed_safely(
@@ -246,7 +237,6 @@ def safe_enqueue_generation_task(

    Raises:
        GlobalQueueFull: 全局 pending 超限时抛出，任务会被标记为 failed
-        UserPendingLimitExceeded: 用户 pending 超限时抛出，任务会被标记为 failed
    """
    # ── 入队前检查：任务已是 pending，用 > 判断（包含当前任务） ──

@@ -263,19 +253,18 @@ def safe_enqueue_generation_task(
        _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc))
        raise exc

-    # 用户级限流检查（传了 user_id 才做）
+    # Bug B #2098: 用户级限流改为软提示，不再硬拒；所有任务都入队等待 worker 自然消费。
+    # user_pending_limit 作为兜底阈值保留（默认 20），达到时打 warning 日志但仍入队，
+    # 避免极端情况下恶意用户无限堆积任务。真正的系统保护由全局 GLOBAL_PENDING_LIMIT 承担。
    if user_id:
        user_pending = generation_task_repository.count_pending_by_user(user_id)
        if user_pending > user_pending_limit:
            logger.warning(
-                "[队列限流] 用户 pending 任务数超限(入队前): user_id=%s, count=%d/%d",
+                "[队列限流] 用户 pending 任务数超过软上限(入队): user_id=%s, count=%d/%d, 仍允许入队排队",
                user_id,
                user_pending,
                user_pending_limit,
            )
-            exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit)
-            _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc))
-            raise exc

    # ── 发送 Celery 任务 ──
    try:
@@ -317,16 +306,18 @@ def safe_enqueue_generation_task(
    user_after = generation_task_repository.count_pending_by_user(user_id) if user_id else 0

    global_over = global_after > global_pending_limit
-    user_over = bool(user_id and user_after > user_pending_limit)
-
-    if global_over or user_over:
-        if global_over:
-            reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}"
-            exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
-        else:
-            reason = f"用户 pending 超限(入队后): {user_after}/{user_pending_limit}"
-            exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_after, limit=user_pending_limit)
+    # Bug B #2098: 用户超限仅日志警告，不回滚任务
+    if user_id and user_after > user_pending_limit:
+        logger.warning(
+            "[队列限流] 用户 pending 超软上限(入队后): user_id=%s, count=%d/%d",
+            user_id,
+            user_after,
+            user_pending_limit,
+        )

+    if global_over:
+        reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}"
+        exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
        logger.warning(
            "[队列限流] %s, task_id=%s, user_id=%s — 回滚状态为 failed",
            reason,
@@ -11,12 +11,7 @@
 *   防止长标题在窄列里溢出导致与相邻卡片进度条视觉重叠。
 */
 import React from "react"
-import {
-  LoadingOutlined,
-  CheckCircleFilled,
-  CloseCircleOutlined,
-  ClockCircleOutlined,
-} from "@ant-design/icons"
+import { LoadingOutlined, CheckCircleFilled, CloseCircleOutlined } from "@ant-design/icons"
 import type { BatchTaskState } from "../hooks/generate-video/useGenerationPolling"
 import type { GeneratedVideo } from "@/api/template-editor"

@@ -66,11 +61,6 @@ const BatchGenerationGrid: React.FC<BatchGenerationGridProps> = ({
                      className="xx-batch-gen-card-icon"
                      style={{ color: "#ef4444" }}
                    />
-                  ) : task.status === "queued" ? (
-                    <ClockCircleOutlined
-                      className="xx-batch-gen-card-icon"
-                      style={{ color: "#faad14" }}
-                    />
                  ) : (
                    <LoadingOutlined
                      className="xx-batch-gen-card-icon"
@@ -95,21 +85,6 @@ const BatchGenerationGrid: React.FC<BatchGenerationGridProps> = ({
                    <div className="xx-batch-gen-card-pct">{Math.round(task.progress)}%</div>
                  </>
                )}
-                {task.status === "queued" && (
-                  <div
-                    style={{
-                      display: "flex",
-                      alignItems: "center",
-                      gap: 8,
-                      color: "var(--text-secondary, #faad14)",
-                      fontSize: 13,
-                      padding: "8px 0",
-                    }}
-                  >
-                    <ClockCircleOutlined />
-                    <span>排队等待中，前面任务完成后自动开始渲染</span>
-                  </div>
-                )}
                {(task.status === "completed" || task.status === "awaiting_cover") && video && (
                  // 竖屏自适应容器（#1750）：成片固定 1080×1920（9:16），
                  // 视频按真实宽高比 contain 显示，黑底居中，杜绝横屏播放器左右大黑边
@@ -12,7 +12,6 @@ import Step2MaterialSelect from "../components/Step2MaterialSelect"
 import Step4TitleSettings from "../components/Step4TitleSettings"
 import Step6CoverSettings from "../components/Step6CoverSettings"
 import BatchGenerationGrid from "./BatchGenerationGrid"
-import Step3VoiceWithMode from "./Step3VoiceWithMode"
 import type { BatchTaskState } from "../hooks/generate-video/useGenerationPolling"
 import type { GeneratedVideo } from "@/api/template-editor"
 import type { TitleTemplate } from "@/components/title/template-types"
@@ -152,12 +151,6 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
    selectedCoverTemplate,
    onSelectedCoverTemplateChange,
    onConfirmGenerate,
-    selectedVoice,
-    onSelectedVoiceChange,
-    voiceModePerVideo,
-    onVoiceModePerVideoChange,
-    voiceLibraryIds,
-    onVoiceLibraryIdsChange,
  } = props

  switch (currentStep) {
@@ -191,46 +184,32 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
      )
    case 3:
      return (
-        <>
-          <Step4TitleSettings
-            titleSettings={titleSettings}
-            onTitleSettingsChange={onTitleSettingsChange}
-            onUpdatePosition={onUpdatePosition}
-            onUpdateFont={onUpdateFont}
-            onUpdateSize={onUpdateSize}
-            onToggleBold={onToggleBold}
-            onToggleItalic={onToggleItalic}
-            onToggleStroke={onToggleStroke}
-            onToggleShadow={onToggleShadow}
-            onApplyPreset={onApplyPreset}
-            onUpdateStyle={onUpdateStyle}
-            activePreset={activePreset}
-            titlePresets={titlePresets}
-            enableTemplates={enableTemplates}
-            selectedTemplateId={selectedTemplateId}
-            onApplyTemplate={onApplyTemplate}
-            previewCount={previewCount}
-            previewTitles={previewTitles}
-            onPreviewTitlesChange={onPreviewTitlesChange}
-            onConfirmGenerate={onConfirmGenerate}
-            generating={props.generating}
-            selectedCount={
-              props.previewCount && props.previewCount > 1
-                ? props.selectedVariantIds?.length || 1
-                : 1
-            }
-          />
-          {/* 批量配音选择：共用/独立切换（#2096） */}
-          <Step3VoiceWithMode
-            previewCount={previewCount}
-            selectedVoice={selectedVoice}
-            onSelectedVoiceChange={onSelectedVoiceChange}
-            voiceModePerVideo={voiceModePerVideo}
-            onVoiceModePerVideoChange={onVoiceModePerVideoChange}
-            voiceLibraryIds={voiceLibraryIds}
-            onVoiceLibraryIdsChange={onVoiceLibraryIdsChange}
-          />
-        </>
+        <Step4TitleSettings
+          titleSettings={titleSettings}
+          onTitleSettingsChange={onTitleSettingsChange}
+          onUpdatePosition={onUpdatePosition}
+          onUpdateFont={onUpdateFont}
+          onUpdateSize={onUpdateSize}
+          onToggleBold={onToggleBold}
+          onToggleItalic={onToggleItalic}
+          onToggleStroke={onToggleStroke}
+          onToggleShadow={onToggleShadow}
+          onApplyPreset={onApplyPreset}
+          onUpdateStyle={onUpdateStyle}
+          activePreset={activePreset}
+          titlePresets={titlePresets}
+          enableTemplates={enableTemplates}
+          selectedTemplateId={selectedTemplateId}
+          onApplyTemplate={onApplyTemplate}
+          previewCount={previewCount}
+          previewTitles={previewTitles}
+          onPreviewTitlesChange={onPreviewTitlesChange}
+          onConfirmGenerate={onConfirmGenerate}
+          generating={props.generating}
+          selectedCount={
+            props.previewCount && props.previewCount > 1 ? props.selectedVariantIds?.length || 1 : 1
+          }
+        />
      )
    case 4:
      return (
@@ -157,7 +157,7 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
          />
        </div>
      ) : (
-        /* ── 批量：N 个独立标题输入框（两列布局 #2096） ── */
+        /* ── 批量：N 个独立标题输入框 ── */
        <div className="xx-batch-titles">
          <div
            style={{
@@ -170,25 +170,17 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
            为每个视频输入独立标题。标题样式（字体/颜色/位置）全局统一。
          </div>

-          <div
-            style={{
-              display: "grid",
-              gridTemplateColumns: "repeat(2, minmax(0, 1fr))",
-              gap: 16,
-            }}
-          >
-            {Array.from({ length: previewCount }, (_, i) => (
-              <div className="xx-form-field" key={i} style={{ maxWidth: "100%" }}>
-                <label>视频 {i + 1} 标题</label>
-                <TitleLibraryAutoComplete
-                  placeholder={`输入或选择视频 ${i + 1} 的标题`}
-                  value={previewTitles?.[i] || ""}
-                  onChange={(val) => updateVariantTitle(i, val)}
-                  options={titleOptions}
-                />
-              </div>
-            ))}
-          </div>
+          {Array.from({ length: previewCount }, (_, i) => (
+            <div className="xx-form-field" key={i} style={{ maxWidth: 640 }}>
+              <label>视频 {i + 1} 标题</label>
+              <TitleLibraryAutoComplete
+                placeholder={`输入或选择视频 ${i + 1} 的标题`}
+                value={previewTitles?.[i] || ""}
+                onChange={(val) => updateVariantTitle(i, val)}
+                options={titleOptions}
+              />
+            </div>
+          ))}
        </div>
      )}

@@ -10,7 +10,7 @@ export interface BatchTaskState {
  taskId: string
  /** 变体序号（0-based，与标题/封面数组对齐） */
  variantIndex: number
-  status: "running" | "completed" | "awaiting_cover" | "failed" | "queued"
+  status: "running" | "completed" | "awaiting_cover" | "failed"
  progress: number
  error: string | null
  /** 完成后的成片视频 */
@@ -374,42 +374,5 @@ export function useGenerationPolling({
    }
  }, [])

-  /**
-   * 批量队列模式：逐任务追加到轮询队列（支持串行提交、429 排队重试场景）。
-   * 与 startPollingBatch 不同的是：
-   * - 不会 reset batchContextRef；多次调用会累积
-   * - 不触发整体 onComplete / onFailed（完成判定交给外层 useEffect 按状态聚合）
-   * - 仍通过 onBatchTaskUpdate 回传单任务状态
-   */
-  const pollBatchTaskQueued = useCallback(
-    (taskId: string, variantIndex: number) => {
-      cancelledRef.current = false
-      batchContextRef.current.set(taskId, variantIndex)
-      onBatchTaskUpdate?.(taskId, {
-        taskId,
-        variantIndex,
-        status: "running",
-        progress: 0,
-        error: null,
-        videos: [],
-      })
-      pollSingleTask(taskId, Date.now(), {
-        onTaskProgress: (pct) => {
-          onBatchTaskUpdate?.(taskId, { status: "running", progress: pct })
-        },
-        onTaskCompleted: (videos, taskStatus) => {
-          const finalStatus: "completed" | "awaiting_cover" = taskStatus ?? "completed"
-          onBatchTaskUpdate?.(taskId, { status: finalStatus, progress: 100, videos })
-        },
-        onTaskFailed: (msg) => {
-          onBatchTaskUpdate?.(taskId, { status: "failed", error: msg })
-        },
-      }).catch(() => {
-        /* onTaskFailed 已处理 */
-      })
-    },
-    [pollSingleTask, onBatchTaskUpdate],
-  )
-
-  return { startPolling, startPollingBatch, pollBatchTaskQueued, retryTask, clearTimer }
+  return { startPolling, startPollingBatch, retryTask, clearTimer }
 }
@@ -2,12 +2,10 @@
 * 视频生成 Hook
 * 封装视频生成的核心逻辑、状态管理、轮询等
 */
-import { useState, useCallback, useEffect, useRef } from "react"
+import { useState, useCallback, useEffect } from "react"
 import { message } from "antd"
-import axios from "axios"
 import { type GeneratedVideo, getEditPlanClips, createClipsFromAssets } from "@/api/template-editor"
 import { createGenerationTask } from "@/api/tasks/tasks"
-import type { CreateGenerationTaskRequest } from "@/api/tasks/types"
 import type { UseGenerateVideoProps } from "./generate-video/types"
 import { getGenerationPhase } from "./generate-video/phase"
 import { useGenerationPolling, type BatchTaskState } from "./generate-video/useGenerationPolling"
@@ -17,26 +15,6 @@ import { extractBackendError, translateError } from "./generate-video/errorUtils

 export type GenerationCompleteStatus = "completed" | "awaiting_cover" | null

-/** 判断是否是用户队列已满 429（需要排队重试而非直接报错） */
-function isUserQueueFullError(err: unknown): { waitMs: number } | null {
-  if (!axios.isAxiosError(err)) return null
-  if (err.response?.status !== 429 && err.response?.status !== 503) return null
-  const detail = (err.response?.data as { detail?: unknown })?.detail
-  const code =
-    typeof detail === "object" && detail !== null ? (detail as { code?: string }).code : undefined
-  if (code === "USER_QUEUE_FULL" || code === "SYSTEM_QUEUE_FULL") {
-    const waitSec =
-      typeof detail === "object" && detail !== null
-        ? Number((detail as { estimated_wait_seconds?: number }).estimated_wait_seconds) || 0
-        : 0
-    return { waitMs: Math.max(15_000, waitSec * 1000 || 30_000) }
-  }
-  return null
-}
-
-/** sleep */
-const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms))
-
 export function useGenerateVideo(props: UseGenerateVideoProps) {
  const { selectedTemplate, onGenerationSuccess } = props

@@ -53,22 +31,6 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
  /** 批量模式：每个正式生成任务的独立状态（第5步逐卡片展示） */
  const [batchTasks, setBatchTasks] = useState<BatchTaskState[]>([])

-  /** 排队中重试的定时器，unmount / 新提交时清理 */
-  const queueTimersRef = useRef<number[]>([])
-  const cancelledRef = useRef(false)
-
-  const clearQueueTimers = useCallback(() => {
-    queueTimersRef.current.forEach((id) => clearTimeout(id))
-    queueTimersRef.current = []
-  }, [])
-
-  useEffect(() => {
-    return () => {
-      cancelledRef.current = true
-      clearQueueTimers()
-    }
-  }, [clearQueueTimers])
-
  const handleBatchTaskUpdate = useCallback((taskId: string, patch: Partial<BatchTaskState>) => {
    setBatchTasks((prev) => {
      const list = prev || []
@@ -96,6 +58,7 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
  const handleProgress = useCallback((p: number) => setProgress(p), [])
  const handleComplete = useCallback(
    (videos: unknown[], taskStatus?: "completed" | "awaiting_cover") => {
+      setGenerating(false)
      setGenerated(true)
      const finalStatus: GenerationCompleteStatus = taskStatus ?? "completed"
      setCompletionStatus(finalStatus)
@@ -118,30 +81,21 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
    [onGenerationSuccess],
  )
  const handleFailed = useCallback((errorMsg: string) => {
+    setGenerating(false)
    setGenerateError(errorMsg)
  }, [])

  /* 批量：任务状态变化时聚合已完成成片（含失败重试成功后补入），
-     按变体索引排序，供步骤6封面按勾选顺序逐个取视频。
-     当全部任务都已结束（completed/awaiting_cover/failed）且无排队/渲染中任务时，关闭 generating。 */
+     按变体索引排序，供步骤6封面按勾选顺序逐个取视频 */
  useEffect(() => {
    if (batchTasks.length === 0) return
    const byVariant = new Map<number, GeneratedVideo>()
-    let hasQueued = false
-    let hasRunning = false
-    let hasSuccess = false
-    let allDone = true
    batchTasks.forEach((t) => {
-      if (t.status === "queued") hasQueued = true
-      else if (t.status === "running") hasRunning = true
-      if (t.status === "completed" || t.status === "awaiting_cover") {
-        hasSuccess = true
-        if (t.videos && t.videos.length > 0) {
-          byVariant.set(t.variantIndex, t.videos[0] as GeneratedVideo)
-        }
-      }
-      if (t.status !== "completed" && t.status !== "awaiting_cover" && t.status !== "failed") {
-        allDone = false
+      if (
+        t.status === "completed" ||
+        (t.status === "awaiting_cover" && t.videos && t.videos.length > 0)
+      ) {
+        byVariant.set(t.variantIndex, t.videos[0] as GeneratedVideo)
      }
    })
    const ordered = [...byVariant.entries()].sort((a, b) => a[0] - b[0]).map(([, v]) => v)
@@ -151,185 +105,17 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
      }
      return ordered
    })
-    if (allDone && !hasQueued && !hasRunning) {
-      setGenerating(false)
-      if (hasSuccess) {
-        setGenerated(true)
-        setCompletionStatus("awaiting_cover")
-      }
-    }
  }, [batchTasks])

-  const { startPolling, pollBatchTaskQueued, retryTask, clearTimer } = useGenerationPolling({
+  const { startPolling, startPollingBatch, retryTask, clearTimer } = useGenerationPolling({
    onProgress: handleProgress,
    onComplete: handleComplete,
    onFailed: handleFailed,
    onBatchTaskUpdate: handleBatchTaskUpdate,
  })

-  /** 根据 props 构造基础 payload（批量/单任务共用的字段） */
-  const buildBasePayload = useCallback((): Omit<
-    CreateGenerationTaskRequest,
-    "count" | "titles" | "voice_library_ids" | "cover_urls" | "variant_plan_ids"
-  > => {
-    const { width: outputWidth, height: outputHeight } = calculateResolution(
-      props.videoRatio || "9:16",
-    )
-    const editMode = props.editMode ?? "random"
-    const dedupEnabled = props.dedupEnabled !== false
-    const assetIds =
-      props.materialMode === "auto" ? props.smartSelectedIds : props.selectedMaterials
-    const coverUrl = props.coverSettings?.thumbnail_url || props.coverSettings?.upload_url || ""
-    // #1970：叙事模式下 ttsVoiceId 作为配音 id；随机模式用 selectedVoice
-    const voiceLibraryId =
-      editMode === "narrative"
-        ? props.ttsVoiceId || ""
-        : props.voiceMode === "clone"
-          ? props.selectedClonedVoice || props.selectedVoice || ""
-          : props.selectedVoice || ""
-
-    const bgmConfig = {
-      enabled: props.bgm !== false,
-      ...(props.bgmConfig?.music_id ? { preset_id: props.bgmConfig.music_id } : {}),
-    }
-
-    const titleConfig = props.titleSettings?.title
-      ? {
-          text: props.titleSettings.title,
-          font: props.titleSettings.font,
-          font_size: props.titleSettings.size,
-          font_color: props.titleSettings.color,
-          position: props.titleSettings.position,
-          ...(props.titleSettings.position === "custom" &&
-          props.titleSettings.posX != null &&
-          props.titleSettings.posY != null
-            ? {
-                pos_x: Math.round(props.titleSettings.posX),
-                pos_y: Math.round(props.titleSettings.posY),
-              }
-            : {}),
-          bold: props.titleSettings.bold,
-          italic: props.titleSettings.italic,
-          stroke: props.titleSettings.stroke
-            ? {
-                enabled: true,
-                width: props.titleSettings.strokeWidth ?? 4,
-                color: props.titleSettings.strokeColor ?? "#000000",
-              }
-            : { enabled: false },
-          shadow: props.titleSettings.shadow
-            ? {
-                enabled: true,
-                offset_x: props.titleSettings.shadowOffsetX ?? 2,
-                offset_y: props.titleSettings.shadowOffsetY ?? 2,
-                blur: props.titleSettings.shadowBlur ?? 4,
-                color: props.titleSettings.shadowColor ?? "rgba(0,0,0,0.8)",
-              }
-            : { enabled: false },
-          line_height: props.titleSettings.lineHeight ?? 1.2,
-          margin_top: props.titleSettings.marginTop ?? 24,
-          max_chars_per_line: props.titleSettings.maxCharsPerLine ?? 0,
-          ...(props.titleSettings.bgEnabled
-            ? {
-                background: {
-                  enabled: true,
-                  color: props.titleSettings.bgColor,
-                  padding: props.titleSettings.bgPadding,
-                  radius: props.titleSettings.bgRadius,
-                },
-              }
-            : { background: { enabled: false } }),
-          line_overrides: (props.titleSettings.lineOverrides ?? []).map((lo) => ({
-            line_index: lo.line_index,
-            text: lo.text,
-            size: lo.size,
-            color: lo.color,
-            bold: lo.bold,
-            italic: lo.italic,
-            stroke: lo.stroke,
-            highlights: lo.highlights?.map((h) => ({
-              word: h.word,
-              color: h.color,
-              bold: h.bold,
-              scale: h.scale,
-            })),
-          })),
-          ...(props.titleSettings.coverTitle
-            ? {
-                cover_title_config: {
-                  title: props.titleSettings.coverTitle.title,
-                  font: props.titleSettings.coverTitle.font,
-                  font_size: props.titleSettings.coverTitle.size,
-                  font_color: props.titleSettings.coverTitle.color,
-                  bold: props.titleSettings.coverTitle.bold,
-                  italic: props.titleSettings.coverTitle.italic,
-                  position: props.titleSettings.coverTitle.position,
-                  stroke: props.titleSettings.coverTitle.stroke
-                    ? {
-                        enabled: true,
-                        width: props.titleSettings.coverTitle.strokeWidth ?? 4,
-                        color: props.titleSettings.coverTitle.strokeColor ?? "#000000",
-                      }
-                    : { enabled: false },
-                  shadow: props.titleSettings.coverTitle.shadow
-                    ? {
-                        enabled: true,
-                        offset_x: props.titleSettings.coverTitle.shadowOffsetX ?? 2,
-                        offset_y: props.titleSettings.coverTitle.shadowOffsetY ?? 2,
-                        blur: props.titleSettings.coverTitle.shadowBlur ?? 4,
-                        color: props.titleSettings.coverTitle.shadowColor ?? "rgba(0,0,0,0.8)",
-                      }
-                    : { enabled: false },
-                  ...(props.titleSettings.coverTitle.bgEnabled
-                    ? {
-                        background: {
-                          enabled: true,
-                          color: props.titleSettings.coverTitle.bgColor,
-                          padding: props.titleSettings.coverTitle.bgPadding,
-                          radius: props.titleSettings.coverTitle.bgRadius,
-                        },
-                      }
-                    : { background: { enabled: false } }),
-                },
-              }
-            : {}),
-        }
-      : undefined
-
-    const payload: Omit<
-      CreateGenerationTaskRequest,
-      "count" | "titles" | "voice_library_ids" | "cover_urls" | "variant_plan_ids"
-    > = {
-      template_id: selectedTemplate,
-      asset_ids: assetIds,
-      output_width: outputWidth,
-      output_height: outputHeight,
-      cover_url: coverUrl,
-      custom_title: props.titleSettings?.title || "",
-      duration: props.duration || undefined,
-      video_ratio: props.videoRatio,
-      assembly_mode: editMode,
-      ...(editMode === "narrative" && props.selectedScript?.id
-        ? {
-            script_id: props.selectedScript.id,
-            tts_voice_id: props.ttsVoiceId || undefined,
-            tts_voice_source: props.ttsVoiceSource || undefined,
-            tts_style: props.ttsStyle || undefined,
-          }
-        : {}),
-      dedup_enabled: dedupEnabled,
-      voice_library_id: voiceLibraryId,
-      ...(props.selectedVoice && !voiceLibraryId ? { voice_ids: [props.selectedVoice] } : {}),
-      bgm_config: bgmConfig as CreateGenerationTaskRequest["bgm_config"],
-      ...(props.sourceEditPlanId ? { source_edit_plan_id: props.sourceEditPlanId } : {}),
-      ...(titleConfig ? ({ title_config: titleConfig } as Record<string, unknown>) : {}),
-    }
-
-    return payload
-  }, [props, selectedTemplate])
-
  /* ── 生成视频 ──
-     返回 true 表示任务创建成功并已开始轮询（含排队中）；false 表示校验未通过或创建失败 */
+     返回 true 表示任务创建成功并已开始轮询；false 表示校验未通过或创建失败 */
  const generate = useCallback(async (): Promise<boolean> => {
    const errorMsg = validateGenerateInputs(props)
    if (errorMsg) {
@@ -337,8 +123,6 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
      return false
    }

-    cancelledRef.current = false
-    clearQueueTimers()
    setGenerating(true)
    setProgress(0)
    setGenerated(false)
@@ -349,16 +133,25 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
    setCurrentTaskId("")
    clearTimer()

-    const basePayload = buildBasePayload()
-    const assetIds = basePayload.asset_ids
-    const isBatch = (props.previewCount || 1) > 1
-
    try {
-      // from-assets 兜底：片段不存在则补一次
+      const { width: outputWidth, height: outputHeight } = calculateResolution(
+        props.videoRatio || "9:16",
+      )
+      const editMode = props.editMode ?? "random"
+      const dedupEnabled = props.dedupEnabled !== false
+
+      const assetIds =
+        props.materialMode === "auto" ? props.smartSelectedIds : props.selectedMaterials
+
+      // from-assets 已由 useStep2Materials 在用户选素材时（debounce 800ms）调用，
+      // 后端已改为异步秒级返回，这里做一次轻量兜底：
+      // 单次查 clips，已有则直接放行；没有则再调一次 from-assets。
      if (assetIds.length > 0 && selectedTemplate) {
        try {
          const clipList = await getEditPlanClips(selectedTemplate, { limit: 500 })
          if (clipList.items.length === 0) {
+            // 片段不存在（极端情况：useStep2Materials 的 debounce 还没触发）
+            // 手动补一次 from-assets（后端秒级返回）
            await createClipsFromAssets(selectedTemplate, assetIds, "main")
          }
        } catch {
@@ -366,185 +159,221 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
        }
      }

-      if (!isBatch) {
-        /* ── 单视频：原逻辑（一次提交 count=1） ── */
-        const hide = message.loading("正在生成预览视频...", 0)
-        try {
-          const taskResp = await createGenerationTask({ ...basePayload, count: 1 })
-          hide()
-          const taskIds = (taskResp.items || []).map((it) => it.id).filter(Boolean)
-          if (taskIds.length === 0) {
-            throw new Error("创建任务成功但未返回任务 ID，请稍后在任务列表查看")
-          }
+      const isBatch = (props.previewCount || 1) > 1
+      const hide = message.loading(
+        isBatch ? `正在生成 ${props.previewCount} 个视频...` : "正在生成预览视频...",
+        0,
+      )
+
+      const coverUrl = props.coverSettings?.thumbnail_url || props.coverSettings?.upload_url || ""
+
+      // #1970：叙事模式下 ttsVoiceId 作为配音 id；随机模式用 selectedVoice
+      const voiceLibraryId =
+        editMode === "narrative"
+          ? props.ttsVoiceId || ""
+          : props.voiceMode === "clone"
+            ? props.selectedClonedVoice || props.selectedVoice || ""
+            : props.selectedVoice || ""
+
+      /* ── 批量变体数组（长度1=共用，长度=count=独立，空=回退单值） ── */
+      const indexes =
+        isBatch && props.selectedVariantIndexes?.length
+          ? props.selectedVariantIndexes
+          : Array.from({ length: props.previewCount || 1 }, (_, i) => i)
+      const batchCount = isBatch ? indexes.length : 1
+
+      // 标题文字数组：批量时按勾选顺序
+      const titlesArr =
+        isBatch && (props.variantTitles?.length || 0) >= batchCount
+          ? indexes.map((i) => props.variantTitles![i] || props.titleSettings?.title || "")
+          : []
+      // 配音数组：独立配音模式按勾选顺序；否则不传（回退共用 voice_library_id）
+      const voiceArr =
+        isBatch && props.voiceModePerVideo && props.variantVoiceLibraryIds?.length
+          ? indexes.map((i) => props.variantVoiceLibraryIds![i] || voiceLibraryId)
+          : []
+      // 封面数组：批量时按勾选顺序（未设置封面的变体传空串，后端回退智能封面）
+      const coversArr =
+        isBatch && props.variantCoverUrls?.length
+          ? indexes.map((i) => props.variantCoverUrls![i] || "")
+          : []
+      // #1744 变体 plan 数组：预览阶段后端独立选片产出的 plan id，按勾选顺序回传，
+      // 后端直接关联这些 plan 渲染（不再重新选片）→ 预览所见即成片。
+      // 全部为空（降级本地模拟/后端端点未上线）时不传，后端走自身独立选片。
+      const variantPlansArr =
+        isBatch && props.variantPlanIds?.length
+          ? indexes.map((i) => props.variantPlanIds![i] || "")
+          : []
+      const hasVariantPlans = variantPlansArr.some((id) => !!id)
+
+      try {
+        const taskResp = await createGenerationTask({
+          template_id: selectedTemplate,
+          asset_ids: assetIds,
+          output_width: outputWidth,
+          output_height: outputHeight,
+          cover_url: coverUrl,
+          custom_title: props.titleSettings?.title || "",
+          duration: props.duration || undefined,
+          video_ratio: props.videoRatio,
+          assembly_mode: editMode,
+          ...(editMode === "narrative" && props.selectedScript?.id
+            ? {
+                script_id: props.selectedScript.id,
+                tts_voice_id: props.ttsVoiceId || undefined,
+                tts_voice_source: props.ttsVoiceSource || undefined,
+                tts_style: props.ttsStyle || undefined,
+              }
+            : {}),
+          dedup_enabled: dedupEnabled,
+          voice_library_id: voiceLibraryId,
+          ...(props.selectedVoice && !voiceLibraryId ? { voice_ids: [props.selectedVoice] } : {}),
+          bgm_config: {
+            enabled: props.bgm !== false,
+            ...(props.bgmConfig?.music_id ? { preset_id: props.bgmConfig.music_id } : {}),
+          },
+          ...(props.sourceEditPlanId ? { source_edit_plan_id: props.sourceEditPlanId } : {}),
+          ...(isBatch ? { count: batchCount } : {}),
+          ...(titlesArr.length ? { titles: titlesArr } : {}),
+          ...(voiceArr.length ? { voice_library_ids: voiceArr } : {}),
+          ...(coversArr.length ? { cover_urls: coversArr } : {}),
+          ...(hasVariantPlans ? { variant_plan_ids: variantPlansArr } : {}),
+          ...(props.titleSettings?.title
+            ? {
+                title_config: {
+                  text: props.titleSettings.title,
+                  font: props.titleSettings.font,
+                  font_size: props.titleSettings.size,
+                  font_color: props.titleSettings.color,
+                  position: props.titleSettings.position,
+                  ...(props.titleSettings.position === "custom" &&
+                  props.titleSettings.posX != null &&
+                  props.titleSettings.posY != null
+                    ? {
+                        pos_x: Math.round(props.titleSettings.posX),
+                        pos_y: Math.round(props.titleSettings.posY),
+                      }
+                    : {}),
+                  bold: props.titleSettings.bold,
+                  italic: props.titleSettings.italic,
+                  stroke: props.titleSettings.stroke
+                    ? {
+                        enabled: true,
+                        width: props.titleSettings.strokeWidth ?? 4,
+                        color: props.titleSettings.strokeColor ?? "#000000",
+                      }
+                    : { enabled: false },
+                  shadow: props.titleSettings.shadow
+                    ? {
+                        enabled: true,
+                        offset_x: props.titleSettings.shadowOffsetX ?? 2,
+                        offset_y: props.titleSettings.shadowOffsetY ?? 2,
+                        blur: props.titleSettings.shadowBlur ?? 4,
+                        color: props.titleSettings.shadowColor ?? "rgba(0,0,0,0.8)",
+                      }
+                    : { enabled: false },
+                  line_height: props.titleSettings.lineHeight ?? 1.2,
+                  margin_top: props.titleSettings.marginTop ?? 24,
+                  max_chars_per_line: props.titleSettings.maxCharsPerLine ?? 0,
+                  ...(props.titleSettings.bgEnabled
+                    ? {
+                        background: {
+                          enabled: true,
+                          color: props.titleSettings.bgColor,
+                          padding: props.titleSettings.bgPadding,
+                          radius: props.titleSettings.bgRadius,
+                        },
+                      }
+                    : { background: { enabled: false } }),
+                  line_overrides: (props.titleSettings.lineOverrides ?? []).map((lo) => ({
+                    line_index: lo.line_index,
+                    text: lo.text,
+                    size: lo.size,
+                    color: lo.color,
+                    bold: lo.bold,
+                    italic: lo.italic,
+                    stroke: lo.stroke,
+                    highlights: lo.highlights?.map((h) => ({
+                      word: h.word,
+                      color: h.color,
+                      bold: h.bold,
+                      scale: h.scale,
+                    })),
+                  })),
+                  ...(props.titleSettings.coverTitle
+                    ? {
+                        cover_title_config: {
+                          title: props.titleSettings.coverTitle.title,
+                          font: props.titleSettings.coverTitle.font,
+                          font_size: props.titleSettings.coverTitle.size,
+                          font_color: props.titleSettings.coverTitle.color,
+                          bold: props.titleSettings.coverTitle.bold,
+                          italic: props.titleSettings.coverTitle.italic,
+                          position: props.titleSettings.coverTitle.position,
+                          stroke: props.titleSettings.coverTitle.stroke
+                            ? {
+                                enabled: true,
+                                width: props.titleSettings.coverTitle.strokeWidth ?? 4,
+                                color: props.titleSettings.coverTitle.strokeColor ?? "#000000",
+                              }
+                            : { enabled: false },
+                          shadow: props.titleSettings.coverTitle.shadow
+                            ? {
+                                enabled: true,
+                                offset_x: props.titleSettings.coverTitle.shadowOffsetX ?? 2,
+                                offset_y: props.titleSettings.coverTitle.shadowOffsetY ?? 2,
+                                blur: props.titleSettings.coverTitle.shadowBlur ?? 4,
+                                color:
+                                  props.titleSettings.coverTitle.shadowColor ?? "rgba(0,0,0,0.8)",
+                              }
+                            : { enabled: false },
+                          ...(props.titleSettings.coverTitle.bgEnabled
+                            ? {
+                                background: {
+                                  enabled: true,
+                                  color: props.titleSettings.coverTitle.bgColor,
+                                  padding: props.titleSettings.coverTitle.bgPadding,
+                                  radius: props.titleSettings.coverTitle.bgRadius,
+                                },
+                              }
+                            : { background: { enabled: false } }),
+                        },
+                      }
+                    : {}),
+                },
+              }
+            : {}),
+        })
+        hide()
+        const taskIds = (taskResp.items || []).map((it) => it.id).filter(Boolean)
+
+        if (taskIds.length === 0) {
+          throw new Error("创建任务成功但未返回任务 ID，请稍后在任务列表查看")
+        }
+        if (taskIds.length > 1) {
+          // 批量：任务按创建顺序与勾选变体一一对应（后端按 count 顺序创建）
+          setCurrentTaskId("")
+          startPollingBatch(taskIds.map((taskId, i) => ({ taskId, variantIndex: indexes[i] ?? i })))
+        } else {
          setCurrentTaskId(taskIds[0])
          startPolling(taskIds[0])
-        } catch (err) {
-          hide()
-          throw err
        }
-        return true
+      } catch (err) {
+        hide()
+        throw err
      }
-
-      /* ── 批量：支持任意数量视频，按队列容量串行提交，429 自动排队重试 ── */
-      const indexes = props.selectedVariantIndexes?.length
-        ? props.selectedVariantIndexes
-        : Array.from({ length: props.previewCount || 1 }, (_, i) => i)
-      const batchCount = indexes.length
-
-      const titlesAll =
-        (props.variantTitles?.length || 0) >= batchCount
-          ? indexes.map((i) => props.variantTitles![i] || props.titleSettings?.title || "")
-          : indexes.map(() => props.titleSettings?.title || "")
-      const voiceArrAll =
-        props.voiceModePerVideo && props.variantVoiceLibraryIds?.length
-          ? indexes.map(
-              (i) => props.variantVoiceLibraryIds![i] || basePayload.voice_library_id || "",
-            )
-          : []
-      const coversAll = props.variantCoverUrls?.length
-        ? indexes.map((i) => props.variantCoverUrls![i] || "")
-        : indexes.map(() => "")
-      const plansAll = props.variantPlanIds?.length
-        ? indexes.map((i) => props.variantPlanIds![i] || "")
-        : indexes.map(() => "")
-
-      const hasAnyVoice = voiceArrAll.some((v) => !!v)
-      const hasAnyCover = coversAll.some((u) => !!u)
-      const hasAnyPlan = plansAll.some((id) => !!id)
-
-      // 先用占位 ID 把所有变体卡片置为 queued，UI 可见
-      const placeholderIds = indexes.map((_, i) => `__queued_${Date.now()}_${i}`)
-      const initialTasks: BatchTaskState[] = indexes.map((variantIndex, i) => ({
-        taskId: placeholderIds[i],
-        variantIndex,
-        status: "queued",
-        progress: 0,
-        error: null,
-        videos: [],
-      }))
-      setBatchTasks(initialTasks)
-
-      message.loading({
-        content: `已提交 ${batchCount} 个视频任务，系统按队列容量依次渲染…`,
-        key: "batch-gen",
-        duration: 3,
-      })
-
-      /** 将占位 taskId 更新为真实 taskId（卡片引用同一对象） */
-      const replacePlaceholder = (placeholderId: string, realTaskId: string) => {
-        setBatchTasks((prev) => {
-          const idx = prev.findIndex((t) => t.taskId === placeholderId)
-          if (idx === -1) return prev
-          const next = [...prev]
-          next[idx] = { ...next[idx], taskId: realTaskId }
-          return next
-        })
-      }
-
-      /** 提交某一索引的单任务（count=1），成功后返回真实 taskId；429/503 则返回 waitMs */
-      const submitOne = async (
-        i: number,
-      ): Promise<{ queued: true; waitMs: number } | { queued: false; taskId: string }> => {
-        const body: CreateGenerationTaskRequest = {
-          ...basePayload,
-          count: 1,
-          titles: [titlesAll[i] || ""],
-          ...(hasAnyVoice
-            ? { voice_library_ids: [voiceArrAll[i] || basePayload.voice_library_id || ""] }
-            : {}),
-          ...(hasAnyCover ? { cover_urls: [coversAll[i] || ""] } : {}),
-          ...(hasAnyPlan && plansAll[i] ? { variant_plan_ids: [plansAll[i]] } : {}),
-        }
-        try {
-          const resp = await createGenerationTask(body)
-          const item = resp.items?.[0]
-          const tid = item?.id
-          if (!tid) throw new Error("创建任务成功但未返回任务 ID")
-          return { queued: false, taskId: tid }
-        } catch (err) {
-          const q = isUserQueueFullError(err)
-          if (q) return { queued: true, waitMs: q.waitMs }
-          throw err
-        }
-      }
-
-      // 串行提交：每次提交一个；429/503 则等待后重试；其它错误立即标记该任务失败
-      let fatalErr: unknown = null
-      for (let i = 0; i < batchCount; i++) {
-        if (cancelledRef.current) return false
-        const variantIndex = indexes[i]
-        const placeholderId = placeholderIds[i]
-        let attempt = 0
-        let submitted = false
-        while (!submitted) {
-          if (cancelledRef.current) return false
-          attempt++
-          try {
-            const result = await submitOne(i)
-            if (!result.queued) {
-              replacePlaceholder(placeholderId, result.taskId)
-              // 先更新到 running，再启动单任务增量轮询（不触发整体 onComplete）
-              pollBatchTaskQueued(result.taskId, variantIndex)
-              submitted = true
-            } else {
-              // 排队：保持 queued 状态，等待后重试
-              handleBatchTaskUpdate(placeholderId, {
-                taskId: placeholderId,
-                variantIndex,
-                status: "queued",
-                progress: 0,
-                error: null,
-              })
-              if (attempt === 1) {
-                message.info({
-                  content: `队列繁忙，${Math.round(result.waitMs / 1000)} 秒后自动继续提交后续视频…`,
-                  key: "batch-gen",
-                  duration: 4,
-                })
-              }
-              await sleep(Math.min(result.waitMs, 60_000))
-            }
-          } catch (err) {
-            // 非限流错误：该任务标记失败，继续后续任务（不阻断整个批量）
-            console.error("[batch generate] 任务提交失败:", err)
-            const msg = translateError(extractBackendError(err))
-            handleBatchTaskUpdate(placeholderId, {
-              taskId: placeholderId,
-              variantIndex,
-              status: "failed",
-              error: msg,
-              progress: 0,
-            })
-            submitted = true
-            if (!fatalErr) fatalErr = err
-          }
-        }
-      }
-
-      if (fatalErr) {
-        // 有任务失败但其余已成功，整体不 throw；由 UI 展示单个失败卡片
-      }
-      return true
    } catch (err: unknown) {
      console.error("[handleGenerate] 生成失败:", err)
+      setGenerating(false)
      const backendMsg = extractBackendError(err)
      console.error("[handleGenerate] 错误信息:", backendMsg, "完整错误:", err)
      const finalMsg = translateError(backendMsg)
      setGenerateError(finalMsg)
-      setGenerating(false)
      message.error(finalMsg)
      return false
    }
-  }, [
-    props,
-    clearTimer,
-    startPolling,
-    selectedTemplate,
-    buildBasePayload,
-    handleBatchTaskUpdate,
-    clearQueueTimers,
-    pollBatchTaskQueued,
-  ])
+    return true
+  }, [props, clearTimer, startPolling, startPollingBatch, selectedTemplate])

  const retry = useCallback(() => {
    setGenerateError(null)
@@ -554,10 +383,9 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
  /** 第5步：单独重试某个失败任务 */
  const retryBatchTask = useCallback(
    (taskId: string) => {
-      handleBatchTaskUpdate(taskId, { status: "running", progress: 0, error: null, videos: [] })
      retryTask(taskId)
    },
-    [retryTask, handleBatchTaskUpdate],
+    [retryTask],
  )

  const dismissError = useCallback(() => {
@@ -30,11 +30,16 @@ EDGE_CROP_MAX_PCT = 0.05
 # 均以 720p 为基准（见前端 titleCanvas.ts 注释 scale=videoWidth/720，types.ts "px @720p"），
 # 非 720p 输出时按 video_width / TITLE_SIZE_REF_WIDTH 等比缩放，保证成片位置与前端预览一致。
 TITLE_SIZE_REF_WIDTH = 720
-# 与前端 titleCanvas.ts 对齐：top 时文本 top-edge 距视频顶 = PAD(16@720p) + margin_top(默认 24@720p)
-TITLE_PAD_TOP = 16
-TITLE_DEFAULT_MARGIN_TOP = 24
-TITLE_DEFAULT_MARGIN_BOTTOM = 24
-SUBTITLE_DEFAULT_MARGIN_BOTTOM = 60  # 字幕距底边距（720p 基准，与前端字幕面板默认对齐）
+# 与 video_filter_builder.build_title_drawtext_filter（CPU 路径）和 ass_subtitle_builder 对齐：
+# - top/bottom 默认 margin 50@720p（vfb 用 _scale_title_len(50, w)，即 y=50 / y=h-th-50）
+# - margin_top 字段：前端编辑器 marginTop 滑块，叠加在默认 margin 之上（#2095 支持）
+# - PAD 概念仅用于前端 Canvas 预览；ffmpeg drawtext y 是 baseline，无 font metrics 可用，
+#   直接用统一 50@720p baseline 位置即可保持三端（GPU/CPU/前端视觉）一致。
+TITLE_DEFAULT_MARGIN_TOP = 50  # top 位置 baseline 默认距顶 50@720p（与 vfb/CPU 路径一致）
+TITLE_DEFAULT_MARGIN_BOTTOM = 50  # bottom 位置 baseline 默认距底 50@720p
+SUBTITLE_DEFAULT_MARGIN_BOTTOM = 50  # 字幕距底边距 50@720p（与 vfb 一致）
+TITLE_MARGIN_TOP_FROM_CFG_DEFAULT = 24  # 前端 marginTop 滑块默认值（用户未传时叠加 0）
+TITLE_FAUX_BOLD_WIDTH = 2  # 仿粗黑色描边宽度（与 vfb 一致，2@720p 黑色细描边）


 def _scale_title_len(value, video_width: int):
@@ -507,7 +512,7 @@ def build_direct_render(
    except (TypeError, ValueError):
        t_size_720 = 0
    if t_size_720 <= 0:
-        t_size_720 = 36  # 与 config_schemas 默认 48 接近；AI Avatar 前端默认 48，剪辑前端默认 28，取中
+        t_size_720 = 48  # 与 config_schemas.DEFAULT_EDIT_PLAN_CONFIG.title.size=48 及 vfb 默认一致
    t_size = _scale_title_len(t_size_720, output_width)
    # stroke/shadow 长度字段也需 720p→输出分辨率缩放
    t_color = str(t_cfg.get("color", t_cfg.get("font_color", "#ffffff")))
@@ -525,14 +530,22 @@ def build_direct_render(
    # baseline 需再下移约 0.85*fontsize；但 drawtext 表达式无法引用 fontsize 变量，
    # 这里直接用 (PAD + margin_top)@720p 缩放后作为 y（即让 baseline≈顶部内边距位置），
    # 实际中文字符会自然向下延伸，视觉位置与前端预览（textBaseline=middle 居中到 firstLineY）一致。
-    _t_margin_top_raw = t_cfg.get("margin_top", TITLE_DEFAULT_MARGIN_TOP)
+    # margin_top：前端滑块值（默认 24@720p），叠加在默认 50@720p 基线之上
+    _t_user_margin_top = t_cfg.get("margin_top")
    try:
-        _t_margin_top_720 = int(_t_margin_top_raw)
+        _t_user_margin_top_720 = int(_t_user_margin_top) if _t_user_margin_top is not None else 0
    except (TypeError, ValueError):
-        _t_margin_top_720 = TITLE_DEFAULT_MARGIN_TOP
-    t_margin_top = _scale_title_len(TITLE_PAD_TOP + _t_margin_top_720, output_width)
-    # bottom margin（标题放在 bottom 时也支持）
-    t_margin_bottom = _scale_title_len(TITLE_PAD_TOP + TITLE_DEFAULT_MARGIN_BOTTOM, output_width)
+        _t_user_margin_top_720 = 0
+    t_margin_top_720 = TITLE_DEFAULT_MARGIN_TOP + _t_user_margin_top_720
+    t_margin_top = _scale_title_len(t_margin_top_720, output_width)
+    # bottom margin（标题放在 bottom 时）：用户 margin_bottom 透传，默认 50@720p
+    _t_user_margin_bottom = t_cfg.get("margin_bottom")
+    try:
+        _t_user_margin_bottom_720 = int(_t_user_margin_bottom) if _t_user_margin_bottom is not None else 0
+    except (TypeError, ValueError):
+        _t_user_margin_bottom_720 = 0
+    t_margin_bottom_720 = TITLE_DEFAULT_MARGIN_BOTTOM + _t_user_margin_bottom_720
+    t_margin_bottom = _scale_title_len(t_margin_bottom_720, output_width)
    t_borderw = 0
    t_border_color = "#000000"
    t_box = False
@@ -566,11 +579,11 @@ def build_direct_render(
    t_shadow_x = _scale_title_len(t_shadow_x_720, output_width)
    t_shadow_y = _scale_title_len(t_shadow_y_720, output_width)
    # bold/italic：drawtext 原生无粗斜体选项；通过同色描边模拟粗体
-    t_bold = bool(t_cfg.get("bold", True))  # 与 ASS 路径默认 bold=True 对齐
+    t_bold = bool(t_cfg.get("bold", True))  # 与 ASS/vfb 路径默认 bold=True 对齐
    if t_bold and t_borderw < 1:
-        # 同色描边宽度 1@720p，按宽度缩放保证各分辨率视觉一致
-        t_borderw = max(1, _scale_title_len(1, output_width))
-        t_border_color = t_color  # 用文字色描边模拟加粗
+        # 粗体未配用户描边时：黑色细描边 2@720p（与 vfb 一致，避免同色描边导致重影）
+        t_borderw = _scale_title_len(TITLE_FAUX_BOLD_WIDTH, output_width)
+        t_border_color = "#000000"  # 黑色细描边模拟粗体

    # ── 解析 subtitle_config ──
    s_cfg = dict(subtitle_config) if isinstance(subtitle_config, dict) else {}
@@ -595,8 +608,21 @@ def build_direct_render(
        s_pos_x, s_pos_y = None, None
    s_margin_top = _scale_title_len(60, output_width)  # subtitle top (not commonly used)
    s_margin_bottom = _scale_title_len(SUBTITLE_DEFAULT_MARGIN_BOTTOM, output_width)
-    s_borderw = max(1, _scale_title_len(2, output_width))  # 字幕默认描边保证可读性
+    # subtitle stroke/bold：先解析用户 stroke，再按 bold 默认补描边
+    s_borderw = 0
    s_border_color = "#000000"
+    _s_stroke = s_cfg.get("stroke")
+    if isinstance(_s_stroke, dict) and _s_stroke.get("enabled", False):
+        try:
+            s_borderw = _scale_title_len(int(float(_s_stroke.get("width", 2))), output_width)
+        except (TypeError, ValueError):
+            s_borderw = 0
+        s_border_color = str(_s_stroke.get("color", "#000000"))
+    s_bold = bool(s_cfg.get("bold", False))
+    if s_bold and s_borderw < 1:
+        # 粗体默认黑色细描边 2@720p（与 title/CPU vfb 一致）
+        s_borderw = _scale_title_len(TITLE_FAUX_BOLD_WIDTH, output_width)
+        s_border_color = "#000000"

    # 静态字幕：static_subtitle_text 非空时构造全片长 segment（0 → total_duration）
    static_text = (static_subtitle_text or "").strip()
@@ -132,6 +132,7 @@ class RenderAdapter:
        work_dir: Path | None = None,
        progress_cb: ProgressCallback | None = None,
        voiceover_audio_path: str | None = None,
+        task_config_override: dict | None = None,  # Bug A: task 级 config 覆盖，防并发竞态
    ) -> RenderAdapterResult:
        """渲染一个 EditPlan。

@@ -217,6 +218,7 @@ class RenderAdapter:
                rendered_clip_ids=rendered_clip_ids,
                failed_clip_ids=failed_clip_ids,
                voiceover_audio_path=voiceover_audio_path,
+                task_config_override=task_config_override,
            )
            # 成功时将临时目录所有权转移给调用方，阻止 finally 清理
            if result.success and temp_dir:
@@ -392,7 +394,7 @@ class RenderAdapter:

        return asset_path_map, rendered_clip_ids, failed_clip_ids, asset_storage_map

-    def _prepare_bgm(self, plan, work_dir: Path, plan_id: str) -> str | None:
+    def _prepare_bgm(self, plan, work_dir: Path, plan_id: str, *, bgm_override: dict | None = None) -> str | None:
        """准备 BGM 音频文件（从 plan.config.bgm 读取配置）。

        支持 3 种来源（按优先级）：
@@ -405,7 +407,9 @@ class RenderAdapter:
        from urllib.parse import urlparse

        plan_config = plan.config or {}
-        bgm_config = plan_config.get("bgm", {}) or {}
+        bgm_config = dict(plan_config.get("bgm", {}) or {})
+        if isinstance(bgm_override, dict) and bgm_override:
+            bgm_config.update(bgm_override)  # Bug A: 任务级 BGM 覆盖，防并发竞态

        if not bgm_config.get("enabled", False):
            return None
@@ -567,6 +571,7 @@ class RenderAdapter:
        failed_clip_ids: list[str] | None = None,
        voiceover_audio_path: str | None = None,
        asset_storage_map: dict[str, str] | None = None,
+        task_config_override: dict | None = None,  # Bug A: task 级 config 覆盖，防并发竞态
    ) -> RenderAdapterResult:
        """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。

@@ -580,8 +585,9 @@ class RenderAdapter:
        Returns:
            RenderAdapterResult
        """
-        # 1. 准备 BGM
-        bgm_path = self._prepare_bgm(plan, work_dir, plan_id)
+        # 1. 准备 BGM（Bug A: 传 task 级 bgm override）
+        _bgm_override = (task_config_override or {}).get("bgm") if isinstance(task_config_override, dict) else None
+        bgm_path = self._prepare_bgm(plan, work_dir, plan_id, bgm_override=_bgm_override)

        self._report_progress(progress_cb, 40.0, "执行视频渲染")

@@ -589,8 +595,10 @@ class RenderAdapter:
        plan_config = plan.config or {}
        asr_service = self._get_asr_service()

-        # 3. 读取输出分辨率
-        export_config = plan_config.get("export", {}) or {}
+        # 3. 读取输出分辨率（Bug A: task override 优先）
+        export_config = dict(plan_config.get("export", {}) or {})
+        if isinstance(task_config_override, dict) and isinstance(task_config_override.get("export"), dict):
+            export_config.update(task_config_override["export"])
        if not isinstance(export_config, dict):
            export_config = {}
        output_width, output_height = _parse_resolution(export_config.get("resolution"))
@@ -615,6 +623,7 @@ class RenderAdapter:
            asr_service=asr_service,
            voiceover_audio_path=voiceover_audio_path,
            clip_has_text=clip_has_text,
+            override_config=task_config_override,
        )
        # 注入每个视频段对应素材的 storage_key，供全 GPU 直连管线直接签名下载
        _storage_map = asset_storage_map or {}
@@ -158,6 +158,7 @@ class UnifiedRenderService:
        bgm_path: str | None = None,  # BGM 本地文件路径
        voiceover_audio_path: str | None = None,  # 配音素材库音频本地路径
        clip_has_text: list[bool] | None = None,  # 源视频片段是否有文字（来自 atom_clip.ai_tags.has_text）
+        override_config: dict | None = None,  # Bug A: task 级 config 覆盖（title/bgm/export/subtitle），防并发竞态
    ):
        self.plan = plan
        self.clips = clips
@@ -170,6 +171,9 @@ class UnifiedRenderService:
        self.asr_service = asr_service
        self.bgm_path = bgm_path
        self.voiceover_audio_path = voiceover_audio_path
+        # Bug A: task 级 config override（深拷贝），优先级高于 plan.config；
+        # 避免同 plan 多任务并发渲染时 _sync_task_config_to_plan 写 plan.config["title"] 互相覆盖。
+        self._override_config = dict(override_config) if isinstance(override_config, dict) else {}
        # #1970：片段级文字检测（顺序与非 audio 的源视频片段一致）；None 表示无可靠检测，保守不翻转
        self._clip_has_text = clip_has_text
        self._transition_engine = TransitionEngine(default_duration=transition_duration)
@@ -180,6 +184,28 @@ class UnifiedRenderService:
        self._micro_plan_cache: Any = None
        self._micro_plan_loaded = False

+    def _cfg_section(self, section: str) -> dict:
+        """读取单个配置段：override_config 优先于 plan.config（Bug A 防并发竞态）。"""
+        base = dict((self.plan.config or {}).get(section, {}) or {})
+        override = self._override_config.get(section)
+        if isinstance(override, dict) and override:
+            base.update(override)  # 浅合并，保留 base 中未被覆盖字段
+        return base
+
+    def _effective_config(self) -> dict:
+        """读取完整 config：override_config 顶层段覆盖 plan.config（Bug A 防并发竞态）。"""
+        import copy
+
+        full = copy.deepcopy(self.plan.config or {})
+        for k, v in self._override_config.items():
+            if isinstance(v, dict):
+                sec = dict(full.get(k, {}) or {})
+                sec.update(v)
+                full[k] = sec
+            else:
+                full[k] = v
+        return full
+
    # ── #1970 PR2 智能降重：片段级微变换 ───────────────────────────────────
    def _dedup_enabled(self) -> bool:
        """读取 plan.config.dedup_enabled，缺省视为 True（向后兼容）。"""
@@ -431,7 +457,7 @@ class UnifiedRenderService:
            has_audio = pass_through_has_audio
            # 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频
            if self.bgm_path and pass_through_has_audio:
-                config = self.plan.config or {}
+                config = self._effective_config()
                bgm_config = config.get("bgm", {}) or {}
                if bgm_config.get("enabled", False):
                    ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
@@ -471,7 +497,7 @@ class UnifiedRenderService:
                            "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
                        )
        else:
-            config = self.plan.config or {}
+            config = self._effective_config()
            bgm_config = config.get("bgm", {}) or {}
            if not isinstance(bgm_config, dict):
                bgm_config = {}
@@ -696,7 +722,7 @@ class UnifiedRenderService:
        Returns:
            ASS 文件路径，没有字幕时返回 None
        """
-        config = self.plan.config or {}
+        config = self._effective_config()
        # #1901 统一读 "title"，兼容老数据 "title_config"
        title_cfg = config.get("title", {}) or {}
        if not isinstance(title_cfg, dict) or not (title_cfg.get("text") or "").strip():
@@ -881,7 +907,7 @@ class UnifiedRenderService:
        Returns:
            是否成功添加了配音音轨
        """
-        config = self.plan.config or {}
+        config = self._effective_config()
        tts_cfg = config.get("tts", {}) or {}
        if not isinstance(tts_cfg, dict):
            tts_cfg = {}
@@ -2275,7 +2301,7 @@ class UnifiedRenderService:
        try:
            from video_processing import gpu_direct_pipeline as gdp

-            cfg = self.plan.config or {}
+            cfg = self._effective_config()
            video_layer = next(_lyr for _lyr in layers if _lyr.role not in ("audio",))
            video_clips = [c for c in video_layer.clips if c.clip_type != "audio"]

@@ -616,69 +616,49 @@ def _precompute_render_metadata(
 # ── Celery Task ──────────────────────────────────────────────────────────────


-def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None:
-    """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。
+def _build_task_config_override(task_info: dict) -> dict:
+    """Bug A: 从 task_info 构建任务级 config override 深拷贝，供渲染时覆盖 plan.config。

-    包括：title_config、BGM、输出分辨率。配音单独处理（需下载到本地）。
+    所有渲染相关配置（title/bgm/export）从任务自身读取，不再依赖共享 plan.config，
+    彻底消除同 plan 多任务并发渲染时的竞态覆盖问题。
    """
-    from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
-        SQLAlchemyEditPlanRepository,
-    )
+    import copy

-    plan_repo = SQLAlchemyEditPlanRepository(db)
-    plan = plan_repo.get(source_edit_plan_id)
-    if plan is None:
-        logger.error("[task] EditPlan not found: %s", source_edit_plan_id)
-        return None
+    override: dict = {}

-    plan_config = dict(plan.config or {})
-    changed = False
-
-    # 标题配置
+    # 标题配置（字段名归一化）
    title_config = task_info.get("title_config") or {}
-    if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip():
-        cfg = dict(title_config)
-        # 字段名归一化
+    if isinstance(title_config, dict) and title_config:
+        cfg = copy.deepcopy(title_config)
        if "font_size" in cfg and "size" not in cfg:
            cfg["size"] = cfg["font_size"]
        if "font_color" in cfg and "color" not in cfg:
            cfg["color"] = cfg["font_color"]
-        plan_config["title"] = cfg
-        changed = True
-        logger.info("[task] title_config synced to plan: %s", cfg.get("text", "")[:30])
+        override["title"] = cfg

    # BGM 配置
    bgm_config = task_info.get("bgm_config") or {}
-    if bgm_config:
-        from packages.domain.bgm_utils import merge_bgm_config
-
-        existing_bgm = plan_config.get("bgm", {}) or {}
-        plan_config["bgm"] = merge_bgm_config(existing_bgm, bgm_config)
-        changed = True
+    if isinstance(bgm_config, dict) and bgm_config:
+        override["bgm"] = copy.deepcopy(bgm_config)

    # 输出分辨率
    ow = task_info.get("output_width") or OUTPUT_WIDTH
    oh = task_info.get("output_height") or OUTPUT_HEIGHT
    if ow >= 100 and oh >= 100:
-        export_cfg = dict(plan_config.get("export", {}) or {})
-        export_cfg["resolution"] = f"{ow}x{oh}"
-        plan_config["export"] = export_cfg
-        changed = True
+        override["export"] = {"resolution": f"{ow}x{oh}"}

-    if changed:
-        plan.config = plan_config
-        plan_repo.update(plan)
-        logger.info("[task] plan.config synced: plan_id=%s", source_edit_plan_id)
+    return override
+
+
+def _download_voice_for_task(task_info: dict, source_edit_plan_id: str) -> str | None:
+    """下载任务配音到本地临时文件，返回路径（不读写 plan.config）。"""
+    import tempfile

-    # 配音下载
    voiceover_path: str | None = None
    voice_library_id = task_info.get("voice_library_id", "")
-    # #1749：voice_ids 冗余字段已移除；配音一律以 voice_library_id 为准（独立配音每变体各自绑定）
    effective_voice_id = voice_library_id or ""

    if effective_voice_id:
-        import tempfile
-
        voice_tmp = Path(tempfile.gettempdir()) / f"voice_{source_edit_plan_id}_{id(task_info)}.mp3"
        try:
            if _download_voice_asset(effective_voice_id, voice_tmp):
@@ -697,6 +677,9 @@ def _render_from_edit_plan(
 ) -> tuple[Path, float, list[dict] | None, str | None, str | None, str, bool]:
    """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。

+    Bug A 修复：不再通过 _sync_task_config_to_plan 写共享 plan.config；
+    渲染配置通过 task_config_override 参数直接传入渲染层，彻底消除并发竞态。
+
    Returns:
        (output_path, render_duration, cover_candidates, voiceover_path, temp_dir, thumbnail_url, edge_crop_applied)
    """
@@ -705,8 +688,10 @@ def _render_from_edit_plan(

    db = SessionLocal()
    try:
-        # 同步配置到 plan.config + 下载配音
-        voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db)
+        # Bug A: 构建任务级 config override（深拷贝自 task_info），不写 plan.config，避免并发竞态
+        task_override = _build_task_config_override(task_info)
+        # 下载配音到本地临时文件（不依赖 plan.config）
+        voiceover_path = _download_voice_for_task(task_info, source_edit_plan_id)

        # 进度回调
        def _progress_cb(progress: float, stage: str):
@@ -722,6 +707,7 @@ def _render_from_edit_plan(
            job_id=task_id,
            progress_cb=_progress_cb,
            voiceover_audio_path=voiceover_path,
+            task_config_override=task_override,
        )

        if not result.success:
@@ -918,3 +918,71 @@ class GpuWorkerModel(Base):
    capabilities = Column(String(500), nullable=False, default="")  # 逗号分隔，如 "musetalk"
    last_heartbeat_at = Column(DateTime, nullable=True, index=True)
    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+
+
+class ViralVideoJobModel(Base):
+    """爆款视频任务"""
+
+    __tablename__ = "viral_video_jobs"
+
+    id = Column(String(36), primary_key=True)
+    user_id = Column(String(36), nullable=False, index=True)
+    images = Column(JSON, nullable=False, default=list)  # 产品图片 URL 列表
+    industry = Column(String(100), nullable=False, default="")
+    target_customer = Column(String(500), nullable=False, default="")
+    persona_id = Column(String(36), nullable=False, default="")
+    viral_structure = Column(String(50), nullable=False, default="")
+    marketing_purpose = Column(String(100), nullable=False, default="")
+    bgm_preference = Column(String(50), nullable=False, default="")
+    duration = Column(Integer, nullable=False, default=30)
+    user_copy_text = Column(Text, nullable=False, default="")
+    fusion_level = Column(String(20), nullable=False, default="ai_polish")
+    reference_audio_path = Column(String(1000), nullable=False, default="")
+    # v1.3 新增字段
+    reference_video_url = Column(String(1000), nullable=False, default="")
+    style_strength = Column(String(20), nullable=False, default="medium")
+    style_guide = Column(JSON, nullable=True)
+    style_template_id = Column(String(36), nullable=False, default="", index=True)
+    # 结果与状态
+    status = Column(String(30), nullable=False, default="pending", index=True)
+    intent_result = Column(JSON, nullable=True)
+    result_video_url = Column(String(1000), nullable=False, default="")
+    credits_cost = Column(Integer, nullable=False, default=0)
+    error_msg = Column(Text, nullable=False, default="")
+    retry_count = Column(Integer, nullable=False, default=0)
+    started_at = Column(DateTime(timezone=True), nullable=True)
+    completed_at = Column(DateTime(timezone=True), nullable=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
+
+
+class ViralVideoStyleTemplateModel(Base):
+    """爆款视频风格模板配置表"""
+
+    __tablename__ = "viral_video_style_templates"
+
+    id = Column(String(36), primary_key=True)
+    name = Column(String(200), nullable=False)
+    description = Column(Text, nullable=False, default="")
+    thumbnail_url = Column(String(1000), nullable=False, default="")
+    style_config = Column(JSON, nullable=False, default=dict)
+    is_system = Column(Boolean, nullable=False, default=True, index=True)
+    sort_order = Column(Integer, nullable=False, default=0)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
+
+
+class ViralVideoPromptTemplateModel(Base):
+    """爆款视频 Prompt 模板表（由 #2040 seed）"""
+
+    __tablename__ = "viral_video_prompt_templates"
+
+    id = Column(String(36), primary_key=True)
+    prompt_type = Column(String(50), nullable=False, index=True)
+    name = Column(String(200), nullable=False)
+    content = Column(Text, nullable=False, default="")
+    variables = Column(JSON, nullable=False, default=list)
+    version = Column(Integer, nullable=False, default=1)
+    is_active = Column(Boolean, nullable=False, default=True, index=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
@@ -322,22 +322,41 @@ class TestBatchPreviewRoute:
                )
        assert captured[0].voice_library_id == "legacy_voice"

-    def test_preview_queue_limit_checks_total_count(self):
-        """限流预检查按变体总数计：用户 pending + N 超限 → 429"""
+    def test_preview_queue_limit_global_returns_503(self):
+        """#2098：仅全局硬上限仍 503 拒绝；用户超限自动排队不返回 429。"""
        from app.api.routes.generation_preview import create_preview_generation_task
        from fastapi import HTTPException

-        repo = MagicMock()
-        repo.count_pending_by_user.return_value = 3
-        repo.count_pending_total.return_value = 0
+        # 全局硬上限：预检查即 503，不会进入后续流程
+        repo_global = MagicMock()
+        repo_global.count_pending_total.return_value = 100
        with pytest.raises(HTTPException) as exc:
            create_preview_generation_task(
                _make_preview_request(preview_count=5),
                authenticated_user=_make_user(),
-                generation_task_repository=repo,
+                generation_task_repository=repo_global,
                db=MagicMock(),
            )
-        assert exc.value.status_code == 429
+        assert exc.value.status_code == 503
+
+    def test_preview_user_over_limit_does_not_429(self):
+        """#2098：用户 pending 远超软上限也不返回 429/4xx，请求进入业务流程后因 mock 不足走 500。"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+        from fastapi import HTTPException
+
+        repo_user = MagicMock()
+        repo_user.count_pending_by_user.return_value = 100  # 远超用户软上限
+        repo_user.count_pending_total.return_value = 0
+        with pytest.raises(HTTPException) as exc:
+            create_preview_generation_task(
+                _make_preview_request(preview_count=1),
+                authenticated_user=_make_user(),
+                generation_task_repository=repo_user,
+                db=MagicMock(),
+            )
+        # 关键断言：不是 429（也不是 503，因为全局未超限），说明预检查放过了请求
+        assert exc.value.status_code != 429
+        assert exc.value.status_code != 503

    def test_preview_reselect_failure_marks_all_failed(self):
        """#1743：变体独立选片（reselect）重试仍失败 → 已创建任务全部标记 failed 并 500"""
@@ -741,22 +741,23 @@ class TestCreatePreviewRoute:
        assert resp.items[0].status == "pending"
        assert resp.items[0].variant_index == 0

-    def test_user_pending_limit_exceeded(self):
-        """用户待处理任务超限 → 429"""
-        repo = MagicMock()
-        repo.count_pending_by_user.return_value = 3
-        repo.count_pending_total.return_value = 5
-
+    def test_user_pending_limit_no_longer_rejects(self):
+        """#2098: 用户待处理任务超限不再 429 拒绝（预检查仅全局 503）。"""
        from fastapi import HTTPException

-        with pytest.raises(HTTPException) as exc_info:
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 100
+        repo.count_pending_total.return_value = 5
+        with pytest.raises(HTTPException) as exc:
            create_preview_generation_task(
                self._make_request(),
                authenticated_user=_make_user(),
                generation_task_repository=repo,
                db=MagicMock(),
            )
-        assert exc_info.value.status_code == 429
+        # 不是 429（用户级硬拒已移除）也不是 503（全局未超限）
+        assert exc.value.status_code != 429
+        assert exc.value.status_code != 503

    def test_global_queue_full(self):
        """全局队列满 → 503"""
@@ -844,36 +845,32 @@ class TestCreatePreviewRoute:
            assert resp.total == 1
            assert resp.items[0].status == "failed"

-    def test_enqueue_raises_user_limit(self):
-        """safe_enqueue 抛出 UserPendingLimitExceeded → 429"""
+    def test_enqueue_user_limit_exception_no_longer_returns_429(self):
+        """#2098: 即使 safe_enqueue 模拟抛 UserPendingLimitExceeded，也不再触发 429 整体拒绝，
+        变体被标记 failed 后正常返回响应。"""
        repo = MagicMock()
        repo.count_pending_by_user.return_value = 0
        repo.count_pending_total.return_value = 0
-
        task = _make_task()
-        from fastapi import HTTPException

-        def _set_failed_limit(error_message="", **_kwargs):
+        def _set_failed(reason="", **_kw):
            task.status = GenerationTaskStatus.FAILED
-            task.error_message = error_message or "待处理任务超限"
+            task.error_message = reason

-        task.mark_failed.side_effect = _set_failed_limit
-
-        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
-            MockUC.return_value.execute.return_value = task
-            with patch(
-                "app.api.routes.generation_preview.safe_enqueue_generation_task",
-                side_effect=UserPendingLimitExceeded(user_id="u1", pending_count=4, limit=3),
-            ):
-                with pytest.raises(HTTPException) as exc_info:
-                    create_preview_generation_task(
-                        self._make_request(),
-                        authenticated_user=_make_user(),
-                        generation_task_repository=repo,
-                        db=MagicMock(),
-                    )
-                # 全部变体入队失败且错误消息含"待处理任务" → 429
-                assert exc_info.value.status_code == 429
+        task.mark_failed.side_effect = _set_failed
+        repo.create.return_value = task
+        with patch(
+            "app.api.routes.generation_preview.safe_enqueue_generation_task",
+            side_effect=UserPendingLimitExceeded(user_id="u1", pending_count=4, limit=20),
+        ):
+            resp = create_preview_generation_task(
+                self._make_request(),
+                authenticated_user=_make_user(),
+                generation_task_repository=repo,
+                db=MagicMock(),
+            )
+            assert resp.total == 1
+            assert resp.items[0].status == "failed"

    def test_enqueue_raises_global_queue_full(self):
        """safe_enqueue 抛出 GlobalQueueFull → 503"""
@@ -85,7 +85,7 @@ def _patch_pipeline_helpers(monkeypatch):
 # ---------------------------------------------------------------------------
 class TestNoConfigBackwardCompat:
    def test_default_title_drawtext_white_top(self, monkeypatch):
-        """不传 title_config 时：白字、top、36@720 按 width 缩放到 64、y=71（PAD16+margin24）。"""
+        """不传 title_config 时：白字、top、48@720 按 width 缩放到 85、y=89（50@720p baseline）。"""
        import video_processing.gpu_direct_pipeline as gdp

        _patch_pipeline_helpers(monkeypatch)
@@ -101,15 +101,15 @@ class TestNoConfigBackwardCompat:
        )
        fc = " ".join(plan.filter_complex)
        assert "fontcolor=0xffffff" in fc
-        # 默认 position=top → y=71（scale(16+24)=71），不含 h-th
-        assert "y=71" in fc
+        # 默认 position=top → y=89（scale(50)=89，与 CPU/vfb 一致），不含 h-th
+        assert "y=89" in fc
        assert "h-th" not in fc
-        # 默认字号 36@720 经 1280/720 缩放 = 64
-        assert "fontsize=64" in fc
+        # 默认字号 48@720 经 1280/720 缩放 = 85
+        assert "fontsize=85" in fc
        assert "text='默认标题'" in fc
-        # 默认粗体：drawtext 无原生粗体，用同色描边 borderw=1@720（1280/720→scale=2）模拟
-        assert "borderw=2" in fc
-        assert "bordercolor=0xffffff" in fc
+        # 默认粗体：黑色细描边 borderw=2@720（scale=4），与 CPU vfb 一致避免重影
+        assert "borderw=4" in fc
+        assert "bordercolor=0x000000" in fc

    def test_no_subtitle_when_none(self, monkeypatch):
        import video_processing.gpu_direct_pipeline as gdp
@@ -153,8 +153,8 @@ class TestTitleStylePassthrough:
        assert "fontcolor=0xff0000" in fc
        # 60@720 按 1280/720 缩放 = 107
        assert "fontsize=107" in fc
-        # top 位置 y=71（scale(PAD16+margin_top24)=71）
-        assert "y=71" in fc
+        # top 位置默认 margin 50@720p → scale=89（未传 margin_top 用默认）
+        assert "y=89" in fc
        assert "h-th" not in fc

    def test_title_position_center(self, monkeypatch):
@@ -218,9 +218,9 @@ class TestTitleStylePassthrough:
        # 阴影层 + 主字层 = 2 条 drawtext
        assert drawtext_count == 2
        joined = " ".join(plan.filter_complex)
-        # shadow offset 3@720 经 1280/720 缩放 = 5；默认 position=top → y=71
+        # shadow offset 3@720 经 1280/720 缩放 = 5；默认 position=top → y=89
        assert "x=(w-text_w)/2+5" in joined
-        assert "y=71+5" in joined
+        assert "y=89+5" in joined
        assert "h-th" not in joined

    def test_title_font_override(self, monkeypatch):
@@ -282,10 +282,10 @@ class TestTitleStylePassthrough:
        assert "(w-text_w)*0.3000" in fc
        assert "(h-text_h)*0.6000" in fc
        assert "h-th-" not in fc
-        assert "y=71" not in fc
+        assert "y=89" not in fc

    def test_title_margin_top_respected(self, monkeypatch):
-        """margin_top 透传：100@720 → scale(16+100)=206@1280。"""
+        """margin_top 透传：100@720，叠加默认 50 → 150@720 → scale=267@1280。"""
        import video_processing.gpu_direct_pipeline as gdp

        _patch_pipeline_helpers(monkeypatch)
@@ -300,8 +300,8 @@ class TestTitleStylePassthrough:
            title_config={"text": "远离顶部", "position": "top", "margin_top": 100},
        )
        fc = " ".join(plan.filter_complex)
-        # scale(PAD16 + margin_top100) = 116*1280/720 = 206
-        assert "y=206" in fc
+        # 默认 50 + margin_top100 = 150@720 → scale=267
+        assert "y=267" in fc

    def test_title_default_bold_true(self, monkeypatch):
        """不传 bold 时默认粗体：drawtext 用同色描边 borderw 模拟。"""
@@ -319,9 +319,9 @@ class TestTitleStylePassthrough:
            title_config={"text": "粗体"},
        )
        fc = " ".join(plan.filter_complex)
-        # bold=True 默认：同色描边 width=1@720 → scale=2，颜色与文字相同（白色）
-        assert "borderw=2" in fc
-        assert "bordercolor=0xffffff" in fc
+        # bold=True 默认：黑色细描边 width=2@720 → scale=4，与 CPU vfb 一致
+        assert "borderw=4" in fc
+        assert "bordercolor=0x000000" in fc
        assert "text='粗体'" in fc

    def test_title_bold_false_disables_faux_bold(self, monkeypatch):
@@ -359,8 +359,8 @@ class TestTitleStylePassthrough:
            title_config={"text": "底部", "position": "bottom"},
        )
        fc = " ".join(plan.filter_complex)
-        # bottom margin PAD16+24=40 → scale=71
-        assert "y=h-th-71" in fc
+        # bottom margin 50@720 → scale=89
+        assert "y=h-th-89" in fc

    def test_title_size_scales_by_width_not_height(self, monkeypatch):
        """不同分辨率下同 @720 基准的 size 等比缩放：1080x1920 下 36→54。"""
@@ -378,10 +378,10 @@ class TestTitleStylePassthrough:
            clip_volumes=[1.0],
        )
        fc = " ".join(plan.filter_complex)
-        # default size 36@720 → 1080w = 54
-        assert "fontsize=54" in fc
-        # top margin 40@720 → 60
-        assert "y=60" in fc
+        # default size 48@720 → 1080w = 72
+        assert "fontsize=72" in fc
+        # top margin 50@720 → 75
+        assert "y=75" in fc


 # ---------------------------------------------------------------------------
@@ -434,8 +434,8 @@ class TestSubtitlePassthrough:
        assert "fontcolor=0x0000ff" in fc
        # sub size 28@720 经 1280/720 缩放 = 50
        assert "fontsize=50" in fc
-        # bottom margin 60@720 → 107
-        assert "y=h-th-107" in fc
+        # bottom margin 50@720 → 89
+        assert "y=h-th-89" in fc

    def test_subtitle_disabled_hides_subs(self, monkeypatch):
        import video_processing.gpu_direct_pipeline as gdp
@@ -644,18 +644,18 @@ class TestCombinedConfig:
            bgm_config={"volume": 0.2, "fade_in": 0.5, "fade_out": 1.0},
        )
        fc = " ".join(plan.filter_complex)
-        # 标题：size 50@720 → 89，top margin 40@720 → 71，stroke 2@720 → 4
+        # 标题：size 50@720 → 89，top margin 50@720 → 89，stroke 2@720 → 4
        assert "text='主标题'" in fc
        assert "fontcolor=0xffff00" in fc
        assert "fontsize=89" in fc
-        assert "y=71" in fc
+        assert "y=89" in fc
        assert "borderw=4" in fc
-        # 字幕：size 24@720 → 43，bottom margin 60@720 → 107
+        # 字幕：size 24@720 → 43，bottom margin 50@720 → 89（默认不加粗）
        assert "text='成片全字幕'" in fc
        assert "fontsize=43" in fc
-        assert "y=h-th-107" in fc
+        assert "y=h-th-89" in fc
        assert "fontcolor=0xffffff" in fc
-        # subtitle 默认 bold=True 也会加同色描边（width=1→2）；但若主标题已有 stroke 不影响字幕独立 borderw=4（s 默认 2 像素@720→scale=4）
+        # subtitle 默认 bold=False，无额外描边（用户未开 stroke）
        # BGM
        assert "volume=0.200" in fc
        assert "afade=t=in:st=0:d=0.50" in fc
@@ -688,7 +688,7 @@ class TestEdgeCases:
        assert "drawtext=" in fc

    def test_bold_title_increases_borderw(self, monkeypatch):
-        """bold=True 时若原无描边，自动加 borderw=1 用同色描边模拟加粗。"""
+        """bold=True 时若原无描边，自动加 borderw 黑色细描边模拟加粗（与 CPU vfb 一致，避免重影）。"""
        import video_processing.gpu_direct_pipeline as gdp

        _patch_pipeline_helpers(monkeypatch)
@@ -703,9 +703,9 @@ class TestEdgeCases:
            title_config={"text": "粗体", "bold": True, "color": "#ff0000"},
        )
        fc = " ".join(plan.filter_complex)
-        # 加粗模拟 borderw>=1，且 bordercolor 跟字体色一致（0xff0000）
-        assert "borderw=" in fc
-        assert "bordercolor=0xff0000" in fc
+        # 仿粗用黑色细描边 2@720→scale=4，不跟文字色（避免同色描边重影）
+        assert "borderw=4" in fc
+        assert "bordercolor=0x000000" in fc

    def test_asr_and_static_subtitle_asr_wins(self, monkeypatch):
        """同时传 static_subtitle_text 和 ASR segments 时，ASR 优先（不插入静态全文）。"""
@@ -1,4 +1,4 @@
-"""task_enqueue 单测 — 队列限流 + 安全入队逻辑."""
+"""task_enqueue 单测 — 队列限流 + 安全入队逻辑 (#2098)."""

 from __future__ import annotations

@@ -14,12 +14,8 @@ from app.core.task_enqueue import (
    safe_enqueue_generation_task,
 )

-# ── Fixtures / Helpers ─────────────────────────────────────────────────────
-

 class MockRepository:
-    """Mock 任务仓储，用计数器模拟 pending 数量."""
-
    def __init__(self, global_count: int = 0, user_count: int = 0):
        self._global = global_count
        self._user = user_count
@@ -43,19 +39,12 @@ def make_mock_task(task_id: str = "task-1"):
    return task


-# ── check_queue_limits ────────────────────────────────────────────────────
-
-
 class TestCheckQueueLimits:
-    """check_queue_limits 预检查限流."""
-
    def test_below_limits_passes(self):
        repo = MockRepository(global_count=5, user_count=1)
-        # 不抛异常就是通过
        check_queue_limits("user-1", repo)

    def test_global_at_limit_raises(self):
-        """达到全局上限即拒绝."""
        repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=1)
        with pytest.raises(GlobalQueueFull) as exc_info:
            check_queue_limits("user-1", repo)
@@ -67,210 +56,145 @@ class TestCheckQueueLimits:
        with pytest.raises(GlobalQueueFull):
            check_queue_limits("user-1", repo)

-    def test_user_at_limit_raises(self):
-        """达到用户上限即拒绝."""
+    def test_user_at_limit_no_longer_raises(self):
        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT)
-        with pytest.raises(UserPendingLimitExceeded) as exc_info:
-            check_queue_limits("user-1", repo)
-        assert exc_info.value.user_id == "user-1"
-        assert exc_info.value.pending_count == USER_PENDING_LIMIT
-        assert exc_info.value.limit == USER_PENDING_LIMIT
+        check_queue_limits("user-1", repo)

-    def test_user_over_limit_raises(self):
-        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 1)
-        with pytest.raises(UserPendingLimitExceeded):
-            check_queue_limits("user-1", repo)
-
-    def test_global_priority_over_user(self):
-        """全局和用户都超限时，优先抛全局异常."""
-        repo = MockRepository(
-            global_count=GLOBAL_PENDING_LIMIT + 1,
-            user_count=USER_PENDING_LIMIT + 1,
-        )
-        with pytest.raises(GlobalQueueFull):
-            check_queue_limits("user-1", repo)
+    def test_user_over_limit_no_longer_raises(self):
+        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100)
+        check_queue_limits("user-1", repo)

    def test_empty_user_id_skips_user_check(self):
-        """user_id 为空时跳过用户级检查."""
        repo = MockRepository(global_count=5, user_count=999)
-        # 不抛异常 = 通过（只检查全局）
        check_queue_limits("", repo)

-    def test_custom_limits(self):
-        """支持自定义限流阈值."""
-        repo = MockRepository(global_count=5, user_count=5)
-        # 默认阈值下 user 5 > 3 会被拒
-        with pytest.raises(UserPendingLimitExceeded):
-            check_queue_limits("u1", repo)
-
-        # 自定义更高阈值就能通过
-        check_queue_limits("u1", repo, user_pending_limit=10, global_pending_limit=10)
-
-
-# ── safe_enqueue_generation_task ──────────────────────────────────────────
+    def test_custom_global_limit_still_honored(self):
+        repo = MockRepository(global_count=15, user_count=999)
+        with pytest.raises(GlobalQueueFull):
+            check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=10)
+        check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=20)


 class TestSafeEnqueueGenerationTask:
-    """safe_enqueue_generation_task 安全入队."""
-
    @patch("app.core.task_enqueue.celery_app")
    def test_success_path(self, mock_celery):
-        """正常路径：入队前检查通过 → 发送Celery → 入队后检查通过."""
        repo = MockRepository(global_count=1, user_count=1)
        task = make_mock_task()
-
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
-
        assert result is True
        mock_celery.send_task.assert_called_once_with("worker.generate_video", args=[task.id])
        task.mark_failed.assert_not_called()

    @patch("app.core.task_enqueue.celery_app")
    def test_no_user_id_skips_user_check(self, mock_celery):
-        """不传 user_id 跳过用户级限流."""
        repo = MockRepository(global_count=1, user_count=999)
        task = make_mock_task()
-
        result = safe_enqueue_generation_task(task, repo, user_id="")
        assert result is True

    @patch("app.core.task_enqueue.celery_app")
    def test_precheck_global_over_marks_failed(self, mock_celery):
-        """入队前全局超限：标记 failed，抛异常."""
        repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT + 1, user_count=0)
        task = make_mock_task()
-
        with pytest.raises(GlobalQueueFull):
            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
        task.mark_failed.assert_called_once()
        mock_celery.send_task.assert_not_called()
        assert repo.update_called == 1

    @patch("app.core.task_enqueue.celery_app")
-    def test_precheck_user_over_marks_failed(self, mock_celery):
-        """入队前用户超限：标记 failed，抛异常."""
-        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 1)
+    def test_precheck_user_over_still_enqueues(self, mock_celery, caplog):
+        import logging
+
+        caplog.set_level(logging.WARNING)
+        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100)
        task = make_mock_task()
-
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        task.mark_failed.assert_called_once()
-        mock_celery.send_task.assert_not_called()
+        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
+        assert result is True
+        mock_celery.send_task.assert_called_once()
+        task.mark_failed.assert_not_called()
+        assert any("超过软上限" in r.message for r in caplog.records)

    @patch("app.core.task_enqueue.celery_app")
    def test_celery_send_false_returns_false(self, mock_celery):
-        """Celery 发送失败：返回 False，任务标记 failed."""
        repo = MockRepository(global_count=1, user_count=1)
        task = make_mock_task()
        mock_celery.send_task.side_effect = Exception("celery down")
-
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
-
        assert result is False
        task.mark_failed.assert_called_once()
        assert "入队失败" in task.mark_failed.call_args[0][0]

    @patch("app.core.task_enqueue.celery_app")
    def test_celery_send_failure_update_also_fails(self, mock_celery):
-        """Celery 发送失败 + mark_failed 更新也失败：不崩溃."""
        repo = MockRepository(global_count=1, user_count=1)
        repo.update = MagicMock(side_effect=Exception("db down"))
        task = make_mock_task()
        mock_celery.send_task.side_effect = Exception("celery down")
-
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
-
        assert result is False
-        # 不抛异常就是胜利

    @patch("app.core.task_enqueue.celery_app")
    def test_postcheck_global_over_rollback(self, mock_celery):
-        """入队后全局超限（并发竞态）：回滚标记 failed，抛异常."""
-        # 入队前刚好通过，但入队后再查发现超限
        call_count = [0]

        def count_pending_total_side_effect():
            call_count[0] += 1
-            if call_count[0] == 1:  # 入队前检查
-                return GLOBAL_PENDING_LIMIT  # 等于上限，用 > 判断所以通过
-            return GLOBAL_PENDING_LIMIT + 1  # 入队后再查，超限
+            if call_count[0] == 1:
+                return GLOBAL_PENDING_LIMIT
+            return GLOBAL_PENDING_LIMIT + 1

        repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=0)
        repo.count_pending_total = MagicMock(side_effect=count_pending_total_side_effect)
        task = make_mock_task()
-
        with pytest.raises(GlobalQueueFull):
            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        # 异常是 GlobalQueueFull 类型，且任务已被标记为 failed（含"入队后"原因）
        task.mark_failed.assert_called_once()
        assert "入队后" in task.mark_failed.call_args[0][0]
        mock_celery.send_task.assert_called_once()

    @patch("app.core.task_enqueue.celery_app")
-    def test_postcheck_user_over_rollback(self, mock_celery):
-        """入队后用户超限：回滚标记 failed，抛异常."""
-        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT)
-        # 入队前用 > 判断，等于上限通过；入队后模拟并发超限
-        original_user_count = repo.count_pending_by_user
+    def test_postcheck_user_over_does_not_rollback(self, mock_celery, caplog):
+        import logging

+        caplog.set_level(logging.WARNING)
+        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT)
        call_count = [0]

        def count_by_user_side_effect(user_id):
            call_count[0] += 1
-            if call_count[0] <= 1:  # 入队前
-                return USER_PENDING_LIMIT  # 用 > 判断，等于时通过
-            return USER_PENDING_LIMIT + 1  # 入队后，超限
+            if call_count[0] <= 1:
+                return USER_PENDING_LIMIT
+            return USER_PENDING_LIMIT + 1

        repo.count_pending_by_user = MagicMock(side_effect=count_by_user_side_effect)
        task = make_mock_task()
-
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        task.mark_failed.assert_called_once()
+        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
+        assert result is True
+        mock_celery.send_task.assert_called_once()
+        task.mark_failed.assert_not_called()
+        assert any("超软上限(入队后)" in r.message for r in caplog.records)

    @patch("app.core.task_enqueue.celery_app")
    def test_log_task_status_enabled(self, mock_celery):
-        """log_task_status=True 时日志中包含状态."""
        repo = MockRepository(global_count=1, user_count=1)
        task = make_mock_task()
-
        result = safe_enqueue_generation_task(task, repo, user_id="user-1", log_task_status=True)
        assert result is True

    @patch("app.core.task_enqueue.celery_app")
-    def test_custom_limits_in_enqueue(self, mock_celery):
-        """自定义限流阈值用于入队检查."""
-        repo = MockRepository(global_count=5, user_count=5)
+    def test_custom_global_limit_in_enqueue(self, mock_celery):
+        repo = MockRepository(global_count=15, user_count=999)
        task = make_mock_task()
-
-        # 默认阈值下用户 5 > 3 会被拒
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        # 重置 mock 计数
-        task.mark_failed.reset_mock()
-
-        # 调大阈值后通过
-        result = safe_enqueue_generation_task(
-            task,
-            repo,
-            user_id="user-1",
-            user_pending_limit=10,
-            global_pending_limit=10,
-        )
+        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
        assert result is True
-
-
-# ── 异常类 ────────────────────────────────────────────────────────────────
+        task2 = make_mock_task("task-2")
+        mock_celery.reset_mock()
+        with pytest.raises(GlobalQueueFull):
+            safe_enqueue_generation_task(task2, repo, user_id="user-1", user_pending_limit=999, global_pending_limit=10)


 class TestExceptionClasses:
-    """异常类消息格式."""
-
    def test_user_pending_limit_message(self):
        exc = UserPendingLimitExceeded("u1", 5, 3)
        assert "u1" in str(exc)
@@ -281,3 +205,6 @@ class TestExceptionClasses:
        exc = GlobalQueueFull(25, 20)
        assert "25" in str(exc)
        assert "20" in str(exc)
+
+    def test_user_pending_limit_constant(self):
+        assert USER_PENDING_LIMIT == 20
@@ -1,4 +1,4 @@
-"""任务队列限流防护单元测试。"""
+"""任务队列限流防护单元测试 (#2098: 用户级改为软上限，仅全局硬拒)."""

 from __future__ import annotations

@@ -19,17 +19,8 @@ from app.core.task_enqueue import (
    safe_enqueue_generation_task,
 )

-# ---------------------------------------------------------------------------
-# Mock helpers
-# ---------------------------------------------------------------------------
-

 class MockRepository:
-    """支持 pending 计数的 mock repository。
-
-    支持通过 set_pending 动态修改计数，用于模拟入队后计数变化的并发场景。
-    """
-
    def __init__(self, user_pending: int = 0, global_pending: int = 0):
        self._user_pending = user_pending
        self._global_pending = global_pending
@@ -47,7 +38,6 @@ class MockRepository:
        return task

    def set_pending(self, *, user_pending: int | None = None, global_pending: int | None = None):
-        """动态修改 pending 计数，模拟并发场景。"""
        if user_pending is not None:
            self._user_pending = user_pending
        if global_pending is not None:
@@ -67,58 +57,39 @@ class MockTask:

@pytest.fixture(autouse=True)
 def mock_celery(monkeypatch):
-    """mock 掉 celery_app.send_task，避免真实发送。"""
    mock_send = MagicMock()
    monkeypatch.setattr("app.core.celery_app.celery_app.send_task", mock_send)
    return mock_send


-# ---------------------------------------------------------------------------
-# 常量导出测试
-# ---------------------------------------------------------------------------
-
-
 def test_limit_constants_are_exported():
-    """限流阈值常量已导出，供业务代码引用。"""
-    assert USER_PENDING_LIMIT == 3
+    """#2098: USER_PENDING_LIMIT 从 3 提到 20 作为软上限；GLOBAL_PENDING_LIMIT 保持 20 为硬上限。"""
+    assert USER_PENDING_LIMIT == 20
    assert GLOBAL_PENDING_LIMIT == 20


-# ---------------------------------------------------------------------------
-# check_queue_limits 单元测试（预检查用，>= 边界）
-# ---------------------------------------------------------------------------
-
-
 class TestCheckQueueLimits:
-    """队列限流检查函数测试（预检查语义，>= 上限即拒绝）。"""
+    """check_queue_limits 预检查：仅全局硬上限拒绝，用户级改为软提示。"""

    def test_normal_passes_through(self):
-        """正常范围内的任务不受限制。"""
        repo = MockRepository(user_pending=1, global_pending=5)
        check_queue_limits("user-1", repo)

-    def test_user_limit_exceeded_raises(self):
-        """用户 pending 超过上限抛 UserPendingLimitExceeded。"""
-        repo = MockRepository(user_pending=4, global_pending=5)
-        with pytest.raises(UserPendingLimitExceeded) as exc_info:
-            check_queue_limits("user-1", repo)
-        assert exc_info.value.user_id == "user-1"
-        assert exc_info.value.pending_count == 4
-        assert exc_info.value.limit == 3
+    def test_user_limit_exceeded_no_longer_raises(self):
+        """#2098: 用户 pending 超过软上限不再抛异常。"""
+        repo = MockRepository(user_pending=100, global_pending=5)
+        check_queue_limits("user-1", repo)  # 不抛即通过

-    def test_user_at_limit_also_raises(self):
-        """用户 pending 刚好等于上限也拒绝（>= 边界）。"""
-        repo = MockRepository(user_pending=3, global_pending=5)
-        with pytest.raises(UserPendingLimitExceeded):
-            check_queue_limits("user-1", repo)
+    def test_user_at_limit_no_longer_raises(self):
+        """#2098: 用户 pending 等于软上限也不拒绝。"""
+        repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5)
+        check_queue_limits("user-1", repo)

    def test_user_below_limit_passes(self):
-        """用户 pending 比上限少 1，通过。"""
        repo = MockRepository(user_pending=2, global_pending=5)
        check_queue_limits("user-1", repo)

    def test_global_limit_exceeded_raises(self):
-        """全局 pending 超过上限抛 GlobalQueueFull。"""
        repo = MockRepository(user_pending=1, global_pending=21)
        with pytest.raises(GlobalQueueFull) as exc_info:
            check_queue_limits("user-1", repo)
@@ -126,82 +97,52 @@ class TestCheckQueueLimits:
        assert exc_info.value.limit == 20

    def test_global_at_limit_also_raises(self):
-        """全局 pending 刚好等于上限也拒绝（>= 边界）。"""
        repo = MockRepository(user_pending=1, global_pending=20)
        with pytest.raises(GlobalQueueFull):
            check_queue_limits("user-1", repo)

    def test_global_below_limit_passes(self):
-        """全局 pending 比上限少 1，通过。"""
        repo = MockRepository(user_pending=1, global_pending=19)
        check_queue_limits("user-1", repo)

-    def test_global_takes_priority_over_user(self):
-        """全局和用户都超限时，优先抛全局异常。"""
-        repo = MockRepository(user_pending=5, global_pending=25)
-        with pytest.raises(GlobalQueueFull):
-            check_queue_limits("user-1", repo)
-
    def test_empty_user_id_skips_user_check(self):
-        """不传 user_id 时跳过用户级检查，只做全局检查。"""
        repo = MockRepository(user_pending=10, global_pending=5)
-        # 用户超限但不传 user_id → 全局未超限，应该通过
        check_queue_limits("", repo)


-# ---------------------------------------------------------------------------
-# safe_enqueue_generation_task 限流集成测试（入队前用 >，包含当前任务）
-# ---------------------------------------------------------------------------
-
-
 class TestSafeEnqueueWithLimits:
-    """安全入队函数的限流功能测试。"""
+    """safe_enqueue_generation_task：用户超限仅 warning 仍入队；全局超限硬拒。"""

    def test_normal_task_enqueues_successfully(self, mock_celery):
-        """正常任务入队成功，返回 True。"""
        repo = MockRepository(user_pending=0, global_pending=0)
        task = MockTask("task-1")
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
        assert result is True
        mock_celery.assert_called_once_with("worker.generate_video", args=["task-1"])
-        # 成功入队后持久化 celery 消息 ID（#1714：孤儿清理据此 revoke/清队列）
        assert len(repo.updated_tasks) == 1
        assert task.celery_task_id

-    def test_user_limit_rejected_with_failed_status(self, mock_celery):
-        """用户超限：任务标记为 failed，抛 UserPendingLimitExceeded。"""
-        repo = MockRepository(user_pending=5, global_pending=5)
+    def test_user_limit_exceeded_still_enqueues(self, mock_celery, caplog):
+        """#2098: 用户远超软上限仍入队，任务不被标记 failed。"""
+        import logging
+
+        caplog.set_level(logging.WARNING)
+        repo = MockRepository(user_pending=100, global_pending=5)
        task = MockTask("task-1")
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-        mock_celery.assert_not_called()
-        assert task.status == "failed"
-        assert "限流" in task.error_message
-        assert len(repo.updated_tasks) == 1
+        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
+        assert result is True
+        mock_celery.assert_called_once()
+        assert task.status == "pending"  # 没被标记 failed
+        assert any("超过软上限" in r.message for r in caplog.records)

    def test_user_at_limit_still_passes(self, mock_celery):
-        """用户 pending 刚好等于上限：入队前检查用 >，包含当前任务，刚好到上限不算超。
-
-        与预检查的 >= 语义一致：预检查时 pending=3 拒绝（不能再加新的），
-        但 safe_enqueue 被调用时任务已是 pending（就是第3个），
-        pending=3 不满足 >3，所以通过。
-        """
-        repo = MockRepository(user_pending=3, global_pending=5)
+        repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5)
        task = MockTask("task-1")
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
        assert result is True
        mock_celery.assert_called_once()

-    def test_user_one_over_limit_rejected(self, mock_celery):
-        """用户 pending = limit + 1：超限被拒。"""
-        repo = MockRepository(user_pending=4, global_pending=5)
-        task = MockTask("task-1")
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-        mock_celery.assert_not_called()
-
    def test_global_limit_rejected_with_failed_status(self, mock_celery):
-        """全局超限：任务标记为 failed，抛 GlobalQueueFull。"""
        repo = MockRepository(user_pending=1, global_pending=21)
        task = MockTask("task-1")
        with pytest.raises(GlobalQueueFull):
@@ -211,7 +152,6 @@ class TestSafeEnqueueWithLimits:
        assert len(repo.updated_tasks) == 1

    def test_global_at_limit_still_passes(self, mock_celery):
-        """全局 pending 刚好等于上限：入队前检查用 >，包含当前任务，刚好到上限不算超。"""
        repo = MockRepository(user_pending=1, global_pending=20)
        task = MockTask("task-1")
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
@@ -219,7 +159,6 @@ class TestSafeEnqueueWithLimits:
        mock_celery.assert_called_once()

    def test_no_user_id_skips_user_limit(self, mock_celery):
-        """不传 user_id 时跳过用户级限流，只做全局检查。"""
        repo = MockRepository(user_pending=10, global_pending=5)
        task = MockTask("task-1")
        result = safe_enqueue_generation_task(task, repo, user_id="")
@@ -227,7 +166,6 @@ class TestSafeEnqueueWithLimits:
        mock_celery.assert_called_once()

    def test_no_user_id_still_checks_global(self, mock_celery):
-        """不传 user_id 时全局超限仍然被拦。"""
        repo = MockRepository(user_pending=10, global_pending=25)
        task = MockTask("task-1")
        with pytest.raises(GlobalQueueFull):
@@ -235,121 +173,74 @@ class TestSafeEnqueueWithLimits:
        mock_celery.assert_not_called()

    def test_default_limits_match_constants(self, mock_celery):
-        """默认配置与导出常量一致。"""
-        # 刚好在默认限制内（limit - 1）
        repo = MockRepository(user_pending=2, global_pending=19)
        task = MockTask("task-1")
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
        assert result is True

-    def test_update_failure_does_not_crash(self, mock_celery):
-        """repository.update 失败也不崩溃，异常继续向上抛。"""
-
-        class BadRepo(MockRepository):
-            def update(self, task):
-                raise RuntimeError("db down")
-
-        repo = BadRepo(user_pending=5, global_pending=5)
-        task = MockTask("task-1")
-        # 仍然抛 UserPendingLimitExceeded，不会被 update 失败掩盖
-        with pytest.raises(UserPendingLimitExceeded):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-        mock_celery.assert_not_called()
-        # 任务状态还是变了（内存里改了）
-        assert task.status == "failed"
-
-
-# ---------------------------------------------------------------------------
-# 入队后最终校验（并发竞态兜底）测试
-# ---------------------------------------------------------------------------
-

 class TestPostEnqueueFinalCheck:
-    """入队后最终校验：模拟并发场景，Celery发送后计数增加被兜住。"""
+    """入队后校验：仅全局超限回滚；用户超限仅 warning。"""

    def test_post_enqueue_global_overflow_rollback(self, mock_celery):
-        """并发场景：入队前检查通过，但发送Celery后全局计数超限 → 回滚为failed。
-
-        模拟两个请求同时通过入队前检查（都查到 global=19），
-        都创建了任务（DB里变成 21），先发送Celery的那个在最终校验时被兜住。
-        """
-        repo = MockRepository(user_pending=1, global_pending=20)  # 入队前：20 > 20？否
+        repo = MockRepository(user_pending=1, global_pending=20)
        task = MockTask("task-1")

-        # 模拟发送Celery后，另一个并发请求也创建了任务，全局变成21
        def side_effect(*args, **kwargs):
            repo.set_pending(global_pending=21)

        mock_celery.side_effect = side_effect
-
        with pytest.raises(GlobalQueueFull) as exc_info:
            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        # Celery 确实发出去了（兜底不撤销 Celery，只回滚 DB 状态）
        mock_celery.assert_called_once()
-        # 任务被标记为 failed
        assert task.status == "failed"
        assert "入队后" in task.error_message
        assert exc_info.value.pending_count == 21
        assert len(repo.updated_tasks) == 1

-    def test_post_enqueue_user_overflow_rollback(self, mock_celery):
-        """并发场景：入队前检查通过，但发送Celery后用户计数超限 → 回滚为failed。"""
-        repo = MockRepository(user_pending=3, global_pending=5)  # 入队前：3 > 3？否
+    def test_post_enqueue_user_overflow_no_rollback(self, mock_celery, caplog):
+        """#2098: 入队后用户超软上限仅 warning，不回滚。"""
+        import logging
+
+        caplog.set_level(logging.WARNING)
+        repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5)
        task = MockTask("task-1")

        def side_effect(*args, **kwargs):
-            repo.set_pending(user_pending=4)
+            repo.set_pending(user_pending=USER_PENDING_LIMIT + 1)

        mock_celery.side_effect = side_effect
-
-        with pytest.raises(UserPendingLimitExceeded) as exc_info:
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        mock_celery.assert_called_once()
-        assert task.status == "failed"
-        assert "入队后" in task.error_message
-        assert exc_info.value.user_id == "user-1"
-        assert exc_info.value.pending_count == 4
-
-    def test_post_enqueue_global_priority_over_user(self, mock_celery):
-        """入队后校验：全局和用户都超限时，优先抛全局异常。"""
-        repo = MockRepository(user_pending=3, global_pending=20)
-        task = MockTask("task-1")
-
-        def side_effect(*args, **kwargs):
-            repo.set_pending(user_pending=5, global_pending=22)
-
-        mock_celery.side_effect = side_effect
-
-        with pytest.raises(GlobalQueueFull):
-            safe_enqueue_generation_task(task, repo, user_id="user-1")
-
-        assert task.status == "failed"
-
-    def test_post_enqueue_no_change_still_passes(self, mock_celery):
-        """入队后计数没变 → 正常通过，不回滚。"""
-        repo = MockRepository(user_pending=2, global_pending=10)
-        task = MockTask("task-1")
-
        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
-
        assert result is True
        mock_celery.assert_called_once()
-        assert task.status == "pending"  # 状态没变
-        # 入队成功后持久化 celery_task_id（#1714），业务状态不变
+        assert task.status == "pending"  # 不回滚
+        assert any("超软上限(入队后)" in r.message for r in caplog.records)
+
+    def test_post_enqueue_no_change_still_passes(self, mock_celery):
+        repo = MockRepository(user_pending=2, global_pending=10)
+        task = MockTask("task-1")
+        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
+        assert result is True
+        mock_celery.assert_called_once()
+        assert task.status == "pending"
        assert len(repo.updated_tasks) == 1
        assert task.celery_task_id

    def test_post_enqueue_no_user_id_skips_user_check(self, mock_celery):
-        """不传 user_id 时，入队后校验也跳过用户级，只查全局。"""
        repo = MockRepository(user_pending=10, global_pending=5)
        task = MockTask("task-1")

        def side_effect(*args, **kwargs):
-            repo.set_pending(user_pending=15, global_pending=5)  # 用户超限但全局没超
+            repo.set_pending(user_pending=15, global_pending=5)

        mock_celery.side_effect = side_effect
-
        result = safe_enqueue_generation_task(task, repo, user_id="")
-        assert result is True  # 用户级不检查，全局没超限 → 通过
+        assert result is True
+
+
+def test_user_pending_limit_exceeded_class_still_exists():
+    """UserPendingLimitExceeded 保留用于兼容历史 import/except（#2098 后不再主动 raise）。"""
+    exc = UserPendingLimitExceeded("u1", 5, 3)
+    assert exc.user_id == "u1"
+    assert exc.pending_count == 5
+    assert exc.limit == 3
@@ -0,0 +1,99 @@
+"""爆款视频 DB 模型单元测试（#2039 PR1：DB + migration）。
+
+验证：
+- 3 张新表可在内存 SQLite 上创建
+- 默认值与基本 CRUD 正常
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.models import (
+    Base,
+    ViralVideoJobModel,
+    ViralVideoPromptTemplateModel,
+    ViralVideoStyleTemplateModel,
+)
+
+
+def _make_session():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    return sessionmaker(bind=engine)()
+
+
+class TestViralVideoJobModel:
+    def test_create_and_get(self):
+        session = _make_session()
+        job = ViralVideoJobModel(
+            id="job-001",
+            user_id="user-001",
+            images=["https://img.com/1.jpg"],
+            industry="美妆",
+            duration=60,
+        )
+        session.add(job)
+        session.commit()
+
+        fetched = session.query(ViralVideoJobModel).filter_by(id="job-001").one()
+        assert fetched.user_id == "user-001"
+        assert fetched.images == ["https://img.com/1.jpg"]
+        assert fetched.industry == "美妆"
+        assert fetched.duration == 60
+
+    def test_default_values(self):
+        session = _make_session()
+        job = ViralVideoJobModel(id="job-002", user_id="user-002")
+        session.add(job)
+        session.commit()
+
+        fetched = session.get(ViralVideoJobModel, "job-002")
+        assert fetched.images == []
+        assert fetched.fusion_level == "ai_polish"
+        assert fetched.style_strength == "medium"
+        assert fetched.status == "pending"
+        assert fetched.credits_cost == 0
+        assert fetched.retry_count == 0
+        assert fetched.style_guide is None
+        assert fetched.intent_result is None
+
+
+class TestViralVideoStyleTemplateModel:
+    def test_create_and_get(self):
+        session = _make_session()
+        tpl = ViralVideoStyleTemplateModel(
+            id="tpl-001",
+            name="快节奏",
+            style_config={"cut_speed": "fast"},
+            sort_order=1,
+        )
+        session.add(tpl)
+        session.commit()
+
+        fetched = session.get(ViralVideoStyleTemplateModel, "tpl-001")
+        assert fetched.name == "快节奏"
+        assert fetched.style_config == {"cut_speed": "fast"}
+        assert fetched.sort_order == 1
+
+
+class TestViralVideoPromptTemplateModel:
+    def test_create_and_get(self):
+        session = _make_session()
+        tpl = ViralVideoPromptTemplateModel(
+            id="pt-001",
+            prompt_type="image_analysis",
+            name="图片分析模板",
+            content="请分析图片：{image_url}",
+            variables=["image_url"],
+        )
+        session.add(tpl)
+        session.commit()
+
+        fetched = session.get(ViralVideoPromptTemplateModel, "pt-001")
+        assert fetched.prompt_type == "image_analysis"
+        assert fetched.content == "请分析图片：{image_url}"
+        assert fetched.variables == ["image_url"]
+        assert fetched.version == 1
+        assert fetched.is_active is True
@@ -65,10 +65,24 @@ def test_generate_video_preserves_original_function():
    assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'"


-def test_sync_task_config_to_plan_is_plain_function():
-    """Helper must NOT be registered as a Celery task."""
-    from worker_app.tasks.generation import _sync_task_config_to_plan
+def test_build_task_config_override_is_plain_function():
+    """#2098: 原 _sync_task_config_to_plan 已拆分为 _build_task_config_override + _download_voice_for_task，
+    均为普通函数，不应被注册为 Celery task。"""
+    from worker_app.tasks.generation import _build_task_config_override, _download_voice_for_task

-    assert not hasattr(
-        _sync_task_config_to_plan, "run"
-    ), "_sync_task_config_to_plan must be a plain function, not a Celery task"
+    for fn in (_build_task_config_override, _download_voice_for_task):
+        assert not hasattr(fn, "run"), f"{fn.__name__} must be a plain function, not a Celery task"
+
+    # Bug A: override 对 title_config 做 key 归一化 (font_size→size, font_color→color)
+    override = _build_task_config_override(
+        {
+            "title_config": {"font_size": 48, "font_color": "#ff0000", "text": "hi"},
+            "bgm_config": {"url": "http://x/bgm.mp3"},
+            "output_width": 1080,
+            "output_height": 1920,
+        }
+    )
+    assert override["title"]["size"] == 48
+    assert override["title"]["color"] == "#ff0000"
+    assert override["bgm"]["url"] == "http://x/bgm.mp3"
+    assert override["export"]["resolution"] == "1080x1920"