feat(generate): 批量生成支持>3视频排队提交 + 两列标题 + 独立配音 (#2096 )

- P0: 批量生成由一次性提交count=N改为串行count=1逐任务提交；遇到 429 USER_QUEUE_FULL / 503 SYSTEM_QUEUE_FULL 自动读取 estimated_wait_seconds 后重试，不再弹错误阻止提交；新增 useGenerationPolling.pollBatchTaskQueued 支持增量排队轮询（不触发整体 onComplete）；卡片新增 queued 状态（时钟图标+排队文案）；单个任务提交失败不阻断其余任务，失败卡片独立展示错误+重试按钮； generating 状态改为所有任务到达终态(completed/awaiting_cover/failed)且无queued/running才关闭。 - P1: Step4 批量标题改为 CSS Grid 两列布局（gridTemplateColumns: repeat(2,minmax(0,1fr))），单字段 maxWidth 由 640 改为 100%，纵向空间节省约一半。 - P1: Step3（标题页）集成 Step3VoiceWithMode，批量模式下支持共用/独立配音切换；独立配音时每个视频可单独选择音色，默认继承全局配音； GenerateStepContent props 已从 GeneratePage 透传，之前组件只定义未渲染，本次补齐 JSX 引用。 - BatchTaskState.status 新增 queued 枚举； - useGenerateVideo 重构 generate 函数，拆分 buildBasePayload/submitOne 逻辑，新增 cancelledRef/queueTimersRef 做 unmount 清理。
2026-09-29 18:05:20 +08:00
21 changed files with 1068 additions and 998 deletions
@@ -1,100 +0,0 @@
-"""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,7 +11,9 @@ 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,
 )
@@ -300,17 +302,26 @@ def create_preview_generation_task(
        count,
    )

-    # 预检查队列限流（按变体总数计）——仅保留全局硬上限，用户上限改为软 warning 在 safe_enqueue 内处理（#2098）
-    global_pending = generation_task_repository.count_pending_total()
-    if global_pending + count > GLOBAL_PENDING_LIMIT:
+    # 预检查队列限流（按变体总数计）
+    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,
-            detail=build_rate_limit_detail(
-                GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT),
-                generation_task_repository,
-                scope="global",
-            ),
-        )
+            detail=build_rate_limit_detail(e, generation_task_repository, scope="global"),
+        ) from e

    # 确定视频比例：优先前端传入，否则从模板 mode 推断
    video_ratio = request.video_ratio or ""
@@ -578,6 +589,9 @@ 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
@@ -589,6 +603,11 @@ 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,6 +7,7 @@ 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,
@@ -49,98 +50,12 @@ 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", "演出", "乐队", "钢琴", "吉他", "节奏"},
 }
@@ -162,7 +77,6 @@ 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__)
@@ -279,13 +193,10 @@ 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)
@@ -683,13 +594,21 @@ 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,
@@ -1004,10 +923,14 @@ def create_generation_task(
                else:
                    failed_tasks.append(task)
            except UserPendingLimitExceeded as _e:
-                # Bug B #2098: 用户级限流已改为软限制，此分支理论上不再触发；
-                # 极端并发兜底仍入队（safe_enqueue 内部会打 warning 日志），不 429 拒绝
-                logger.warning("[生成任务] 用户 pending 超软限制，仍允许入队: task_id=%s", task.id)
-                created_tasks.append(task)
+                # 兜底：如果预检查后又并发提交了，在这里也拦住
+                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
            except GlobalQueueFull as _e:
                failed_tasks.append(task)
                if not created_tasks:
@@ -1158,8 +1081,10 @@ def confirm_generation(
        ):
            logger.warning("[确认生成] 入队失败: task_id=%s", new_task.id)
    except UserPendingLimitExceeded as _e:
-        # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行
-        logger.warning("[任务] 用户 pending 超软限制，任务已入队")
+        raise HTTPException(
+            status_code=429,
+            detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"),
+        ) from None
    except GlobalQueueFull as _e:
        raise HTTPException(
            status_code=503,
@@ -1338,8 +1263,22 @@ def retry_generation_task(
        raise HTTPException(status_code=409, detail="Only failed tasks can be retried")

    user_id = authenticated_user.user.id
-    # 预检查（Bug B #2098）：只保留全局 503，用户级不再硬拒
+    # 预检查：创建前判断，>= 上限就拒绝
+    user_pending = generation_task_repository.count_pending_by_user(user_id)
    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,
@@ -1382,8 +1321,10 @@ def retry_generation_task(
        ):
            logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id)
    except UserPendingLimitExceeded as _e:
-        # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行
-        logger.warning("[任务] 用户 pending 超软限制，任务已入队")
+        raise HTTPException(
+            status_code=429,
+            detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"),
+        ) from None
    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 = 20  # 单用户 pending 上限（#2098: 从 3 提到 20，支持批量任务自动排队）
+USER_PENDING_LIMIT = 3  # 单用户 pending 上限
 GLOBAL_PENDING_LIMIT = 20  # 全局 pending 上限
 WORKER_CONCURRENCY = 4  # worker 渲染并发数（infra/docker/compose.yml WORKER_CONCURRENCY 默认值）

@@ -154,18 +154,19 @@ 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: 单用户 pending 上限，默认 USER_PENDING_LIMIT
        global_pending_limit: 全局 pending 上限，默认 GLOBAL_PENDING_LIMIT

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

-    # #2098: 用户级限流改为软提示，不在预检查阶段拒绝（超额任务仍入队排队）。
-    # 真正的系统保护由全局 GLOBAL_PENDING_LIMIT 硬上限承担。
-    # UserPendingLimitExceeded 保留以兼容历史 import/except，但预检查与 safe_enqueue 均不再 raise。
+    # 再查用户级
+    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)


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

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

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

-    # Bug B #2098: 用户级限流改为软提示，不再硬拒；所有任务都入队等待 worker 自然消费。
-    # user_pending_limit 作为兜底阈值保留（默认 20），达到时打 warning 日志但仍入队，
-    # 避免极端情况下恶意用户无限堆积任务。真正的系统保护由全局 GLOBAL_PENDING_LIMIT 承担。
+    # 用户级限流检查（传了 user_id 才做）
    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:
@@ -306,18 +317,16 @@ 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
-    # 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,
-        )
+    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)

-    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,7 +11,12 @@
 *   防止长标题在窄列里溢出导致与相邻卡片进度条视觉重叠。
 */
 import React from "react"
-import { LoadingOutlined, CheckCircleFilled, CloseCircleOutlined } from "@ant-design/icons"
+import {
+  LoadingOutlined,
+  CheckCircleFilled,
+  CloseCircleOutlined,
+  ClockCircleOutlined,
+} from "@ant-design/icons"
 import type { BatchTaskState } from "../hooks/generate-video/useGenerationPolling"
 import type { GeneratedVideo } from "@/api/template-editor"

@@ -61,6 +66,11 @@ 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"
@@ -85,6 +95,21 @@ 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,6 +12,7 @@ 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"
@@ -151,6 +152,12 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
    selectedCoverTemplate,
    onSelectedCoverTemplateChange,
    onConfirmGenerate,
+    selectedVoice,
+    onSelectedVoiceChange,
+    voiceModePerVideo,
+    onVoiceModePerVideoChange,
+    voiceLibraryIds,
+    onVoiceLibraryIdsChange,
  } = props

  switch (currentStep) {
@@ -184,32 +191,46 @@ 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
-          }
-        />
+        <>
+          <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}
+          />
+        </>
      )
    case 4:
      return (
@@ -157,7 +157,7 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
          />
        </div>
      ) : (
-        /* ── 批量：N 个独立标题输入框 ── */
+        /* ── 批量：N 个独立标题输入框（两列布局 #2096） ── */
        <div className="xx-batch-titles">
          <div
            style={{
@@ -170,17 +170,25 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
            为每个视频输入独立标题。标题样式（字体/颜色/位置）全局统一。
          </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
+            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>
        </div>
      )}

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

-  return { startPolling, startPollingBatch, retryTask, clearTimer }
+  /**
+   * 批量队列模式：逐任务追加到轮询队列（支持串行提交、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 }
 }
@@ -2,10 +2,12 @@
 * 视频生成 Hook
 * 封装视频生成的核心逻辑、状态管理、轮询等
 */
-import { useState, useCallback, useEffect } from "react"
+import { useState, useCallback, useEffect, useRef } 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"
@@ -15,6 +17,26 @@ 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

@@ -31,6 +53,22 @@ 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 || []
@@ -58,7 +96,6 @@ 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)
@@ -81,21 +118,30 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
    [onGenerationSuccess],
  )
  const handleFailed = useCallback((errorMsg: string) => {
-    setGenerating(false)
    setGenerateError(errorMsg)
  }, [])

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

-  const { startPolling, startPollingBatch, retryTask, clearTimer } = useGenerationPolling({
+  const { startPolling, pollBatchTaskQueued, 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) {
@@ -123,6 +337,8 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
      return false
    }

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

+    const basePayload = buildBasePayload()
+    const assetIds = basePayload.asset_ids
+    const isBatch = (props.previewCount || 1) > 1
+
    try {
-      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。
+      // 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 {
@@ -159,221 +366,185 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
        }
      }

-      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 {
+      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，请稍后在任务列表查看")
+          }
          setCurrentTaskId(taskIds[0])
          startPolling(taskIds[0])
+        } catch (err) {
+          hide()
+          throw err
        }
-      } catch (err) {
-        hide()
-        throw err
+        return true
      }
+
+      /* ── 批量：支持任意数量视频，按队列容量串行提交，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
    }
-    return true
-  }, [props, clearTimer, startPolling, startPollingBatch, selectedTemplate])
+  }, [
+    props,
+    clearTimer,
+    startPolling,
+    selectedTemplate,
+    buildBasePayload,
+    handleBatchTaskUpdate,
+    clearQueueTimers,
+    pollBatchTaskQueued,
+  ])

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

  const dismissError = useCallback(() => {
@@ -30,16 +30,11 @@ 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
-# 与 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 黑色细描边）
+# 与前端 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 基准，与前端字幕面板默认对齐）


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

    # ── 解析 subtitle_config ──
    s_cfg = dict(subtitle_config) if isinstance(subtitle_config, dict) else {}
@@ -608,21 +595,8 @@ 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)
-    # subtitle stroke/bold：先解析用户 stroke，再按 bold 默认补描边
-    s_borderw = 0
+    s_borderw = max(1, _scale_title_len(2, output_width))  # 字幕默认描边保证可读性
    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,7 +132,6 @@ 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。

@@ -218,7 +217,6 @@ 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:
@@ -394,7 +392,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, *, bgm_override: dict | None = None) -> str | None:
+    def _prepare_bgm(self, plan, work_dir: Path, plan_id: str) -> str | None:
        """准备 BGM 音频文件（从 plan.config.bgm 读取配置）。

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

        plan_config = plan.config 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 覆盖，防并发竞态
+        bgm_config = plan_config.get("bgm", {}) or {}

        if not bgm_config.get("enabled", False):
            return None
@@ -571,7 +567,6 @@ 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 + 渲染 + 缩略图 + 上传）。

@@ -585,9 +580,8 @@ class RenderAdapter:
        Returns:
            RenderAdapterResult
        """
-        # 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)
+        # 1. 准备 BGM
+        bgm_path = self._prepare_bgm(plan, work_dir, plan_id)

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

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

-        # 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"])
+        # 3. 读取输出分辨率
+        export_config = plan_config.get("export", {}) or {}
        if not isinstance(export_config, dict):
            export_config = {}
        output_width, output_height = _parse_resolution(export_config.get("resolution"))
@@ -623,7 +615,6 @@ 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,7 +158,6 @@ 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
@@ -171,9 +170,6 @@ 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)
@@ -184,28 +180,6 @@ 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（向后兼容）。"""
@@ -457,7 +431,7 @@ class UnifiedRenderService:
            has_audio = pass_through_has_audio
            # 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频
            if self.bgm_path and pass_through_has_audio:
-                config = self._effective_config()
+                config = self.plan.config or {}
                bgm_config = config.get("bgm", {}) or {}
                if bgm_config.get("enabled", False):
                    ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
@@ -497,7 +471,7 @@ class UnifiedRenderService:
                            "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
                        )
        else:
-            config = self._effective_config()
+            config = self.plan.config or {}
            bgm_config = config.get("bgm", {}) or {}
            if not isinstance(bgm_config, dict):
                bgm_config = {}
@@ -722,7 +696,7 @@ class UnifiedRenderService:
        Returns:
            ASS 文件路径，没有字幕时返回 None
        """
-        config = self._effective_config()
+        config = self.plan.config or {}
        # #1901 统一读 "title"，兼容老数据 "title_config"
        title_cfg = config.get("title", {}) or {}
        if not isinstance(title_cfg, dict) or not (title_cfg.get("text") or "").strip():
@@ -907,7 +881,7 @@ class UnifiedRenderService:
        Returns:
            是否成功添加了配音音轨
        """
-        config = self._effective_config()
+        config = self.plan.config or {}
        tts_cfg = config.get("tts", {}) or {}
        if not isinstance(tts_cfg, dict):
            tts_cfg = {}
@@ -2301,7 +2275,7 @@ class UnifiedRenderService:
        try:
            from video_processing import gpu_direct_pipeline as gdp

-            cfg = self._effective_config()
+            cfg = self.plan.config or {}
            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,49 +616,69 @@ def _precompute_render_metadata(
 # ── Celery Task ──────────────────────────────────────────────────────────────


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

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

-    override: dict = {}
+    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

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

    # BGM 配置
    bgm_config = task_info.get("bgm_config") or {}
-    if isinstance(bgm_config, dict) and bgm_config:
-        override["bgm"] = copy.deepcopy(bgm_config)
+    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

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

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

+    # 配音下载
    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):
@@ -677,9 +697,6 @@ 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)
    """
@@ -688,10 +705,8 @@ def _render_from_edit_plan(

    db = SessionLocal()
    try:
-        # 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)
+        # 同步配置到 plan.config + 下载配音
+        voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db)

        # 进度回调
        def _progress_cb(progress: float, stage: str):
@@ -707,7 +722,6 @@ 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,71 +918,3 @@ 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,41 +322,22 @@ class TestBatchPreviewRoute:
                )
        assert captured[0].voice_library_id == "legacy_voice"

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

-        # 全局硬上限：预检查即 503，不会进入后续流程
-        repo_global = MagicMock()
-        repo_global.count_pending_total.return_value = 100
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 3
+        repo.count_pending_total.return_value = 0
        with pytest.raises(HTTPException) as exc:
            create_preview_generation_task(
                _make_preview_request(preview_count=5),
                authenticated_user=_make_user(),
-                generation_task_repository=repo_global,
+                generation_task_repository=repo,
                db=MagicMock(),
            )
-        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
+        assert exc.value.status_code == 429

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

-    def test_user_pending_limit_no_longer_rejects(self):
-        """#2098: 用户待处理任务超限不再 429 拒绝（预检查仅全局 503）。"""
+    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
+
        from fastapi import HTTPException

-        repo = MagicMock()
-        repo.count_pending_by_user.return_value = 100
-        repo.count_pending_total.return_value = 5
-        with pytest.raises(HTTPException) as exc:
+        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（用户级硬拒已移除）也不是 503（全局未超限）
-        assert exc.value.status_code != 429
-        assert exc.value.status_code != 503
+        assert exc_info.value.status_code == 429

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

-    def test_enqueue_user_limit_exception_no_longer_returns_429(self):
-        """#2098: 即使 safe_enqueue 模拟抛 UserPendingLimitExceeded，也不再触发 429 整体拒绝，
-        变体被标记 failed 后正常返回响应。"""
+    def test_enqueue_raises_user_limit(self):
+        """safe_enqueue 抛出 UserPendingLimitExceeded → 429"""
        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(reason="", **_kw):
+        def _set_failed_limit(error_message="", **_kwargs):
            task.status = GenerationTaskStatus.FAILED
-            task.error_message = reason
+            task.error_message = error_message or "待处理任务超限"

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

    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、48@720 按 width 缩放到 85、y=89（50@720p baseline）。"""
+        """不传 title_config 时：白字、top、36@720 按 width 缩放到 64、y=71（PAD16+margin24）。"""
        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=89（scale(50)=89，与 CPU/vfb 一致），不含 h-th
-        assert "y=89" in fc
+        # 默认 position=top → y=71（scale(16+24)=71），不含 h-th
+        assert "y=71" in fc
        assert "h-th" not in fc
-        # 默认字号 48@720 经 1280/720 缩放 = 85
-        assert "fontsize=85" in fc
+        # 默认字号 36@720 经 1280/720 缩放 = 64
+        assert "fontsize=64" in fc
        assert "text='默认标题'" in fc
-        # 默认粗体：黑色细描边 borderw=2@720（scale=4），与 CPU vfb 一致避免重影
-        assert "borderw=4" in fc
-        assert "bordercolor=0x000000" in fc
+        # 默认粗体：drawtext 无原生粗体，用同色描边 borderw=1@720（1280/720→scale=2）模拟
+        assert "borderw=2" in fc
+        assert "bordercolor=0xffffff" 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 位置默认 margin 50@720p → scale=89（未传 margin_top 用默认）
-        assert "y=89" in fc
+        # top 位置 y=71（scale(PAD16+margin_top24)=71）
+        assert "y=71" 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=89
+        # shadow offset 3@720 经 1280/720 缩放 = 5；默认 position=top → y=71
        assert "x=(w-text_w)/2+5" in joined
-        assert "y=89+5" in joined
+        assert "y=71+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=89" not in fc
+        assert "y=71" not in fc

    def test_title_margin_top_respected(self, monkeypatch):
-        """margin_top 透传：100@720，叠加默认 50 → 150@720 → scale=267@1280。"""
+        """margin_top 透传：100@720 → scale(16+100)=206@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)
-        # 默认 50 + margin_top100 = 150@720 → scale=267
-        assert "y=267" in fc
+        # scale(PAD16 + margin_top100) = 116*1280/720 = 206
+        assert "y=206" 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=2@720 → scale=4，与 CPU vfb 一致
-        assert "borderw=4" in fc
-        assert "bordercolor=0x000000" in fc
+        # bold=True 默认：同色描边 width=1@720 → scale=2，颜色与文字相同（白色）
+        assert "borderw=2" in fc
+        assert "bordercolor=0xffffff" 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 50@720 → scale=89
-        assert "y=h-th-89" in fc
+        # bottom margin PAD16+24=40 → scale=71
+        assert "y=h-th-71" 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 48@720 → 1080w = 72
-        assert "fontsize=72" in fc
-        # top margin 50@720 → 75
-        assert "y=75" in fc
+        # default size 36@720 → 1080w = 54
+        assert "fontsize=54" in fc
+        # top margin 40@720 → 60
+        assert "y=60" 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 50@720 → 89
-        assert "y=h-th-89" in fc
+        # bottom margin 60@720 → 107
+        assert "y=h-th-107" 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 50@720 → 89，stroke 2@720 → 4
+        # 标题：size 50@720 → 89，top margin 40@720 → 71，stroke 2@720 → 4
        assert "text='主标题'" in fc
        assert "fontcolor=0xffff00" in fc
        assert "fontsize=89" in fc
-        assert "y=89" in fc
+        assert "y=71" in fc
        assert "borderw=4" in fc
-        # 字幕：size 24@720 → 43，bottom margin 50@720 → 89（默认不加粗）
+        # 字幕：size 24@720 → 43，bottom margin 60@720 → 107
        assert "text='成片全字幕'" in fc
        assert "fontsize=43" in fc
-        assert "y=h-th-89" in fc
+        assert "y=h-th-107" in fc
        assert "fontcolor=0xffffff" in fc
-        # subtitle 默认 bold=False，无额外描边（用户未开 stroke）
+        # subtitle 默认 bold=True 也会加同色描边（width=1→2）；但若主标题已有 stroke 不影响字幕独立 borderw=4（s 默认 2 像素@720→scale=4）
        # 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 黑色细描边模拟加粗（与 CPU vfb 一致，避免重影）。"""
+        """bold=True 时若原无描边，自动加 borderw=1 用同色描边模拟加粗。"""
        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)
-        # 仿粗用黑色细描边 2@720→scale=4，不跟文字色（避免同色描边重影）
-        assert "borderw=4" in fc
-        assert "bordercolor=0x000000" in fc
+        # 加粗模拟 borderw>=1，且 bordercolor 跟字体色一致（0xff0000）
+        assert "borderw=" in fc
+        assert "bordercolor=0xff0000" in fc

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

 from __future__ import annotations

@@ -14,8 +14,12 @@ 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
@@ -39,12 +43,19 @@ 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)
@@ -56,145 +67,210 @@ class TestCheckQueueLimits:
        with pytest.raises(GlobalQueueFull):
            check_queue_limits("user-1", repo)

-    def test_user_at_limit_no_longer_raises(self):
+    def test_user_at_limit_raises(self):
+        """达到用户上限即拒绝."""
        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT)
-        check_queue_limits("user-1", repo)
+        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

-    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_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_empty_user_id_skips_user_check(self):
+        """user_id 为空时跳过用户级检查."""
        repo = MockRepository(global_count=5, user_count=999)
+        # 不抛异常 = 通过（只检查全局）
        check_queue_limits("", repo)

-    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)
+    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 ──────────────────────────────────────────


 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_still_enqueues(self, mock_celery, caplog):
-        import logging
-
-        caplog.set_level(logging.WARNING)
-        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100)
+    def test_precheck_user_over_marks_failed(self, mock_celery):
+        """入队前用户超限：标记 failed，抛异常."""
+        repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 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()
-        task.mark_failed.assert_not_called()
-        assert any("超过软上限" in r.message for r in caplog.records)
+
+        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()

    @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_does_not_rollback(self, mock_celery, caplog):
-        import logging
-
-        caplog.set_level(logging.WARNING)
+    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
+
        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()
-        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)
+
+        with pytest.raises(UserPendingLimitExceeded):
+            safe_enqueue_generation_task(task, repo, user_id="user-1")
+
+        task.mark_failed.assert_called_once()

    @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_global_limit_in_enqueue(self, mock_celery):
-        repo = MockRepository(global_count=15, user_count=999)
+    def test_custom_limits_in_enqueue(self, mock_celery):
+        """自定义限流阈值用于入队检查."""
+        repo = MockRepository(global_count=5, user_count=5)
        task = make_mock_task()
-        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
+
+        # 默认阈值下用户 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,
+        )
        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)
@@ -205,6 +281,3 @@ 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,8 +19,17 @@ 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
@@ -38,6 +47,7 @@ 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:
@@ -57,39 +67,58 @@ 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():
-    """#2098: USER_PENDING_LIMIT 从 3 提到 20 作为软上限；GLOBAL_PENDING_LIMIT 保持 20 为硬上限。"""
-    assert USER_PENDING_LIMIT == 20
+    """限流阈值常量已导出，供业务代码引用。"""
+    assert USER_PENDING_LIMIT == 3
    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_no_longer_raises(self):
-        """#2098: 用户 pending 超过软上限不再抛异常。"""
-        repo = MockRepository(user_pending=100, 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_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_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_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)
@@ -97,52 +126,82 @@ 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_exceeded_still_enqueues(self, mock_celery, caplog):
-        """#2098: 用户远超软上限仍入队，任务不被标记 failed。"""
-        import logging
-
-        caplog.set_level(logging.WARNING)
-        repo = MockRepository(user_pending=100, global_pending=5)
+    def test_user_limit_rejected_with_failed_status(self, mock_celery):
+        """用户超限：任务标记为 failed，抛 UserPendingLimitExceeded。"""
+        repo = MockRepository(user_pending=5, 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()
-        assert task.status == "pending"  # 没被标记 failed
-        assert any("超过软上限" in r.message for r in caplog.records)
+        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

    def test_user_at_limit_still_passes(self, mock_celery):
-        repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5)
+        """用户 pending 刚好等于上限：入队前检查用 >，包含当前任务，刚好到上限不算超。
+
+        与预检查的 >= 语义一致：预检查时 pending=3 拒绝（不能再加新的），
+        但 safe_enqueue 被调用时任务已是 pending（就是第3个），
+        pending=3 不满足 >3，所以通过。
+        """
+        repo = MockRepository(user_pending=3, 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):
@@ -152,6 +211,7 @@ 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")
@@ -159,6 +219,7 @@ 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="")
@@ -166,6 +227,7 @@ 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):
@@ -173,74 +235,121 @@ 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:
-    """入队后校验：仅全局超限回滚；用户超限仅 warning。"""
+    """入队后最终校验：模拟并发场景，Celery发送后计数增加被兜住。"""

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

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

        mock_celery.side_effect = side_effect
-        result = safe_enqueue_generation_task(task, repo, user_id="user-1")
-        assert result is True
+
+        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 == "pending"  # 不回滚
-        assert any("超软上限(入队后)" in r.message for r in caplog.records)
+        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"
+        assert task.status == "pending"  # 状态没变
+        # 入队成功后持久化 celery_task_id（#1714），业务状态不变
        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
-
-
-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
+        assert result is True  # 用户级不检查，全局没超限 → 通过
@@ -1,99 +0,0 @@
-"""爆款视频 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,24 +65,10 @@ def test_generate_video_preserves_original_function():
    assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'"


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

-    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"
+    assert not hasattr(
+        _sync_task_config_to_plan, "run"
+    ), "_sync_task_config_to_plan must be a plain function, not a Celery task"