Merge pull request 'fix(worker+api): batch title vfb alignment, race-free config, auto-queue (3 bugs)' (#2097 ) from fix/gpu-direct-title-vfb-alignment into develop

style: auto-format with black + isort + ruff + prettier [skip ci-format-check]
test(api+worker): align unit tests with #2098 soft user limit + override refactor
2026-09-29 19:51:39 +08:00 · 2026-09-29 11:17:23 +00:00 · 2026-09-29 19:06:12 +08:00 · 2026-09-29 18:31:05 +08:00 · 2026-09-29 18:08:18 +08:00
13 changed files with 467 additions and 541 deletions
@@ -11,9 +11,7 @@ from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
 from app.core.task_enqueue import (
    GLOBAL_PENDING_LIMIT,
-    USER_PENDING_LIMIT,
    GlobalQueueFull,
-    UserPendingLimitExceeded,
    build_rate_limit_detail,
    safe_enqueue_generation_task,
 )
@@ -302,26 +300,17 @@ def create_preview_generation_task(
        count,
    )

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

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

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

+
 from packages.middleware.points_gate import points_gate

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

+    if global_over:
+        reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}"
+        exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
        logger.warning(
            "[队列限流] %s, task_id=%s, user_id=%s — 回滚状态为 failed",
            reason,
@@ -30,11 +30,16 @@ EDGE_CROP_MAX_PCT = 0.05
 # 均以 720p 为基准（见前端 titleCanvas.ts 注释 scale=videoWidth/720，types.ts "px @720p"），
 # 非 720p 输出时按 video_width / TITLE_SIZE_REF_WIDTH 等比缩放，保证成片位置与前端预览一致。
 TITLE_SIZE_REF_WIDTH = 720
-# 与前端 titleCanvas.ts 对齐：top 时文本 top-edge 距视频顶 = PAD(16@720p) + margin_top(默认 24@720p)
-TITLE_PAD_TOP = 16
-TITLE_DEFAULT_MARGIN_TOP = 24
-TITLE_DEFAULT_MARGIN_BOTTOM = 24
-SUBTITLE_DEFAULT_MARGIN_BOTTOM = 60  # 字幕距底边距（720p 基准，与前端字幕面板默认对齐）
+# 与 video_filter_builder.build_title_drawtext_filter（CPU 路径）和 ass_subtitle_builder 对齐：
+# - top/bottom 默认 margin 50@720p（vfb 用 _scale_title_len(50, w)，即 y=50 / y=h-th-50）
+# - margin_top 字段：前端编辑器 marginTop 滑块，叠加在默认 margin 之上（#2095 支持）
+# - PAD 概念仅用于前端 Canvas 预览；ffmpeg drawtext y 是 baseline，无 font metrics 可用，
+#   直接用统一 50@720p baseline 位置即可保持三端（GPU/CPU/前端视觉）一致。
+TITLE_DEFAULT_MARGIN_TOP = 50  # top 位置 baseline 默认距顶 50@720p（与 vfb/CPU 路径一致）
+TITLE_DEFAULT_MARGIN_BOTTOM = 50  # bottom 位置 baseline 默认距底 50@720p
+SUBTITLE_DEFAULT_MARGIN_BOTTOM = 50  # 字幕距底边距 50@720p（与 vfb 一致）
+TITLE_MARGIN_TOP_FROM_CFG_DEFAULT = 24  # 前端 marginTop 滑块默认值（用户未传时叠加 0）
+TITLE_FAUX_BOLD_WIDTH = 2  # 仿粗黑色描边宽度（与 vfb 一致，2@720p 黑色细描边）


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

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

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

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

        return asset_path_map, rendered_clip_ids, failed_clip_ids, asset_storage_map

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

        if not result.success:
@@ -322,22 +322,41 @@ class TestBatchPreviewRoute:
                )
        assert captured[0].voice_library_id == "legacy_voice"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

 from __future__ import annotations

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

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

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


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

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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

 from __future__ import annotations

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

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

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

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

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


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


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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        mock_celery.side_effect = side_effect
-
        result = safe_enqueue_generation_task(task, repo, user_id="")
-        assert result is True  # 用户级不检查，全局没超限 → 通过
+        assert result is True
+
+
+def test_user_pending_limit_exceeded_class_still_exists():
+    """UserPendingLimitExceeded 保留用于兼容历史 import/except（#2098 后不再主动 raise）。"""
+    exc = UserPendingLimitExceeded("u1", 5, 3)
+    assert exc.user_id == "u1"
+    assert exc.pending_count == 5
+    assert exc.limit == 3
@@ -65,10 +65,24 @@ def test_generate_video_preserves_original_function():
    assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'"


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

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