fix(worker): 修复 worker 镜像漏 COPY apps/api/app 目录导致转码队列不消费（Staging E2E 阻塞） (#1850)

- fix_python() 在 black/isort 之后追加 ruff check --fix --unsafe-fixes
- 目标范围与 validate_style.sh 对齐：apps packages tests（alembic 被 ruff exclude，scripts 不在检查范围）
- ruff 不可用或修复后仍有残余问题时不阻断，剩余 lint 由 style check 继续拦截
- workflow step 名同步更新

解决 F401 等 ruff 可自动修复错误无法自动推修复、开发者反复打回的问题

.gitea/workflows/ci-pipeline.yml

@@ -196,7 +196,7 @@ jobs:
     - name: Run style checks
       shell: bash
       run: bash scripts/ci/validate_style.sh
-    - name: Auto-fix formatting (black + isort)
+    - name: Auto-fix formatting (black + isort + ruff)
       if: failure()
       shell: sh
       env:

apps/api/app/api/routes/ai_avatar_render.py

@@ -198,12 +198,21 @@ def generate_avatar_smart_cover(
     if not video_url.startswith(("http://", "https://")):
         raise HTTPException(status_code=400, detail="video_url 必须是合法的 HTTP/HTTPS URL")
 
-    cover_url = generate_smart_cover(video_url, max_frames=body.max_frames)
+    try:
+        cover_url = generate_smart_cover(video_url, max_frames=body.max_frames)
+    except Exception as exc:
+        logger.error(
+            "智能封面生成异常: user=%s video_url=%s err=%s",
+            current_user.user.id, video_url[:80], exc,
+            exc_info=True,
+        )
+        cover_url = ""
+
     if not cover_url:
         return SmartCoverResponse(
             cover_url="",
             status="fallback_failed",
             message="智能抽帧失败（MediaKit 不可用或抽帧异常），请稍后重试",
         )
-    logger.info("智能封面生成成功: user=%s", current_user.user.id)
+    logger.info("智能封面生成成功: user=%s cover_url=%s", current_user.user.id, cover_url[:120])
     return SmartCoverResponse(cover_url=cover_url, status="completed")

apps/api/app/api/routes/chunked_upload.py

@@ -177,8 +177,8 @@ def _cleanup_expired_uploads() -> int:
                 meta_file.unlink()
                 cleaned += 1
                 logger.info(f"Cleaned up expired upload: {upload_id}")
-        except Exception as e:
-            logger.warning(f"Failed to cleanup upload metadata {meta_file}: {e}")
+        except Exception:
+            logger.exception("Failed to cleanup upload metadata: %s", meta_file)
 
     return cleaned
 

apps/api/app/api/routes/generation_variant_plans.py

@@ -108,7 +108,7 @@ def create_variant_plans(
             if _latest:
                 source_plan_id = _latest.id
         except Exception:
-            logger.warning("[variant-plans] 源 plan 解析失败", exc_info=True)
+            logger.exception("[variant-plans] 源 plan 解析失败")
 
     if not source_plan_id:
         raise HTTPException(
@@ -122,7 +122,7 @@ def create_variant_plans(
 
         voice_durations = _query_voice_durations(db, voices)
     except Exception:
-        logger.warning("[variant-plans] 配音时长查询失败（按占位段长选片）", exc_info=True)
+        logger.exception("[variant-plans] 配音时长查询失败（按占位段长选片）")
         voice_durations = [0.0] * request.count
 
     from app.services.edit_plan_service import EditPlanService
@@ -143,7 +143,7 @@ def create_variant_plans(
     except HTTPException:
         raise
     except Exception as e:
-        logger.error("[variant-plans] 选片异常: %s", e, exc_info=True)
+        logger.exception("[variant-plans] 选片异常")
         raise HTTPException(status_code=500, detail="选片失败，请稍后重试") from e
 
     # 组装 clips 响应

apps/api/app/api/routes/lipsync.py

@@ -14,7 +14,6 @@ import logging
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import (
-    get_cosyvoice_service,
     get_db_session,
     get_voice_clone_profile_repository,
 )
@@ -24,8 +23,6 @@ from app.services.mediakit_client import MediaKitError
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
-from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
-
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -34,13 +31,11 @@ router = APIRouter()
 def _get_service(
     db: Session = Depends(get_db_session),
     voice_clone_repo=Depends(get_voice_clone_profile_repository),
-    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
 ) -> LipsyncService:
-    # voice_clone_repo 用于克隆音色 profile 解析；cosyvoice_service 用于 TTS 直生
-    # （TTS 合成、音色解析、错误码归一化都在 LipsyncService 内部完成）
+    # voice_clone_repo 用于克隆音色 profile 解析
+    # TTS 合成已移至 Celery 异步任务，无需同步注入 cosyvoice_service
     return LipsyncService(
         db,
-        cosyvoice_service=cosyvoice_service,
         voice_clone_repo=voice_clone_repo,
     )
 
@@ -57,8 +52,8 @@ def create_lipsync_job(
     """提交对口型任务.
 
     #1809/#1822: 前端传 {video_url, voice_id, script_text, speed?, emotion?}，
-    后端内部解析音色、调 TTS 合成音频、转存 OSS，再提交 MediaKit；
-    也支持直接传 {video_url, audio_url}。
+    后端创建任务记录（状态 tts_processing），dispatch Celery 异步任务执行 TTS 合成 + MediaKit 提交；
+    也支持直接传 {video_url, audio_url}（同步提交 MediaKit）。
     """
     try:
         job = svc.create_job(
@@ -75,21 +70,13 @@ def create_lipsync_job(
     except ValueError as exc:
         # 参数无效（如 voice_id 格式不对、文本过长等）
         raise HTTPException(status_code=400, detail=str(exc)) from exc
-    except CosyVoiceError as exc:
-        # TTS 合成基础设施失败（API/网络/认证）
-        raise HTTPException(
-            status_code=502,
-            detail={"code": "TTSSynthesisFailed", "message": str(exc)},
-        ) from exc
     except MediaKitError as exc:
-        # TTS 合成失败 / 音色无权访问 → 400/403；MediaKit 提交失败 → 502
+        # 音色无权访问 → 403；参数无效 → 400；MediaKit 提交失败 → 502
         status_code = 502
         if exc.code in ("VoiceForbidden",):
             status_code = 403
         elif exc.code in ("InvalidInput", "TTSInvalidParam", "VoiceNotReady"):
             status_code = 400
-        elif exc.code == "TTSSynthesisFailed":
-            status_code = 502
         raise HTTPException(
             status_code=status_code,
             detail={
@@ -187,13 +174,13 @@ def cancel_lipsync_job(
     current_user: AuthenticatedUser = Depends(get_current_user),
     svc: LipsyncService = Depends(_get_service),
 ):
-    """取消对口型任务（仅 pending/submitted 状态可取消）."""
+    """取消对口型任务（仅 pending/tts_processing/submitted 状态可取消）."""
     job = svc.cancel_job(job_id, current_user.user.id)
     if job is None:
         raise HTTPException(status_code=404, detail="任务不存在")
     if job.status != "cancelled":
         raise HTTPException(
             status_code=400,
-            detail=f"任务状态 {job.status} 不可取消，仅 pending/submitted 可取消",
+            detail=f"任务状态 {job.status} 不可取消，仅 pending/tts_processing/submitted 可取消",
         )
     return job

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

@@ -65,8 +65,8 @@ def _build_asset_analyses(
                 if url:
                     video_urls.append(url)
                     valid_asset_ids.append(aid)
-            except Exception as e:
-                logger.warning("获取素材URL失败: asset_id=%s error=%s", aid, str(e))
+            except Exception:
+                logger.exception("获取素材URL失败: asset_id=%s", aid)
 
         if not video_urls:
             logger.info("无可用视频素材，跳过视频理解分析")
@@ -108,7 +108,7 @@ def _build_asset_analyses(
         return analyses
 
     except Exception as e:
-        logger.warning("MediaKit 视频理解异常，将降级到无分析模式: %s", str(e))
+        logger.exception("MediaKit 视频理解异常，将降级到无分析模式: %s", e)
         return {}
 
 
@@ -177,7 +177,7 @@ def editor_ai_recommend(
         try:
             db.rollback()
         except Exception:
-            pass
+            logger.exception("db rollback failed in ai_recommend")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="AI推荐结果保存失败，请稍后重试",

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

@@ -132,8 +132,8 @@ def _build_asset_url_map(
     result: dict[str, str | None] = {}
     try:
         storage = get_storage_service()
-    except Exception:
-        logger.warning("获取存储服务失败，跳过asset_url生成")
+    except Exception as e:
+        logger.exception("获取存储服务失败，跳过asset_url生成: %s", e)
         return {aid: None for aid in asset_ids}
 
     # 批量查询所有 Asset（单次 SQL IN 查询，避免 N+1）
@@ -141,7 +141,7 @@ def _build_asset_url_map(
         assets = asset_repo.find_by_ids(unique_ids)
         asset_map = {a.id: a for a in assets}
     except Exception:
-        logger.warning("批量查询素材失败: asset_ids=%s", asset_ids, exc_info=True)
+        logger.exception("批量查询素材失败: asset_ids=%s", asset_ids)
         return {aid: None for aid in asset_ids if aid}
 
     for aid in unique_ids:
@@ -156,7 +156,7 @@ def _build_asset_url_map(
                 continue
             result[aid] = storage.get_download_url(storage_key, expires_seconds=3600)
         except Exception:
-            logger.warning("生成素材签名URL失败: asset_id=%s", aid, exc_info=True)
+            logger.exception("生成素材签名URL失败: asset_id=%s", aid)
             result[aid] = None
 
     return result
@@ -486,8 +486,8 @@ def _get_mediakit_recommendations(
                 if url:
                     video_urls.append(url)
                     valid_asset_ids.append(asset_id)
-            except Exception as e:
-                logger.warning("获取素材URL失败: asset_id=%s error=%s", asset_id, e)
+            except Exception:
+                logger.exception("获取素材URL失败: asset_id=%s", asset_id)
 
         if not video_urls:
             return {}
@@ -563,7 +563,7 @@ def _get_mediakit_recommendations(
         return recommendations
 
     except Exception as e:
-        logger.warning("MediaKit 智能选片异常，降级为随机选择: %s", e)
+        logger.exception("MediaKit 智能选片异常，降级为随机选择: %s", e)
         return {}
 
 
@@ -860,7 +860,7 @@ def create_clips_from_assets_editor(
     duplicate_warning = None
     if dup_rate > 50:
         duplicate_warning = f"查重率 {dup_rate:.1f}% 超过50%，建议更换素材或模板"
-        logger.warning(
+        logger.exception(
             "from-assets 成片查重率超标: plan_id=%s dup_rate=%.1f%%",
             plan_id,
             dup_rate,
@@ -960,8 +960,8 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
         # 尝试获取存储服务（用于生成视频 URL）
         try:
             storage = get_storage_service()
-        except Exception:
-            logger.warning("后台任务: 获取存储服务失败，跳过 SceneChange 更新")
+        except Exception as e:
+            logger.exception("后台任务: 获取存储服务失败，跳过 SceneChange 更新: %s", e)
             return
 
         # 获取 MediaKit 客户端
@@ -987,8 +987,8 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
             if storage_key and mime.startswith("video/"):
                 try:
                     video_url = storage.get_download_url(storage_key)
-                except Exception as e:
-                    logger.warning("后台任务: 获取素材URL失败: asset_id=%s error=%s", asset_id, e)
+                except Exception:
+                    logger.exception("后台任务: 获取素材URL失败: asset_id=%s", asset_id)
 
             # 构建该素材的占用区间列表（排除已更新片段）
             def _get_other_segments(asset_id_inner, clip_id_inner):
@@ -1039,12 +1039,11 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                             asset_id,
                             len(scene_changes),
                         )
-                    except Exception as cache_err:
+                    except Exception:
                         # 缓存写入失败不影响本次片段更新
-                        logger.warning(
-                            "后台任务: 场景点缓存写入失败: asset_id=%s error=%s",
+                        logger.exception(
+                            "后台任务: 场景点缓存写入失败: asset_id=%s",
                             asset_id,
-                            cache_err,
                         )
 
             # SceneChange 未获得有效结果 → 尝试 analyze_videos 作为 fallback
@@ -1124,11 +1123,10 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                                 recommended_start + clip_duration,
                                 plan_id,
                             )
-                    except Exception as me:
-                        logger.warning(
-                            "后台任务: 同步素材区间记录失败，回滚本次片段更新: clip_id=%s error=%s",
+                    except Exception:
+                        logger.exception(
+                            "后台任务: 同步素材区间记录失败，回滚本次片段更新: clip_id=%s",
                             clip.id,
-                            me,
                         )
                         db.rollback()
                         continue
@@ -1144,8 +1142,8 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                         asset_id,
                         recommended_start,
                     )
-                except Exception as ue:
-                    logger.warning("后台任务: 单个片段更新失败: clip_id=%s error=%s", clip.id, ue)
+                except Exception:
+                    logger.exception("后台任务: 单个片段更新失败: clip_id=%s", clip.id)
                     try:
                         db.rollback()
                     except Exception:
@@ -1154,9 +1152,9 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
 
         logger.info("后台任务完成: plan_id=%s 成功更新 %d 个片段", plan_id, updated_count)
 
-    except Exception as e:
+    except Exception:
         # 后台任务失败不影响已创建的片段，静默处理
-        logger.warning("后台任务异常: plan_id=%s error=%s", plan_id, e, exc_info=True)
+        logger.exception("后台任务异常: plan_id=%s", plan_id)
         if db:
             try:
                 db.rollback()

apps/api/app/api/routes/voice_clones.py

@@ -163,12 +163,12 @@ def create_voice_clone(
             celery_app.send_task("worker.process_voice_clone", args=[profile.id])
             logger.info(f"Celery task dispatched for voice clone {profile.id}")
         except Exception as e:
-            logger.error(f"Failed to dispatch Celery task: {e}")
+            logger.exception("Failed to dispatch Celery task")
             # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing
             try:
                 workflow.process_clone_failure(profile.id, f"Celery 任务调度失败: {e}")
-            except Exception as inner_e:
-                logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
+            except Exception:
+                logger.exception("Failed to mark profile as failed after dispatch error")
 
     return _to_response(profile)
 
@@ -277,12 +277,12 @@ def retry_voice_clone(
             celery_app.send_task("worker.process_voice_clone", args=[profile.id])
             logger.info(f"Celery task dispatched for voice clone retry {profile.id}")
         except Exception as e:
-            logger.error(f"Failed to dispatch Celery task: {e}")
+            logger.exception("Failed to dispatch Celery task")
             # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing
             try:
                 workflow.process_clone_failure(profile.id, f"Celery 任务调度失败: {e}")
-            except Exception as inner_e:
-                logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
+            except Exception:
+                logger.exception("Failed to mark profile as failed after dispatch error")
 
     return _to_response(profile)
 

apps/api/app/api/routes/voices.py

@@ -105,8 +105,8 @@ def _resolve_preset_preview_url(
         _preset_preview_cache[voice_id] = (audio_url, time.time())
         logger.info("Preset voice preview generated: %s", voice_id)
         return audio_url
-    except Exception as e:
-        logger.warning("Failed to generate preview for %s, using fallback: %s", voice_id, e)
+    except Exception:
+        logger.exception("Failed to generate preset voice preview: voice_id=%s", voice_id)
         return fallback_url
 
 
@@ -127,6 +127,7 @@ def _resolve_all_preset_preview_urls(
         try:
             result_map[p.voice_id] = _resolve_preset_preview_url(p.voice_id, p.preview_url, cosyvoice)
         except Exception:
+            logger.exception("Failed to resolve preset preview URL: voice_id=%s", p.voice_id)
             result_map[p.voice_id] = p.preview_url
     return result_map
 
@@ -732,7 +733,7 @@ def _find_or_create_voice_library_for_extract(*, user_id, project_repository, as
             try:
                 session.rollback()
             except Exception:
-                pass
+                logger.exception("session rollback failed in _find_or_create_voice_library")
         for lib in asset_library_repository.find_by_project(project.id):
             kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
             if kind == AssetLibraryKind.VOICE.value:

apps/api/app/services/ai_avatar_cover_service.py

@@ -15,9 +15,49 @@ import tempfile
 import uuid
 from pathlib import Path
 from typing import Optional
+from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
+# MediaKit 抽帧轮询参数（与 MediaKit API timeout=60s 对齐）
+COVER_POLL_INTERVAL = 3.0
+COVER_MAX_POLL_ATTEMPTS = 20  # 最多等 60 秒
+
+# 帧图片下载超时（秒）
+FRAME_DOWNLOAD_TIMEOUT = 20
+# 最佳帧下载超时（用于 persist）
+BEST_FRAME_DOWNLOAD_TIMEOUT = 30
+
+# 自家 OSS 私有桶 URL 重签有效期（供 MediaKit GPU worker 拉取）
+MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600
+
+
+def _sign_video_url_for_mediakit(video_url: str) -> str:
+    """如果 video_url 是自家 OSS 私有桶 URL，重新签名为长有效期预签名 URL。
+
+    MediaKit GPU worker 需要能公网访问 video_url，裸 public_url 在私有桶下会 403。
+    """
+    if not video_url:
+        return video_url
+    try:
+        from packages.shared.storage import get_shared_storage_service
+
+        storage = get_shared_storage_service()
+        public_base = getattr(storage, "public_url", "")
+        if not isinstance(public_base, str) or not public_base:
+            return video_url
+        own_host = urlparse(public_base).netloc.lower()
+        url_host = urlparse(video_url).netloc.lower()
+        if own_host and url_host == own_host:
+            # 是自家 OSS URL，重签 7 天有效期供 MediaKit 拉取
+            signed = storage.get_download_url(video_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS)
+            if signed:
+                logger.info("[数字人封面] video_url 已重签（自家 OSS 私有桶）")
+                return signed
+    except Exception:
+        logger.warning("[数字人封面] video_url 重签失败，使用原始 URL", exc_info=True)
+    return video_url
+
 
 def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
     """从视频抽取多帧并评分选最佳帧，返回最佳帧的临时 URL.
@@ -31,6 +71,10 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
     """
     if not video_url:
         return ""
+
+    # 确保 MediaKit 能访问 video_url（自家 OSS 私有桶需重签）
+    video_url = _sign_video_url_for_mediakit(video_url)
+
     try:
         from packages.shared.cover_frame_scorer import score_frames
         from packages.shared.mediakit_client import get_mediakit_client
@@ -40,13 +84,21 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
             logger.warning("[数字人封面] MediaKit 未配置，无法智能抽帧")
             return ""
 
+        logger.info(
+            "[数字人封面] 开始抽帧: video_url=%s max_frames=%d poll_interval=%.1f max_poll=%d",
+            video_url[:80],
+            max_frames,
+            COVER_POLL_INTERVAL,
+            COVER_MAX_POLL_ATTEMPTS,
+        )
+
         snapshots = mk.extract_frames(
             video_url=video_url,
             strategy="SpecifiedFrames",
             max_frames=max_frames,
-            poll_interval=2.0,
-            max_poll_attempts=5,
-            max_retries=0,
+            poll_interval=COVER_POLL_INTERVAL,
+            max_poll_attempts=COVER_MAX_POLL_ATTEMPTS,
+            max_retries=1,
         )
         if not snapshots:
             logger.warning("[数字人封面] MediaKit 未返回帧: %s", video_url[:80])
@@ -55,24 +107,26 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
         if len(snapshots) == 1:
             return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
 
-        # 下载各帧评分
+        # 使用连接池下载各帧（复用 TCP 连接，减少延迟）
         import httpx
 
         candidates = []
-        for snap in snapshots:
-            url = snap.get("image_url") or snap.get("url") or ""
-            if not url:
-                continue
-            tmp_path: Optional[str] = None
-            try:
-                resp = httpx.get(url, timeout=15, follow_redirects=True)
-                resp.raise_for_status()
-                with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
-                    tmp.write(resp.content)
-                    tmp_path = tmp.name
-                candidates.append({"image_path": tmp_path, "url": url})
-            except Exception:
-                candidates.append({"image_path": None, "url": url, "score": 0.0})
+        with httpx.Client(timeout=FRAME_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
+            for snap in snapshots:
+                url = snap.get("image_url") or snap.get("url") or ""
+                if not url:
+                    continue
+                tmp_path: Optional[str] = None
+                try:
+                    resp = client.get(url)
+                    resp.raise_for_status()
+                    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
+                        tmp.write(resp.content)
+                        tmp_path = tmp.name
+                    candidates.append({"image_path": tmp_path, "url": url})
+                except Exception as e:
+                    logger.warning("[数字人封面] 帧下载失败，跳过: url=%s err=%s", url[:80], e)
+                    candidates.append({"image_path": None, "url": url, "score": 0.0})
 
         if not candidates:
             return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
@@ -119,14 +173,16 @@ def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-
     try:
         import httpx
 
-        resp = httpx.get(frame_url, timeout=30, follow_redirects=True)
-        resp.raise_for_status()
-        if not resp.content:
-            return frame_url
+        with httpx.Client(timeout=BEST_FRAME_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
+            resp = client.get(frame_url)
+            resp.raise_for_status()
+            if not resp.content:
+                logger.warning("[数字人封面] 帧图内容为空: %s", frame_url[:80])
+                return frame_url
 
-        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
-            tmp.write(resp.content)
-            tmp_path = tmp.name
+            with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
+                tmp.write(resp.content)
+                tmp_path = tmp.name
 
         from packages.shared.storage import get_shared_storage_service
 
@@ -139,7 +195,11 @@ def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-
             content_type="image/jpeg",
         )
         logger.info("[数字人封面] 封面已转存 OSS: key=%s", cover_key)
-        return public_url or frame_url
+        # 私有桶：返回预签名 URL（前端才能加载）
+        if public_url:
+            signed = storage.get_download_url(cover_key, expires_seconds=86400)
+            return signed
+        return frame_url
     except Exception:
         logger.warning("[数字人封面] 封面转存 OSS 失败，返回原始 URL", exc_info=True)
         return frame_url

apps/api/app/services/ai_avatar_render_service.py

@@ -322,6 +322,38 @@ class AiAvatarRenderService:
             self.db.commit()
             logger.info("渲染任务完成: %s", job_id)
 
+            # 7. 自动保存成片记录到成片库
+            if job.output_video_url:
+                try:
+                    from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+                        SQLAlchemyGeneratedVideoRepository,
+                    )
+                    from packages.domain.generated_video import GeneratedVideo
+
+                    clip_name = f"AI数字人_{job_id[:8]}"
+                    clip = GeneratedVideo.create(
+                        project_id=job.project_id,
+                        generation_task_id=job.lipsync_job_id,
+                        name=clip_name,
+                        file_url=job.output_video_url,
+                        user_id=job.user_id,
+                        duration=job.output_duration or 0.0,
+                        thumbnail_url=job.output_cover_url or None,
+                        generation_params={
+                            "source": "ai_avatar_render",
+                            "render_job_id": job.id,
+                        },
+                    )
+                    video_repo = SQLAlchemyGeneratedVideoRepository(self.db)
+                    video_repo.create(clip)
+                    logger.info("成片记录已保存到成片库: clip_id=%s, render_job=%s", clip.id, job_id)
+                except Exception as clip_err:
+                    logger.warning(
+                        "自动保存成片记录失败（不影响渲染任务状态）: render_job=%s, error=%s",
+                        job_id,
+                        clip_err,
+                    )
+
         except AiAvatarRenderError as exc:
             job.status = "failed"
             job.error_message = str(exc)
@@ -376,7 +408,7 @@ class AiAvatarRenderService:
         else:
             filter_arg = ""
 
-        return f"ffmpeg {inputs} {filter_arg} -c:v libx264 -preset fast -crf 23 -y {output_path}"
+        return f"ffmpeg {inputs} {filter_arg} -c:v libx264 -preset veryfast -crf 23 -y {output_path}"
 
     def _upload_to_oss(self, local_path: str, oss_key: str) -> str:
         """上传文件到 OSS，返回 URL.

apps/api/app/services/lipsync_service.py

@@ -25,6 +25,9 @@ from app.services.mediakit_client import (
     MediaKitError,
     get_mediakit_client,
 )
+
+# Celery 异步任务：TTS 合成 + MediaKit 提交（#lipsync-speed-optimization）
+from app.tasks.lipsync_tts import tts_synthesize_and_submit
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
@@ -154,41 +157,31 @@ class LipsyncService:
         enable_video_loop: bool = False,
         project_id: str = "",
     ) -> LipsyncJobModel:
-        """创建对口型任务并提交到 MediaKit.
+        """创建对口型任务.
 
         两种输入模式：
-        - TTS 直生：voice_id + script_text（audio_url 留空），后端先合成音频
+        - TTS 直生：voice_id + script_text（audio_url 留空）
+          → 先创建 DB 记录（状态 tts_processing），再 dispatch Celery 异步任务
+            执行 TTS 合成 + MediaKit 提交。API 响应 <1s。
         - 直接音频：提供 audio_url
+          → 同步提交 MediaKit，状态直接设为 submitted。
 
         Raises:
-            MediaKitError: TTS 合成或 MediaKit 提交失败
+            MediaKitError: 参数校验失败或 MediaKit 提交失败（仅直接音频模式）
         """
-        # 0. TTS 直生模式：先合成音频（在创建 DB 记录之前完成，失败直接抛出）
+        # 0. 输入校验
         if not audio_url:
             if not (voice_id and script_text):
                 raise MediaKitError(
                     "必须提供 audio_url 或 voice_id+script_text",
                     code="InvalidInput",
                 )
-            # 预合成：用临时 job_id 命名 OSS 对象
-            pre_job_id = str(uuid.uuid4())
-            audio_url = self._synthesize_and_persist_audio(
-                user_id=user_id,
-                job_id=pre_job_id,
-                voice_id=voice_id,
-                script_text=script_text,
-                speed=speed,
-                emotion=emotion,
-            )
-
-        # #1839: 私有桶 OSS 的裸 URL / 即将过期的短预签名会让 MediaKit GPU worker 拉取时 403，
-        # 提交前统一重签长有效期；外部临时 URL（CosyVoice/MediaKit）原样透传。
-        video_url = self._sign_media_url(video_url)
-        if audio_url:
-            audio_url = self._sign_media_url(audio_url)
+            # TTS 模式：在 HTTP 请求中同步校验音色归属，快速失败
+            self._resolve_voice_id(voice_id, user_id)
 
         # 1. 创建数据库记录
         job_id = str(uuid.uuid4())
+        is_tts_mode = not bool(audio_url)
         job = LipsyncJobModel(
             id=job_id,
             user_id=user_id,
@@ -200,28 +193,53 @@ class LipsyncService:
             script_text=script_text or "",
             speed=speed,
             emotion=normalize_emotion(emotion),
-            status="pending",
+            status="tts_processing" if is_tts_mode else "pending",
         )
         self.db.add(job)
         self.db.flush()
 
-        # 3. 提交到 MediaKit
-        try:
-            result = self.client.submit_lipsync(
-                video_url=video_url,
-                audio_url=audio_url,
-                enable_video_loop=enable_video_loop,
-                client_token=job_id,  # 幂等控制
-            )
-            job.mediakit_task_id = result["task_id"]
-            job.status = "submitted"
-            job.submitted_at = datetime.now(timezone.utc)
-        except MediaKitError as exc:
-            job.status = "failed"
-            job.error_message = str(exc)
-            job.error_code = exc.code
-            logger.error("提交对口型任务失败: %s", exc)
-            raise
+        if is_tts_mode:
+            # 2a. TTS 模式：dispatch Celery 异步任务处理 TTS 合成 + MediaKit 提交
+            try:
+                tts_synthesize_and_submit.apply_async(
+                    args=(
+                        job_id,
+                        user_id,
+                        voice_id,
+                        script_text,
+                        speed,
+                        normalize_emotion(emotion),
+                    )
+                )
+            except Exception:
+                logger.warning(
+                    "Celery 任务提交失败，TTS 任务已创建但未触发执行: %s",
+                    job_id,
+                    exc_info=True,
+                )
+        else:
+            # 2b. 直接音频模式：同步签名并提交 MediaKit
+            video_url = self._sign_media_url(video_url)
+            if audio_url:
+                audio_url = self._sign_media_url(audio_url)
+                job.audio_url = audio_url
+
+            try:
+                result = self.client.submit_lipsync(
+                    video_url=video_url,
+                    audio_url=audio_url,
+                    enable_video_loop=enable_video_loop,
+                    client_token=job_id,
+                )
+                job.mediakit_task_id = result["task_id"]
+                job.status = "submitted"
+                job.submitted_at = datetime.now(timezone.utc)
+            except MediaKitError as exc:
+                job.status = "failed"
+                job.error_message = str(exc)
+                job.error_code = exc.code
+                logger.error("提交对口型任务失败: %s", exc)
+                raise
 
         self.db.commit()
         self.db.refresh(job)
@@ -360,12 +378,12 @@ class LipsyncService:
     # ── 取消任务 ──────────────────────────────────────────────────────────
 
     def cancel_job(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
-        """取消任务（仅 pending/submitted 状态可取消）."""
+        """取消任务（仅 pending/tts_processing/submitted 状态可取消）."""
         job = self.get_job(job_id, user_id)
         if job is None:
             return None
 
-        if job.status in ("pending", "submitted"):
+        if job.status in ("pending", "tts_processing", "submitted"):
             job.status = "cancelled"
             job.updated_at = datetime.now(timezone.utc)
             self.db.commit()

apps/api/app/tasks/lipsync_tts.py

@@ -0,0 +1,166 @@
+"""AI 数字人对口型 TTS 异步任务 — 将 TTS 合成从 HTTP 请求移至 Celery 后台执行.
+
+优化目标：将 create_job 的 API 响应时间从 6~35s 降到 <1s。
+任务流程：
+1. 创建新 DB session，加载 job 记录
+2. 调用 CosyVoice 合成音频
+3. 下载音频并转存到自家 OSS
+4. 更新 job 的 audio_url
+5. 签名 URL 并提交到 MediaKit
+6. 更新 job 状态为 submitted
+7. 异常时标记 job 为 failed
+"""
+
+import io
+import logging
+from datetime import datetime, timezone
+
+from app.core.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(
+    bind=True,
+    name="lipsync_tts.synthesize_and_submit",
+    max_retries=2,
+    default_retry_delay=30,
+)
+def tts_synthesize_and_submit(
+    self,
+    job_id: str,
+    user_id: str,
+    voice_id: str,
+    script_text: str,
+    speed: float,
+    emotion: str,
+):
+    """异步执行 TTS 合成 + OSS 转存 + MediaKit 提交.
+
+    在 Celery worker 中运行，不阻塞 HTTP 请求。
+    """
+    from app.services.mediakit_client import MediaKitError, get_mediakit_client
+    from sqlalchemy.orm import Session as DBSession
+
+    from packages.adapters.sqlalchemy_impl.database import SessionLocal
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+    from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
+    from packages.shared.storage import get_shared_storage_service
+    from packages.shared.url_security import safe_download_bytes
+
+    db: DBSession = SessionLocal()
+    try:
+        job = (
+            db.query(LipsyncJobModel)
+            .filter(
+                LipsyncJobModel.id == job_id,
+                LipsyncJobModel.user_id == user_id,
+            )
+            .first()
+        )
+
+        if job is None:
+            logger.error("[lipsync_tts] Job not found: job_id=%s", job_id)
+            return
+
+        # 已取消的任务不再处理
+        if job.status == "cancelled":
+            logger.info("[lipsync_tts] Job already cancelled, skipping: job_id=%s", job_id)
+            return
+
+        # 1. TTS 合成
+        try:
+            cosyvoice = CosyVoiceService()
+            result = cosyvoice.submit_synthesize_task(
+                text=script_text,
+                voice_id=voice_id,
+                speed=speed,
+                emotion=emotion,
+            )
+        except CosyVoiceError as exc:
+            logger.error("[lipsync_tts] TTS 合成失败: job_id=%s err=%s", job_id, exc)
+            job.status = "failed"
+            job.error_message = f"TTS 合成失败: {exc}"
+            job.error_code = "TTSSynthesisFailed"
+            job.updated_at = datetime.now(timezone.utc)
+            db.commit()
+            return
+        except ValueError as exc:
+            logger.error("[lipsync_tts] TTS 参数错误: job_id=%s err=%s", job_id, exc)
+            job.status = "failed"
+            job.error_message = f"TTS 参数错误: {exc}"
+            job.error_code = "TTSInvalidParam"
+            job.updated_at = datetime.now(timezone.utc)
+            db.commit()
+            return
+
+        temp_url = result.get("audio_url", "")
+        if not temp_url:
+            logger.error("[lipsync_tts] TTS 未返回音频 URL: job_id=%s", job_id)
+            job.status = "failed"
+            job.error_message = "TTS 未返回音频 URL"
+            job.error_code = "TTSNoAudio"
+            job.updated_at = datetime.now(timezone.utc)
+            db.commit()
+            return
+
+        # 2. 下载并转存到自家 OSS
+        try:
+            audio_data = safe_download_bytes(
+                temp_url,
+                purpose="lipsync_tts_audio",
+                allowed_mime_types=("audio/mpeg", "audio/mp3", "audio/wav", "audio/mp4", "audio/x-m4a"),
+                timeout=60.0,
+            )
+            storage = get_shared_storage_service()
+            storage_key = f"lipsync-tts/{user_id}/{job_id}.mp3"
+            permanent_url = storage.upload_file(io.BytesIO(audio_data), storage_key, content_type="audio/mpeg")
+            logger.info("[lipsync_tts] TTS 音频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
+            job.audio_url = permanent_url
+        except Exception as exc:
+            logger.warning("[lipsync_tts] TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s", job_id, exc)
+            job.audio_url = temp_url
+
+        db.commit()
+
+        # 3. 签名 URL 并提交到 MediaKit
+        from app.services.lipsync_service import LipsyncService
+
+        temp_service = LipsyncService.__new__(LipsyncService)
+        audio_url = temp_service._sign_media_url(job.audio_url)
+        video_url = temp_service._sign_media_url(job.video_url)
+
+        client = get_mediakit_client()
+        try:
+            mk_result = client.submit_lipsync(
+                video_url=video_url,
+                audio_url=audio_url,
+                enable_video_loop=job.enable_video_loop,
+                client_token=job_id,
+            )
+            job.mediakit_task_id = mk_result["task_id"]
+            job.status = "submitted"
+            job.submitted_at = datetime.now(timezone.utc)
+            logger.info("[lipsync_tts] 已提交 MediaKit: job_id=%s task_id=%s", job_id, mk_result["task_id"])
+        except MediaKitError as exc:
+            job.status = "failed"
+            job.error_message = str(exc)
+            job.error_code = exc.code
+            logger.error("[lipsync_tts] 提交 MediaKit 失败: job_id=%s err=%s", job_id, exc)
+
+        db.commit()
+
+    except Exception:
+        logger.exception("[lipsync_tts] 未预期的异常: job_id=%s", job_id)
+        try:
+            job = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id).first()
+            if job and job.status not in ("cancelled", "failed", "completed"):
+                job.status = "failed"
+                job.error_message = "TTS 异步任务执行异常"
+                job.error_code = "AsyncTaskError"
+                job.updated_at = datetime.now(timezone.utc)
+                db.commit()
+        except Exception:
+            logger.exception("[lipsync_tts] 回写失败状态时异常: job_id=%s", job_id)
+    finally:
+        db.close()

apps/web/src/pages/ai-avatar/AiAvatar.css

@@ -484,7 +484,6 @@
   margin-bottom: 12px;
 }
 
-
 .aa-lipsync-preview {
   width: 100%;
   max-width: 240px;
@@ -550,21 +549,23 @@
 /* ── 封面 & 生成 ── */
 .aa-cover-preview {
   width: 100%;
+  max-width: 240px;
   aspect-ratio: 9/16;
-  max-height: 160px;
   background: #f0f0f5;
-  border-radius: 10px;
+  border-radius: 8px;
   overflow: hidden;
   display: flex;
   align-items: center;
   justify-content: center;
-  margin-bottom: 12px;
+  margin: 0 auto 12px auto;
+  position: relative;
 }
 
 .aa-cover-preview img {
   width: 100%;
   height: 100%;
   object-fit: cover;
+  display: block;
 }
 
 .aa-cover-preview__placeholder {
@@ -1249,4 +1250,3 @@
   width: auto;
   min-width: 300px;
 }
-

apps/web/src/pages/ai-avatar/AiAvatarPage.tsx

@@ -5,6 +5,7 @@
  */
 import React, { useState, useCallback, useEffect, useRef } from "react"
 import { message } from "antd"
+import { useNavigate } from "react-router-dom"
 import "./AiAvatar.css"
 import { useAiAvatar } from "./hooks/useAiAvatar"
 import { PanelVideoSelector } from "./components/PanelVideoSelector"
@@ -21,6 +22,7 @@ import {
   createLipsyncJob,
   getLipsyncJob,
   submitRender,
+  getRenderJob,
   generateSmartCover,
 } from "./api/aiAvatar"
 import {
@@ -34,6 +36,7 @@ type PanelKey = "video" | "voice" | "script" | "lipsync" | "title" | "cover"
 
 const AiAvatarPage: React.FC = () => {
   const state = useAiAvatar()
+  const navigate = useNavigate()
   const [currentStep, setCurrentStep] = useState<1 | 2>(1)
   const [collapsed, setCollapsed] = useState<Record<PanelKey, boolean>>({
     video: false,
@@ -52,9 +55,18 @@ const AiAvatarPage: React.FC = () => {
   const [lipsyncErrorMessage, setLipsyncErrorMessage] = useState("")
   /* ── 智能封面加载态 ── */
   const [smartCoverLoading, setSmartCoverLoading] = useState(false)
+  /* ── 渲染进度弹窗 ── */
+  const [showRenderModal, setShowRenderModal] = useState(false)
+  const [renderStatus, setRenderStatus] = useState<"generating" | "completed" | "failed">(
+    "generating",
+  )
+  const [renderProgress, setRenderProgress] = useState(0)
+  const [renderErrorMessage, setRenderErrorMessage] = useState("")
 
   /* ── 对口型轮询 ── */
   const lipsyncTimerRef = useRef<ReturnType<typeof setInterval> | null>(null)
+  /* ── 渲染进度轮询 ── */
+  const renderTimerRef = useRef<ReturnType<typeof setInterval> | null>(null)
 
   const togglePanel = useCallback((key: PanelKey) => {
     setCollapsed((prev) => ({ ...prev, [key]: !prev[key] }))
@@ -182,28 +194,60 @@ const AiAvatarPage: React.FC = () => {
   useEffect(() => {
     return () => {
       if (lipsyncTimerRef.current) clearInterval(lipsyncTimerRef.current)
+      if (renderTimerRef.current) clearInterval(renderTimerRef.current)
     }
   }, [])
 
-  /* ── 生成视频 ── */
+  /* ── 生成视频（含实时进度轮询） ── */
   const handleGenerate = useCallback(async () => {
-    // ② 前置条件提示（#1809）
     if (!state.lipsyncJob || state.lipsyncJob.status !== "completed") {
       message.warning("请先生成对口型视频，待对口型完成后再提交渲染")
       return
     }
     state.setIsGenerating(true)
     try {
-      await submitRender({
+      const job = await submitRender({
         lipsync_job_id: state.lipsyncJob.id,
         script_id: state.script?.id,
-        b_roll_segments: state.bRollSegments as never,
-        // 字段映射：build_title_drawtext_filter 真实口径 text/font_size/font_color/position/...
+        b_roll_segments: state.bRollSegments.map((seg) => ({
+          script_segment_index: seg.script_segment_index,
+          asset_url: seg.asset.file_url || "",
+          mode: seg.mode,
+          start_time: seg.start_time,
+          end_time: seg.end_time,
+          pip_position: seg.pip_position,
+          pip_scale: seg.pip_scale,
+        })) as never,
         title_config: buildTitleConfigPayload(state.titleConfig),
-        // cover_config：智能封面 cover_url + 截帧 timestamp
         cover_config: buildCoverConfigPayload(state.coverConfig, state.coverConfig.smart_cover_url),
       })
-      message.success("渲染任务已提交，可在视频管理中查看进度")
+
+      // 打开渲染进度弹窗，启动轮询
+      setShowRenderModal(true)
+      setRenderStatus("generating")
+      setRenderProgress(job.progress ?? 0)
+      setRenderErrorMessage("")
+
+      if (renderTimerRef.current) clearInterval(renderTimerRef.current)
+      renderTimerRef.current = setInterval(async () => {
+        try {
+          const updated = await getRenderJob(job.id)
+          setRenderProgress(updated.progress ?? 0)
+          if (updated.status === "completed") {
+            if (renderTimerRef.current) clearInterval(renderTimerRef.current)
+            renderTimerRef.current = null
+            setRenderStatus("completed")
+            message.success("视频已生成并保存到成片库")
+          } else if (updated.status === "failed") {
+            if (renderTimerRef.current) clearInterval(renderTimerRef.current)
+            renderTimerRef.current = null
+            setRenderStatus("failed")
+            setRenderErrorMessage(updated.error_message || "渲染失败，请重试")
+          }
+        } catch (pollErr) {
+          console.error("[渲染] 轮询失败:", pollErr)
+        }
+      }, 3000)
     } catch (err) {
       console.error("渲染任务提交失败:", err)
       message.error(err instanceof Error ? err.message : "渲染任务提交失败，请重试")
@@ -211,14 +255,19 @@ const AiAvatarPage: React.FC = () => {
       state.setIsGenerating(false)
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [
-    state.lipsyncJob,
-    state.script,
-    state.bRollSegments,
-    state.titleConfig,
-    state.coverConfig,
+  }, [state.lipsyncJob, state.script, state.bRollSegments, state.titleConfig, state.coverConfig])
 
-  ])
+  /* ── 关闭渲染进度弹窗 ── */
+  const handleCancelRender = useCallback(() => {
+    if (renderTimerRef.current) {
+      clearInterval(renderTimerRef.current)
+      renderTimerRef.current = null
+    }
+    setShowRenderModal(false)
+    setRenderStatus("generating")
+    setRenderProgress(0)
+    setRenderErrorMessage("")
+  }, [])
 
   /* ── 智能封面：调后端 MediaKit 选帧接口（#1822） ── */
   const handleSmartCover = useCallback(async () => {
@@ -359,6 +408,7 @@ const AiAvatarPage: React.FC = () => {
                   onOpenBRollModal={() => state.setShowBRollModal(true)}
                   onRemoveBRoll={state.removeBRollSegment}
                   titleConfig={state.titleConfig}
+                  onTitlePositionChange={(pos) => state.updateTitleConfig(pos)}
                 />
                 <div className="aa-step-btn-row">
                   <button type="button" className="aa-btn" onClick={handlePrevStep}>
@@ -375,7 +425,10 @@ const AiAvatarPage: React.FC = () => {
                 <span className="aa-panel__toggle">▼</span>
               </div>
               <div className="aa-panel__body">
-                <PanelTitleConfig titleConfig={state.titleConfig} onUpdate={state.updateTitleConfig} />
+                <PanelTitleConfig
+                  titleConfig={state.titleConfig}
+                  onUpdate={state.updateTitleConfig}
+                />
               </div>
             </div>
 
@@ -505,6 +558,112 @@ const AiAvatarPage: React.FC = () => {
           </div>
         </div>
       )}
+
+      {/* 渲染进度弹窗 */}
+      {showRenderModal && (
+        <div className="aa-modal-overlay">
+          <div className="aa-modal" onClick={(e) => e.stopPropagation()}>
+            <div className="aa-modal__header">
+              <span className="aa-modal__title">视频渲染</span>
+              <button className="aa-modal__close" onClick={handleCancelRender}>
+                ✕
+              </button>
+            </div>
+            <div
+              className="aa-modal__body"
+              style={{
+                display: "flex",
+                flexDirection: "column",
+                alignItems: "center",
+                padding: "40px 20px",
+              }}
+            >
+              {renderStatus === "generating" && (
+                <>
+                  <div className="aa-lipsync-spinner" />
+                  <div style={{ marginTop: 20, fontSize: 15, color: "#1a1a2e" }}>
+                    正在生成视频，请稍后
+                  </div>
+                  <div
+                    style={{
+                      marginTop: 16,
+                      fontSize: 32,
+                      fontWeight: 700,
+                      color: "#1890ff",
+                    }}
+                  >
+                    {renderProgress}%
+                  </div>
+                  <div
+                    style={{
+                      marginTop: 12,
+                      width: "80%",
+                      height: 8,
+                      backgroundColor: "#f0f0f0",
+                      borderRadius: 4,
+                      overflow: "hidden",
+                    }}
+                  >
+                    <div
+                      style={{
+                        width: `${renderProgress}%`,
+                        height: "100%",
+                        backgroundColor: "#1890ff",
+                        borderRadius: 4,
+                        transition: "width 0.5s ease",
+                      }}
+                    />
+                  </div>
+                  <div style={{ marginTop: 12, fontSize: 13, color: "#8c8ca1" }}>
+                    请勿关闭页面，完成后将自动提示
+                  </div>
+                </>
+              )}
+              {renderStatus === "completed" && (
+                <>
+                  <div style={{ fontSize: 48 }}>✅</div>
+                  <div style={{ marginTop: 16, fontSize: 15, color: "#1a1a2e" }}>
+                    视频已保存到成片库
+                  </div>
+                  <button
+                    className="aa-btn"
+                    style={{ marginTop: 16 }}
+                    onClick={() => {
+                      setShowRenderModal(false)
+                      navigate("/app/products")
+                    }}
+                  >
+                    📁 查看成片
+                  </button>
+                </>
+              )}
+              {renderStatus === "failed" && (
+                <>
+                  <div style={{ fontSize: 48 }}>❌</div>
+                  <div style={{ marginTop: 16, fontSize: 15, color: "#1a1a2e" }}>视频渲染失败</div>
+                  {renderErrorMessage && (
+                    <div style={{ marginTop: 8, fontSize: 13, color: "#ff4d4f" }}>
+                      {renderErrorMessage}
+                    </div>
+                  )}
+                </>
+              )}
+            </div>
+            <div className="aa-modal__footer">
+              {renderStatus === "generating" && (
+                <button className="aa-btn aa-btn--danger" onClick={handleCancelRender}>
+                  关闭窗口
+                </button>
+              )}
+              {renderStatus !== "generating" && (
+                <button className="aa-btn" onClick={handleCancelRender}>
+                  关闭
+                </button>
+              )}
+            </div>
+          </div>
+        </div>
+      )}
     </div>
   )
 }

apps/web/src/pages/ai-avatar/components/PanelLipsyncPreview.tsx

@@ -3,6 +3,7 @@
  * B-roll 画面插入 + 对口型视频预览 + 生成/重新生成按钮
  * v3.1: 预览容器按 1/2 缩放、标题实时叠加预览
  */
+import React, { useRef } from "react"
 import type { LipsyncJob, BRollSegment, AiAvatarTitleConfig } from "../types"
 
 interface PanelLipsyncPreviewProps {
@@ -13,6 +14,8 @@ interface PanelLipsyncPreviewProps {
   onRemoveBRoll: (id: string) => void
   /** 标题配置（实时叠加预览用） */
   titleConfig?: AiAvatarTitleConfig
+  /** 标题位置变更回调（拖拽结束时调用） */
+  onTitlePositionChange?: (pos: { pos_x: number; pos_y: number }) => void
 }
 
 const BROLL_MODE_LABEL: Record<BRollSegment["mode"], string> = {
@@ -33,9 +36,12 @@ export function PanelLipsyncPreview({
   onOpenBRollModal,
   onRemoveBRoll,
   titleConfig,
+  onTitlePositionChange,
 }: PanelLipsyncPreviewProps) {
-  const isGenerating =
-    lipsyncJob?.status === "pending" || lipsyncJob?.status === "processing"
+  const titleDragRef = useRef<HTMLDivElement>(null)
+  const draggingTitleRef = useRef(false)
+  const previewContainerRef = useRef<HTMLDivElement>(null)
+  const isGenerating = lipsyncJob?.status === "pending" || lipsyncJob?.status === "processing"
   const isDone = lipsyncJob?.status === "completed"
   const isFailed = lipsyncJob?.status === "failed"
 
@@ -57,7 +63,6 @@ export function PanelLipsyncPreview({
         fontSize: `${(titleConfig.size || 36) * 0.55}px`, // 预览等比缩
         fontWeight: titleConfig.bold ? 700 : 400,
         fontStyle: titleConfig.italic ? "italic" : "normal",
-        whiteSpace: "pre-wrap",
         textAlign: "center",
         width: "90%",
         padding: "4px 8px",
@@ -71,6 +76,40 @@ export function PanelLipsyncPreview({
       }
     : null
 
+  const handleTitlePointerDown = (e: React.PointerEvent<HTMLDivElement>) => {
+    if (!onTitlePositionChange || !previewContainerRef.current) return
+    e.preventDefault()
+    e.stopPropagation()
+    ;(e.target as Element).setPointerCapture(e.pointerId)
+    draggingTitleRef.current = true
+    ;(e.currentTarget as HTMLDivElement).style.cursor = "grabbing"
+  }
+  const handleTitlePointerMove = (e: React.PointerEvent<HTMLDivElement>) => {
+    if (!draggingTitleRef.current || !previewContainerRef.current) return
+    e.preventDefault()
+    e.stopPropagation()
+    if (titleDragRef.current) {
+      const rect = previewContainerRef.current.getBoundingClientRect()
+      const relX = Math.max(0, Math.min(rect.width, e.clientX - rect.left))
+      const relY = Math.max(0, Math.min(rect.height, e.clientY - rect.top))
+      const xpct = (relX / rect.width) * 100
+      const ypct = (relY / rect.height) * 100
+      titleDragRef.current.style.left = `${xpct}%`
+      titleDragRef.current.style.top = `${ypct}%`
+    }
+  }
+  const handleTitlePointerUp = (e: React.PointerEvent<HTMLDivElement>) => {
+    if (!draggingTitleRef.current) return
+    draggingTitleRef.current = false
+    if (onTitlePositionChange && previewContainerRef.current) {
+      const rect = previewContainerRef.current.getBoundingClientRect()
+      const relX = Math.max(0, Math.min(rect.width, e.clientX - rect.left))
+      const relY = Math.max(0, Math.min(rect.height, e.clientY - rect.top))
+      onTitlePositionChange({ pos_x: relX, pos_y: relY })
+    }
+    ;(e.currentTarget as HTMLDivElement).style.cursor = "grab"
+  }
+
   return (
     <div className="aa-script-lipsync">
       {/* ── B-roll 画面 ── */}
@@ -78,9 +117,7 @@ export function PanelLipsyncPreview({
         <div className="aa-lipsync-section__title">
           <span style={{ marginRight: 8 }}>🎞️ 插入画面</span>
           {bRollSegments.length > 0 && (
-            <span className="aa-broll-badge">
-              🎬 {bRollSegments.length} 个画面
-            </span>
+            <span className="aa-broll-badge">🎬 {bRollSegments.length} 个画面</span>
           )}
         </div>
         <div className="aa-lipsync-actions">
@@ -104,10 +141,7 @@ export function PanelLipsyncPreview({
                     alt={seg.asset.name}
                   />
                 ) : (
-                  <span
-                    className="aa-broll-item__thumb"
-                    style={{ padding: "6px 4px" }}
-                  >
+                  <span className="aa-broll-item__thumb" style={{ padding: "6px 4px" }}>
                     🎬
                   </span>
                 )}
@@ -144,11 +178,31 @@ export function PanelLipsyncPreview({
       <div className="aa-lipsync-section">
         <div className="aa-lipsync-section__title">对口型预览</div>
 
-        <div className="aa-lipsync-preview">
+        <div className="aa-lipsync-preview" ref={previewContainerRef}>
           {isDone && lipsyncJob?.output_video_url ? (
             <div style={{ position: "relative", width: "100%", height: "100%" }}>
               <video src={lipsyncJob.output_video_url} controls />
-              {titleOverlayStyle && <div style={titleOverlayStyle}>{titleConfig!.title}</div>}
+              {titleOverlayStyle && (
+                <div
+                  ref={titleDragRef}
+                  style={{
+                    ...titleOverlayStyle,
+                    cursor: onTitlePositionChange ? "grab" : "default",
+                    pointerEvents: onTitlePositionChange ? "auto" : "none",
+                  }}
+                  onPointerDown={handleTitlePointerDown}
+                  onPointerMove={handleTitlePointerMove}
+                  onPointerUp={handleTitlePointerUp}
+                  onPointerCancel={handleTitlePointerUp}
+                >
+                  {titleConfig!.title.split(/[/／]/).map((part, i) => (
+                    <span key={i}>
+                      {i > 0 && <br />}
+                      {part}
+                    </span>
+                  ))}
+                </div>
+              )}
             </div>
           ) : isGenerating ? (
             <div style={{ width: "80%", textAlign: "center", color: "#fff" }}>
@@ -189,11 +243,7 @@ export function PanelLipsyncPreview({
 
         <div className="aa-lipsync-actions">
           {isDone ? (
-            <button
-              type="button"
-              className="aa-btn aa-btn--full"
-              onClick={onGenerateLipsync}
-            >
+            <button type="button" className="aa-btn aa-btn--full" onClick={onGenerateLipsync}>
               🔄 重新生成对口型
             </button>
           ) : isGenerating ? (

apps/web/src/pages/ai-avatar/components/PanelScript.tsx

@@ -55,9 +55,7 @@ export function PanelScript({
         value={scriptText}
         readOnly={scriptTab === "library"}
         placeholder={
-          scriptTab === "library"
-            ? "点击上方按钮，从文案库选择文案…"
-            : "请输入数字人口播文案…"
+          scriptTab === "library" ? "点击上方按钮，从文案库选择文案…" : "请输入数字人口播文案…"
         }
         onChange={(e) => onScriptTextChange(e.target.value)}
       />

apps/worker/worker_app/celery_app.py

@@ -37,6 +37,7 @@ celery_app.conf.imports = (
     "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",
     "worker_app.tasks.cleanup",
+    "apps.api.app.tasks.lipsync_tts",
 )
 
 # Celery Beat 定时任务调度

infra/docker/worker.Dockerfile

@@ -20,7 +20,7 @@ WORKDIR /app
 
 # 设置 Python 环境变量
 ENV PATH="/opt/venv/bin:$PATH"
-ENV PYTHONPATH=/app:/app/packages
+ENV PYTHONPATH=/app:/app/apps/api:/app/packages
 ENV PYTHONUNBUFFERED=1
 ENV APP_VERSION=$APP_VERSION
 
@@ -28,8 +28,10 @@ ENV APP_VERSION=$APP_VERSION
 COPY alembic.ini /app/alembic.ini
 COPY migrations/ /app/migrations/
 COPY packages/ /app/packages/
-COPY apps/api/app/config.py /app/apps/api/app/config.py
-COPY apps/api/app/core/ /app/apps/api/app/core/
+# PR #1844 起，worker 还需要加载 apps.api.app.tasks.lipsync_tts，
+# 该 task 依赖 app.services.* 与 app.core.celery_app（PYTHONPATH=/app/apps/api 下解析）。
+# 为避免后续新增 task 再次漏 COPY，直接把整个 apps/api/app/ 复制进 worker 镜像。
+COPY apps/api/app/ /app/apps/api/app/
 
 # Worker 启动脚本
 COPY infra/docker/entrypoint-worker.sh /usr/local/bin/entrypoint-worker.sh

scripts/ci/auto_fix_formatting.py

@@ -1,10 +1,13 @@
 #!/usr/bin/env python3
-"""CI中自动修复代码格式（Python: black + isort | Frontend: prettier），并推送回原分支。
+"""CI中自动修复代码格式（Python: black + isort + ruff | Frontend: prettier），并推送回原分支。
 
 - PR事件：所有PR只要Code Quality因格式问题失败，自动修复并push回源分支
 - Push事件（develop/main）：自动修复并push回原分支，保持主干格式永远正确
 - 防循环：修复commit带 [skip ci-format-check] 标记，检测到该标记则跳过修复
-- 只修格式（black/isort/prettier），ruff逻辑类错误不动
+- black/isort/prettier 修格式；ruff check --fix --unsafe-fixes 自动修复
+  ruff 可修复的 lint 规则（含 F401 未使用 import 等 unsafe fix）
+- ruff 目标范围与 validate_style.sh 的检查范围对齐：apps packages tests
+  （alembic/scripts 不在 ruff 检查范围内，不做修复）
 当code quality检查因格式问题失败时触发。
 """
 
@@ -135,7 +138,7 @@ def get_pr_head_branch(pr_number, api_url, token):
 
 
 def fix_python(target_py_files, scan_mode):
-    """修复 Python 文件格式 (black + isort)"""
+    """修复 Python 文件 (black 格式化 + isort 排序 + ruff lint 自动修复)"""
     if not target_py_files:
         print("没有需要修复的 Python 文件，跳过")
         return
@@ -146,14 +149,48 @@ def fix_python(target_py_files, scan_mode):
     result = run(f"python3 -m black {target_str}", check=False)
     print(result.stdout[-500:] if result.stdout else "")
     if result.returncode != 0:
-        print("black执行失败，但继续尝试isort", file=sys.stderr)
+        print("black执行失败，但继续尝试isort/ruff", file=sys.stderr)
 
     print()
     print("--- isort 排序 ---")
     result = run(f"python3 -m isort {target_str}", check=False)
     print(result.stdout[-500:] if result.stdout else "")
     if result.returncode != 0:
-        print("isort执行失败", file=sys.stderr)
+        print("isort执行失败，继续尝试ruff", file=sys.stderr)
+
+    # ruff lint 自动修复
+    # 与 validate_style.sh 的检查范围对齐：只修 apps/packages/tests
+    # （alembic 在 pyproject.toml 中被 exclude，scripts 不在 ruff 检查范围内）
+    ruff_scopes = ("apps/", "packages/", "tests/")
+    ruff_files = [f for f in target_py_files if f.startswith(ruff_scopes)]
+    if scan_mode != "incremental":
+        ruff_targets = "apps packages tests"
+    elif ruff_files:
+        ruff_targets = " ".join(ruff_files)
+    else:
+        ruff_targets = ""
+
+    if ruff_targets:
+        # ruff 由 style job 的 requirements-dev.txt 安装；不可用时跳过（不阻断 black/isort 的修复）
+        avail = run("python3 -m ruff --version", check=False)
+        if avail.returncode != 0:
+            print("ruff 不可用，跳过 ruff 自动修复", file=sys.stderr)
+        else:
+            print()
+            print("--- ruff lint 自动修复 (--fix --unsafe-fixes) ---")
+            # --unsafe-fixes 用于启用 F401（未使用 import）等 ruff 归类为 unsafe 的自动修复；
+            # 安全性由修复后重跑的完整 CI（单测/构建/staging 健康检查）兜底
+            result = run(
+                f"python3 -m ruff check {ruff_targets} --fix --unsafe-fixes",
+                check=False,
+            )
+            print(result.stdout[-1500:] if result.stdout else "")
+            if result.returncode != 0:
+                # 可能是仍有不可自动修复的 lint 错误（留待 style check 再次拦截），或修复过程出错
+                print("ruff 自动修复后仍有未修复项或执行失败，剩余问题由 style check 继续拦截", file=sys.stderr)
+    else:
+        print()
+        print("增量模式且无 apps/packages/tests 范围内的 Python 变更，跳过 ruff 自动修复")
 
 
 def fix_frontend(target_fe_files, scan_mode, repo_root):
@@ -334,7 +371,7 @@ def main():
     # 提交修复
     run("git clean -fd")
     run("git add -u")
-    run('git commit -m "style: auto-format with black + isort + prettier [skip ci-format-check]"')
+    run('git commit -m "style: auto-format with black + isort + ruff + prettier [skip ci-format-check]"')
 
     # 推送（head_branch已从ensure_git_repo获取）
     print(f"\nPR来源分支: {head_branch}")

tests/unit/test_ai_avatar_emotion_tts_lipsync.py

@@ -132,14 +132,8 @@ def _lipsync_service_with_mocks():
 def test_create_job_tts_direct_mode_synthesizes_audio():
     svc, client, cosy = _lipsync_service_with_mocks()
 
-    with (
-        patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch,
-        patch("app.services.lipsync_service.safe_download_bytes") as dl_patch,
-    ):
-        storage = MagicMock()
-        storage.upload_file.return_value = "https://oss/tts.mp3"
-        storage_patch.return_value = storage
-        dl_patch.return_value = b"FAKEAUDIO"
+    with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+        mock_task.apply_async.return_value = MagicMock(id="celery-task-123")
 
         job = svc.create_job(
             user_id="user-1",
@@ -150,19 +144,16 @@ def test_create_job_tts_direct_mode_synthesizes_audio():
             emotion="兴奋",
         )
 
-    # 调了 TTS 合成，带 speed/emotion
-    cosy.submit_synthesize_task.assert_called_once()
-    _, kwargs = cosy.submit_synthesize_task.call_args
-    assert kwargs["speed"] == 1.2
-    assert kwargs["emotion"] == "excited"
-    assert kwargs["voice_id"] == "cosy-v1"
-    # MediaKit 用合成后的 OSS 音频 URL 提交
-    _, submit_kwargs = client.submit_lipsync.call_args
-    assert submit_kwargs["audio_url"] == "https://oss/tts.mp3"
-    assert submit_kwargs["video_url"] == "https://oss/person.mp4"
-    # DB 记录了 TTS 字段
+    # v4: TTS 模式下 create_job 返回 tts_processing 状态，dispatch Celery 任务
+    assert job.status == "tts_processing"
     assert job.emotion == "excited"
     assert job.speed == 1.2
+    # 不直接调用 CosyVoice（由 Celery 任务处理）
+    cosy.submit_synthesize_task.assert_not_called()
+    # 不直接提交 MediaKit（由 Celery 任务处理）
+    client.submit_lipsync.assert_not_called()
+    # dispatch 了 Celery 任务
+    mock_task.apply_async.assert_called_once()
 
 
 def test_create_job_direct_audio_mode_skips_tts():
@@ -180,23 +171,25 @@ def test_create_job_direct_audio_mode_skips_tts():
 
 
 def test_create_job_tts_failure_raises():
-    from app.services.mediakit_client import MediaKitError
-
-    from packages.application.cosyvoice_service import CosyVoiceError
-
+    """v4: TTS 模式下 create_job 不再同步失败，而是 dispatch Celery 任务。
+    TTS 合成失败由 Celery 任务内部处理并更新 job 状态。"""
     svc, client, cosy = _lipsync_service_with_mocks()
-    cosy.submit_synthesize_task.side_effect = CosyVoiceError("Arrearage")
 
-    with pytest.raises(MediaKitError) as exc:
-        svc.create_job(
+    with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+        mock_task.apply_async.return_value = MagicMock(id="celery-task-456")
+
+        job = svc.create_job(
             user_id="user-1",
             video_url="https://oss/person.mp4",
             voice_id="v-1",
             script_text="文本",
         )
-    assert exc.value.code == "TTSSynthesisFailed"
-    # TTS 失败不应提交 MediaKit
+
+    # create_job 成功返回 tts_processing，不直接调用 TTS
+    assert job.status == "tts_processing"
+    cosy.submit_synthesize_task.assert_not_called()
     client.submit_lipsync.assert_not_called()
+    mock_task.apply_async.assert_called_once()
 
 
 # ── refresh 同步中间状态 ────────────────────────────────────────────────
@@ -233,7 +226,7 @@ def test_smart_cover_selects_best_frame_and_persists():
     with (
         patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch,
         patch("packages.shared.cover_frame_scorer.score_frames") as score_patch,
-        patch("httpx.get") as http_get,
+        patch("httpx.Client") as http_client_cls,
         patch("packages.shared.storage.get_shared_storage_service") as storage_patch,
     ):
         mk = MagicMock()
@@ -245,32 +238,107 @@ def test_smart_cover_selects_best_frame_and_persists():
             {"url": "https://mk/f1.jpg", "score": 90.0, "image_path": cands[1]["image_path"]},
             {"url": "https://mk/f0.jpg", "score": 60.0, "image_path": cands[0]["image_path"]},
         ]
+        # httpx.Client 连接池 mock
+        client_instance = MagicMock()
         resp = MagicMock()
         resp.content = b"IMGDATA"
         resp.raise_for_status = MagicMock()
-        http_get.return_value = resp
+        client_instance.get.return_value = resp
+        client_instance.__enter__ = MagicMock(return_value=client_instance)
+        client_instance.__exit__ = MagicMock(return_value=False)
+        http_client_cls.return_value = client_instance
+
         storage = MagicMock()
-        storage.upload_file.return_value = "https://oss/cover.jpg"
+        storage.public_url = "https://oss.example.com"
+        # video_url 不是自家 OSS，不重签
+        storage.get_download_url.side_effect = lambda url, **kw: f"{url}?signed=1"
+        storage.upload_file.return_value = "https://oss.example.com/cover.jpg"
         storage_patch.return_value = storage
 
-        url = cov.generate_smart_cover("https://oss/avatar.mp4", job_id="job-1")
+        url = cov.generate_smart_cover("https://other-host/avatar.mp4", job_id="job-1")
 
-    assert url == "https://oss/cover.jpg"
+    assert "signed=1" in url or url == "https://oss.example.com/cover.jpg"
     mk.extract_frames.assert_called_once()
     score_patch.assert_called_once()
+    # 验证使用了增大的轮询参数
+    call_kwargs = mk.extract_frames.call_args
+    assert call_kwargs.kwargs.get("poll_interval") == 3.0 or call_kwargs[1].get("poll_interval") == 3.0
+    assert call_kwargs.kwargs.get("max_poll_attempts") == 20 or call_kwargs[1].get("max_poll_attempts") == 20
 
 
 def test_smart_cover_returns_empty_when_mediakit_unavailable():
     from app.services import ai_avatar_cover_service as cov
 
-    with patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch:
+    with (
+        patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch,
+        patch("packages.shared.storage.get_shared_storage_service") as storage_patch,
+    ):
         mk = MagicMock()
         mk.is_available = False
         mk_patch.return_value = mk
-        url = cov.generate_smart_cover("https://oss/avatar.mp4")
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage_patch.return_value = storage
+        url = cov.generate_smart_cover("https://oss.example.com/avatar.mp4")
     assert url == ""
 
 
+def test_sign_video_url_resigns_own_oss_url():
+    """自家 OSS 私有桶 URL 应被重签为长有效期预签名 URL"""
+    from app.services.ai_avatar_cover_service import _sign_video_url_for_mediakit
+
+    with patch("packages.shared.storage.get_shared_storage_service") as storage_patch:
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.get_download_url.return_value = "https://oss.example.com/file.mp4?Expires=xxx&Signature=yyy"
+        storage_patch.return_value = storage
+
+        result = _sign_video_url_for_mediakit("https://oss.example.com/file.mp4")
+
+    assert "Signature=yyy" in result
+    storage.get_download_url.assert_called_once()
+
+
+def test_sign_video_url_skips_external_url():
+    """外部 URL（非自家 OSS）应原样返回，不做重签"""
+    from app.services.ai_avatar_cover_service import _sign_video_url_for_mediakit
+
+    with patch("packages.shared.storage.get_shared_storage_service") as storage_patch:
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage_patch.return_value = storage
+
+        result = _sign_video_url_for_mediakit("https://external-cdn.com/video.mp4")
+
+    assert result == "https://external-cdn.com/video.mp4"
+    storage.get_download_url.assert_not_called()
+
+
+def test_extract_frames_uses_extended_poll_params():
+    """验证 select_best_cover_frame 使用增大后的轮询参数"""
+    from app.services import ai_avatar_cover_service as cov
+
+    with (
+        patch("packages.shared.mediakit_client.get_mediakit_client") as mk_patch,
+        patch("packages.shared.storage.get_shared_storage_service") as storage_patch,
+    ):
+        mk = MagicMock()
+        mk.is_available = True
+        mk.extract_frames.return_value = [{"image_url": "https://mk/f0.jpg"}]
+        mk_patch.return_value = mk
+
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage_patch.return_value = storage
+
+        cov.select_best_cover_frame("https://other/avatar.mp4", max_frames=3)
+
+    call_kwargs = mk.extract_frames.call_args
+    assert call_kwargs.kwargs.get("poll_interval") == 3.0 or call_kwargs[1].get("poll_interval") == 3.0
+    assert call_kwargs.kwargs.get("max_poll_attempts") == 20 or call_kwargs[1].get("max_poll_attempts") == 20
+    assert call_kwargs.kwargs.get("max_retries") == 1 or call_kwargs[1].get("max_retries") == 1
+
+
 # ── 渲染 script_id 可选（手动文案直生场景）──────────────────────────────
 
 

tests/unit/test_ai_avatar_render_service.py

@@ -519,6 +519,103 @@ class TestAiAvatarRenderService:
         # 不应执行渲染逻辑
         mock_db.commit.assert_not_called()
 
+    def test_execute_render_success_creates_clip_record(self):
+        """execute_render 完成后自动创建成片记录到成片库."""
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        mock_db = _make_mock_db()
+        mock_lipsync_job = _make_mock_lipsync_job(
+            status="completed",
+            output_video_url="https://oss/lipsync.mp4",
+            output_duration=30.0,
+        )
+        mock_job = _make_mock_render_job(
+            job_id="render-ok",
+            status="pending",
+            output_video_url="",
+            output_cover_url="",
+            output_duration=0.0,
+        )
+        mock_filter = MagicMock()
+        mock_filter.first.side_effect = [mock_job, mock_lipsync_job]
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+
+        svc = AiAvatarRenderService(mock_db)
+
+        with (
+            patch.object(svc, "_download_video", return_value="/tmp/video.mp4"),
+            patch.object(svc, "_upload_to_oss", side_effect=lambda path, key: f"https://oss/{key}"),
+            patch("os.system", return_value=0),
+            patch("tempfile.TemporaryDirectory") as tmpdir_mock,
+            patch(
+                "app.services.ai_avatar_cover_service.generate_smart_cover", return_value="https://oss/smart_cover.jpg"
+            ),
+            patch("packages.domain.generated_video.GeneratedVideo.create") as gv_create,
+            patch(
+                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository"
+            ) as repo_cls,
+        ):
+            import tempfile as _tf
+
+            tmpdir_mock.return_value.__enter__ = MagicMock(return_value="/tmp/testdir")
+            tmpdir_mock.return_value.__exit__ = MagicMock(return_value=False)
+
+            mock_clip = MagicMock()
+            mock_clip.id = "clip-001"
+            gv_create.return_value = mock_clip
+            mock_repo = MagicMock()
+            repo_cls.return_value = mock_repo
+
+            svc.execute_render("render-ok")
+
+        assert mock_job.status == "completed"
+        gv_create.assert_called_once()
+        call_kwargs = gv_create.call_args
+        assert "https://oss/" in call_kwargs.kwargs["file_url"]
+        assert call_kwargs.kwargs["user_id"] == "user-1"
+        mock_repo.create.assert_called_once_with(mock_clip)
+
+    def test_execute_render_clip_failure_does_not_affect_render(self):
+        """成片创建失败不影响渲染任务标记为成功."""
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        mock_db = _make_mock_db()
+        mock_lipsync_job = _make_mock_lipsync_job(
+            status="completed",
+            output_video_url="https://oss/lipsync.mp4",
+            output_duration=30.0,
+        )
+        mock_job = _make_mock_render_job(
+            job_id="render-clip-fail",
+            status="pending",
+            output_video_url="",
+            output_cover_url="",
+            output_duration=0.0,
+        )
+        mock_filter = MagicMock()
+        mock_filter.first.side_effect = [mock_job, mock_lipsync_job]
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+
+        svc = AiAvatarRenderService(mock_db)
+
+        with (
+            patch.object(svc, "_download_video", return_value="/tmp/video.mp4"),
+            patch.object(svc, "_upload_to_oss", side_effect=lambda path, key: f"https://oss/{key}"),
+            patch("os.system", return_value=0),
+            patch("tempfile.TemporaryDirectory") as tmpdir_mock,
+            patch("app.services.ai_avatar_cover_service.generate_smart_cover", side_effect=RuntimeError("DB error")),
+        ):
+            tmpdir_mock.return_value.__enter__ = MagicMock(return_value="/tmp/testdir")
+            tmpdir_mock.return_value.__exit__ = MagicMock(return_value=False)
+            svc.execute_render("render-clip-fail")
+
+        # 即使成片创建失败，渲染任务仍应标记为 completed
+        assert mock_job.status == "completed"
+
     def test_error_exception_has_code(self):
         from app.services.ai_avatar_render_service import AiAvatarRenderError
 

tests/unit/test_lipsync_routes.py

@@ -208,24 +208,21 @@ class TestLipsyncServiceUnit:
     """Service 层单元测试（纯 mock，不依赖数据库）— #1809 更新."""
 
     def test_create_job_success(self, mock_mediakit, mock_cosyvoice):
-        """v3: TTS 直生——service 内部 submit_synthesize_task 合成后转存 OSS，再提交 MediaKit."""
+        """TTS 直生——v4 异步模式：create_job 只创建 DB 记录 + dispatch Celery 任务."""
         from app.services.lipsync_service import LipsyncService
 
         mock_db = MagicMock()
         mock_repo = MagicMock()
         mock_repo.get.return_value = None  # 预置音色，原样返回 voice_id
 
-        with (
-            patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch,
-            patch("app.services.lipsync_service.safe_download_bytes") as dl_patch,
-        ):
-            storage_patch.return_value.upload_file.return_value = "https://my-oss/tts.mp3"
-            dl_patch.return_value = b"audio-bytes"
+        with patch(
+            "app.services.lipsync_service.tts_synthesize_and_submit"
+        ) as mock_task:
+            mock_task.apply_async.return_value = MagicMock(id="celery-task-123")
 
             svc = LipsyncService(
                 mock_db,
                 client=mock_mediakit,
-                cosyvoice_service=mock_cosyvoice,
                 voice_clone_repo=mock_repo,
             )
 
@@ -238,32 +235,23 @@ class TestLipsyncServiceUnit:
                 emotion="兴奋",
             )
 
-        assert job.status == "submitted"
-        assert job.mediakit_task_id == "mk-task-123"
-        # TTS 直生走 submit_synthesize_task，带语速/情绪
-        mock_cosyvoice.submit_synthesize_task.assert_called_once()
-        _, kwargs = mock_cosyvoice.submit_synthesize_task.call_args
-        assert kwargs["text"] == "大家好，欢迎来到直播间"
-        assert kwargs["voice_id"] == "longxiaochun_v3"
-        assert kwargs["speed"] == 1.2
-        assert kwargs["emotion"] == "excited"  # 兴奋→excited
+        # TTS 模式：异步返回，状态为 tts_processing
+        assert job.status == "tts_processing"
+        assert not job.audio_url  # TTS 音频尚未合成（默认空字符串）
+        # 不直接调用 CosyVoice
+        mock_cosyvoice.submit_synthesize_task.assert_not_called()
+        # dispatch 了 Celery 任务
+        mock_task.apply_async.assert_called_once()
         # job 记录透传字段
         assert job.speed == 1.2
         assert job.emotion == "excited"
-        # MediaKit 用转存后的 OSS audio_url
-        call_kwargs = mock_mediakit.submit_lipsync.call_args
-        assert call_kwargs.kwargs["audio_url"] == "https://my-oss/tts.mp3"
+        # MediaKit 尚未提交（由 Celery 任务处理）
+        mock_mediakit.submit_lipsync.assert_not_called()
 
     def test_create_job_tts_failure(self, mock_mediakit):
-        """v3: TTS 合成失败时，CosyVoiceError 被包装为 MediaKitError(TTSSynthesisFailed)，
-        在建 DB 记录之前抛出，不提交 MediaKit。"""
+        """v4: TTS 模式下 create_job 不再同步失败，而是 dispatch Celery 任务。
+        TTS 合成失败由 Celery 任务内部处理（见 test_lipsync_speed_optimization.py）。"""
         from app.services.lipsync_service import LipsyncService
-        from app.services.mediakit_client import MediaKitError
-
-        from packages.application.cosyvoice_service import CosyVoiceError
-
-        mock_cosyvoice = MagicMock()
-        mock_cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("Arrearage 欠费")
 
         mock_db = MagicMock()
         mock_repo = MagicMock()
@@ -272,38 +260,41 @@ class TestLipsyncServiceUnit:
         svc = LipsyncService(
             mock_db,
             client=mock_mediakit,
-            cosyvoice_service=mock_cosyvoice,
             voice_clone_repo=mock_repo,
         )
 
-        with pytest.raises(MediaKitError) as exc_info:
-            svc.create_job(
+        with patch(
+            "app.services.lipsync_service.tts_synthesize_and_submit"
+        ) as mock_task:
+            mock_task.apply_async.return_value = MagicMock(id="celery-task-456")
+
+            job = svc.create_job(
                 user_id="user-1",
                 video_url="https://example.com/video.mp4",
                 voice_id="longxiaochun_v3",
                 script_text="测试文本",
             )
-        assert exc_info.value.code == "TTSSynthesisFailed"
 
-        # 不应提交到 MediaKit
+        # TTS 模式下 create_job 成功返回，状态为 tts_processing
+        assert job.status == "tts_processing"
         mock_mediakit.submit_lipsync.assert_not_called()
+        mock_task.apply_async.assert_called_once()
 
-    def test_create_job_api_failure(self, mock_mediakit, mock_cosyvoice):
-        """MediaKit 提交失败."""
+    def test_create_job_api_failure(self, mock_mediakit):
+        """MediaKit 提交失败（直传音频模式同步触发）."""
         from app.services.lipsync_service import LipsyncService
         from app.services.mediakit_client import MediaKitError
 
         mock_mediakit.submit_lipsync.side_effect = MediaKitError("API 调用失败", code="SubmitFailed")
 
         mock_db = MagicMock()
-        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc = LipsyncService(mock_db, client=mock_mediakit)
 
         with pytest.raises(MediaKitError, match="API 调用失败"):
             svc.create_job(
                 user_id="user-1",
                 video_url="https://example.com/video.mp4",
-                voice_id="longxiaochun_v3",
-                script_text="测试文本",
+                audio_url="https://example.com/audio.mp3",
             )
 
     def test_get_job_delegates_to_db(self, mock_mediakit, mock_cosyvoice):
@@ -434,24 +425,21 @@ class TestLipsyncServiceUnit:
         assert result.status == "completed"
 
     def test_create_job_stores_tts_audio_url(self, mock_mediakit, mock_cosyvoice):
-        """v3: TTS 直生模式下 job.audio_url 为转存到自家 OSS 的永久地址."""
+        """v4: TTS 模式下 create_job 返回 tts_processing 状态，audio_url 尚未设置（由 Celery 任务处理）."""
         from app.services.lipsync_service import LipsyncService
 
         mock_db = MagicMock()
         mock_repo = MagicMock()
         mock_repo.get.return_value = None
 
-        with (
-            patch("app.services.lipsync_service.get_shared_storage_service") as storage_patch,
-            patch("app.services.lipsync_service.safe_download_bytes") as dl_patch,
-        ):
-            storage_patch.return_value.upload_file.return_value = "https://my-oss/permanent.mp3"
-            dl_patch.return_value = b"audio-bytes"
+        with patch(
+            "app.services.lipsync_service.tts_synthesize_and_submit"
+        ) as mock_task:
+            mock_task.apply_async.return_value = MagicMock(id="celery-task-789")
 
             svc = LipsyncService(
                 mock_db,
                 client=mock_mediakit,
-                cosyvoice_service=mock_cosyvoice,
                 voice_clone_repo=mock_repo,
             )
             job = svc.create_job(
@@ -461,8 +449,11 @@ class TestLipsyncServiceUnit:
                 script_text="这是一段测试文本",
             )
 
-        # job.audio_url 是转存 OSS 后的永久地址
-        assert job.audio_url == "https://my-oss/permanent.mp3"
+        # TTS 模式下 create_job 返回 tts_processing 状态
+        assert job.status == "tts_processing"
+        # audio_url 尚未设置（由 Celery 任务异步处理），模型默认为空字符串
+        assert not job.audio_url
+        mock_task.apply_async.assert_called_once()
 
     def test_create_job_direct_audio_skips_tts(self, mock_mediakit, mock_cosyvoice):
         """v3: 直接音频模式（传 audio_url）不触发 TTS，原样把 audio_url 提交 MediaKit."""
@@ -548,9 +539,9 @@ class TestErrorHandling:
         assert exc_info.value.code == "VoiceNotReady"
 
     def test_tts_value_error_mapped_to_invalid_param(self, mock_mediakit):
-        """v3: CosyVoice 抛 ValueError（参数无效）被包装为 TTSInvalidParam（路由映射 400）."""
+        """v4: TTS 模式下 create_job 不再同步调用 CosyVoice，
+        而是 dispatch Celery 任务。ValueError 由 Celery 任务内部处理。"""
         from app.services.lipsync_service import LipsyncService
-        from app.services.mediakit_client import MediaKitError
 
         mock_cosyvoice = MagicMock()
         mock_cosyvoice.submit_synthesize_task.side_effect = ValueError("voice_id 为空")
@@ -562,18 +553,27 @@ class TestErrorHandling:
         svc = LipsyncService(
             mock_db,
             client=mock_mediakit,
-            cosyvoice_service=mock_cosyvoice,
             voice_clone_repo=mock_repo,
         )
 
-        with pytest.raises(MediaKitError) as exc_info:
-            svc.create_job(
+        with patch(
+            "app.services.lipsync_service.tts_synthesize_and_submit"
+        ) as mock_task:
+            mock_task.apply_async.return_value = MagicMock(id="celery-task-789")
+
+            # TTS 模式下 create_job 不再同步失败
+            job = svc.create_job(
                 user_id="user-1",
                 video_url="https://example.com/video.mp4",
                 voice_id="some-voice",
                 script_text="test",
             )
-        assert exc_info.value.code == "TTSInvalidParam"
+
+        # 确认返回 tts_processing 状态
+        assert job.status == "tts_processing"
+        # TTS 合成由 Celery 任务处理，不直接调用 CosyVoice
+        mock_cosyvoice.submit_synthesize_task.assert_not_called()
+        mock_task.apply_async.assert_called_once()
 
     def test_missing_both_inputs_raises_invalid_input(self, mock_mediakit, mock_cosyvoice):
         """v3: 既无 audio_url 又无 voice_id+script_text 时抛 InvalidInput（路由映射 400）."""

tests/unit/test_lipsync_speed_optimization.py

@@ -0,0 +1,623 @@
+"""AI 数字人口型视频生成速度优化 — 单元测试.
+
+验证两个优化点：
+1. FFmpeg 编码 preset 从 fast 改为 veryfast（提速 30~50%）
+2. TTS 合成从同步改为 Celery 异步任务（API 响应从 6~35s 降到 <1s）
+
+Issue: lipsync-speed-optimization
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# 优化1: FFmpeg 编码提速 — preset veryfast
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestFFmpegPresetOptimization:
+    """验证 FFmpeg 编码命令从 -preset fast 改为 -preset veryfast."""
+
+    def test_preset_is_veryfast(self):
+        """_build_ffmpeg_command 输出必须包含 -preset veryfast."""
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
+        cmd = svc._build_ffmpeg_command(
+            input_video="https://example.com/video.mp4",
+            b_roll_segments=[],
+            filter_complex="",
+            final_label=None,
+            output_path="/tmp/output.mp4",
+        )
+        assert "-preset veryfast" in cmd, f"期望 -preset veryfast，实际命令: {cmd}"
+
+    def test_preset_veryfast_with_filter(self):
+        """带滤镜场景下也必须使用 veryfast."""
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
+        cmd = svc._build_ffmpeg_command(
+            input_video="https://example.com/video.mp4",
+            b_roll_segments=[],
+            filter_complex="overlay=0:0",
+            final_label="[v]",
+            output_path="/tmp/output.mp4",
+        )
+        assert "-preset veryfast" in cmd
+        assert "-filter_complex" in cmd
+
+    def test_preset_not_fast(self):
+        """确保不再使用旧的 -preset fast."""
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
+        cmd = svc._build_ffmpeg_command(
+            input_video="https://example.com/video.mp4",
+            b_roll_segments=[],
+            filter_complex="",
+            final_label=None,
+            output_path="/tmp/output.mp4",
+        )
+        # 确保是 veryfast 而不是 fast
+        assert "-preset veryfast" in cmd
+        # 排除 "fast" 单独出现（veryfast 包含 fast 子串，需精确判断）
+        parts = cmd.split()
+        preset_idx = parts.index("-preset")
+        assert parts[preset_idx + 1] == "veryfast"
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# 优化2: TTS 合成 Celery 异步化
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def _make_service_with_mocks():
+    """构造 LipsyncService 测试实例及 mock 依赖."""
+    from app.services.lipsync_service import LipsyncService
+
+    db = MagicMock()
+    client = MagicMock()
+    client.is_available = True
+    client.submit_lipsync.return_value = {
+        "success": True,
+        "task_id": "mk-1",
+        "request_id": "req-1",
+    }
+    cosy = MagicMock()
+    cosy.submit_synthesize_task.return_value = {
+        "audio_url": "https://tts/raw.mp3",
+        "request_id": "tts-req",
+        "audio_duration": 3.0,
+    }
+    svc = LipsyncService(db, client=client, cosyvoice_service=cosy, voice_clone_repo=MagicMock())
+    # _resolve_voice_id 默认原样返回（repo.get 返回 None）
+    svc._voice_clone_repo.get.return_value = None
+    return svc, client, cosy
+
+
+class TestCreateJobAsyncTTS:
+    """验证 TTS 模式改为 Celery 异步后的行为."""
+
+    def test_tts_mode_returns_tts_processing_status(self):
+        """TTS 模式下 create_job 立即返回，状态为 tts_processing."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+            mock_task.apply_async = MagicMock()
+
+            job = svc.create_job(
+                user_id="user-1",
+                video_url="https://example.com/video.mp4",
+                voice_id="longxiaochun_v3",
+                script_text="大家好",
+                speed=1.0,
+                emotion="",
+            )
+
+        assert job.status == "tts_processing"
+
+    def test_tts_mode_dispatches_celery_task(self):
+        """TTS 模式必须 dispatch Celery 异步任务."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+            mock_task.apply_async = MagicMock()
+
+            svc.create_job(
+                user_id="user-1",
+                video_url="https://example.com/video.mp4",
+                voice_id="v-1",
+                script_text="测试文本",
+            )
+
+        mock_task.apply_async.assert_called_once()
+        call_kwargs = mock_task.apply_async.call_args
+        args = call_kwargs.kwargs.get("args") or call_kwargs[1].get("args", call_kwargs[0][0] if call_kwargs[0] else ())
+        assert args[1] == "user-1"  # user_id
+        assert args[2] == "v-1"  # voice_id
+        assert args[3] == "测试文本"  # script_text
+
+    def test_tts_mode_celery_dispatch_failure_still_creates_job(self):
+        """Celery dispatch 失败时，job 记录已创建，状态保持 tts_processing."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+            mock_task.apply_async = MagicMock(side_effect=Exception("Celery broker down"))
+
+            job = svc.create_job(
+                user_id="user-1",
+                video_url="https://example.com/video.mp4",
+                voice_id="v-1",
+                script_text="测试文本",
+            )
+
+        # job 已创建
+        assert job is not None
+        assert job.status == "tts_processing"
+        # MediaKit 未被调用
+        client.submit_lipsync.assert_not_called()
+
+    def test_tts_mode_voice_validation_still_sync(self):
+        """TTS 模式下音色校验仍在 HTTP 请求中同步执行."""
+        from app.services.mediakit_client import MediaKitError
+
+        svc, client, cosy = _make_service_with_mocks()
+        # 模拟音色属于其他用户
+        other_profile = MagicMock()
+        other_profile.user_id = "user-other"
+        svc._voice_clone_repo.get.return_value = other_profile
+
+        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit"):
+            with pytest.raises(MediaKitError) as exc:
+                svc.create_job(
+                    user_id="user-1",
+                    video_url="https://example.com/video.mp4",
+                    voice_id="clone-profile-id",
+                    script_text="测试",
+                )
+        assert exc.value.code == "VoiceForbidden"
+
+    def test_tts_mode_missing_input_raises_immediately(self):
+        """缺少 voice_id 或 script_text 时立即报错，不 dispatch Celery 任务."""
+        from app.services.mediakit_client import MediaKitError
+
+        svc, client, cosy = _make_service_with_mocks()
+
+        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task:
+            mock_task.delay = MagicMock()
+
+            with pytest.raises(MediaKitError) as exc:
+                svc.create_job(
+                    user_id="user-1",
+                    video_url="https://example.com/video.mp4",
+                    # 缺少 voice_id 和 script_text
+                )
+            assert exc.value.code == "InvalidInput"
+
+        # Celery 任务未被 dispatch
+        mock_task.delay.assert_not_called()
+        # TTS 和 MediaKit 均未调用
+        cosy.submit_synthesize_task.assert_not_called()
+        client.submit_lipsync.assert_not_called()
+
+
+class TestCreateJobDirectAudio:
+    """验证直接音频模式不受异步化影响."""
+
+    def test_direct_audio_still_submits_synchronously(self):
+        """直接音频模式仍然同步提交 MediaKit，状态为 submitted."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task:
+            mock_task.delay = MagicMock()
+
+            job = svc.create_job(
+                user_id="user-1",
+                video_url="https://example.com/video.mp4",
+                audio_url="https://example.com/audio.mp3",
+            )
+
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-1"
+        client.submit_lipsync.assert_called_once()
+        # TTS Celery 任务不应被调用
+        mock_task.delay.assert_not_called()
+
+    def test_direct_audio_skips_tts(self):
+        """直接音频模式不调用 CosyVoice TTS."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        job = svc.create_job(
+            user_id="user-1",
+            video_url="https://example.com/video.mp4",
+            audio_url="https://example.com/audio.mp3",
+        )
+
+        cosy.submit_synthesize_task.assert_not_called()
+        call_kwargs = client.submit_lipsync.call_args
+        assert call_kwargs.kwargs["audio_url"] == "https://example.com/audio.mp3"
+
+
+class TestCancelJobTtsProcessing:
+    """验证 tts_processing 状态的任务可以被取消."""
+
+    def test_cancel_tts_processing(self):
+        """tts_processing 状态的任务可以成功取消."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        mock_job = MagicMock()
+        mock_job.status = "tts_processing"
+        mock_job.id = "job-1"
+        svc.get_job = MagicMock(return_value=mock_job)
+
+        result = svc.cancel_job("job-1", "user-1")
+        assert result.status == "cancelled"
+
+    def test_cancel_pending_still_works(self):
+        """pending 状态仍可取消."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        mock_job = MagicMock()
+        mock_job.status = "pending"
+        mock_job.id = "job-1"
+        svc.get_job = MagicMock(return_value=mock_job)
+
+        result = svc.cancel_job("job-1", "user-1")
+        assert result.status == "cancelled"
+
+    def test_cancel_submitted_still_works(self):
+        """submitted 状态仍可取消."""
+        svc, client, cosy = _make_service_with_mocks()
+
+        mock_job = MagicMock()
+        mock_job.status = "submitted"
+        mock_job.id = "job-1"
+        svc.get_job = MagicMock(return_value=mock_job)
+
+        result = svc.cancel_job("job-1", "user-1")
+        assert result.status == "cancelled"
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# lipsync_tts.py — Celery 异步任务单元测试
+# ═══════════════════════════════════════════════════════════════════════════════
+
+import sys
+import types
+
+
+class _FakeQuery:
+    """模拟 SQLAlchemy query.filter().first() 链式调用."""
+
+    def __init__(self, job):
+        self._job = job
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def first(self):
+        return self._job
+
+
+def _make_fake_job(**kwargs):
+    """构造可 setattr 的 job 记录."""
+    job = MagicMock()
+    job.id = kwargs.get("job_id", "job-1")
+    job.user_id = kwargs.get("user_id", "user-1")
+    job.status = kwargs.get("status", "tts_processing")
+    job.audio_url = kwargs.get("audio_url", "")
+    job.video_url = kwargs.get("video_url", "https://oss/video.mp4")
+    job.mediakit_task_id = kwargs.get("mediakit_task_id", "")
+    job.enable_video_loop = kwargs.get("enable_video_loop", False)
+    job.error_code = ""
+    job.error_message = ""
+    job.submitted_at = None
+    job.updated_at = None
+    return job
+
+
+def _build_session(job):
+    """构造 mock DB session + factory. 返回 (session, factory_patch_ctx_value)."""
+    session = MagicMock()
+    session.query.return_value = _FakeQuery(job)
+    session.commit = MagicMock()
+    session.close = MagicMock()
+    factory = MagicMock(return_value=session)
+    return session, factory
+
+
+def _apply_all_patches(
+    *,
+    job=None,
+    cosyvoice_service=None,
+    cosyvoice_side_effect=None,
+    cosyvoice_error=None,
+    download_bytes=b"AUDIO",
+    download_error=None,
+    storage=None,
+    mk_client=None,
+    mk_submit_return=None,
+    mk_submit_error=None,
+):
+    """统一构造测试需要的 patch 列表.
+
+    lipsync_tts.run() 在函数体内部懒 import 多个模块，通过 sys.modules 注入
+    伪造包路径避免真实导入；对存在的模块用 patch() 替换返回值/side_effect。
+    """
+    # 构造不存在的 database 模块
+    fake_db_mod = types.ModuleType("packages.adapters.sqlalchemy_impl.database")
+    session, factory = _build_session(job)
+    fake_db_mod.SessionLocal = factory
+
+    patches = [
+        patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.database": fake_db_mod}),
+        patch(
+            "app.services.lipsync_service.LipsyncService._sign_media_url",
+            side_effect=lambda url: url + "?signed" if url else url,
+        ),
+    ]
+
+    # CosyVoice
+    if cosyvoice_service is not None:
+        cosy_instance = cosyvoice_service
+    else:
+        cosy_instance = MagicMock()
+        if cosyvoice_side_effect is not None:
+            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_side_effect
+        elif cosyvoice_error is not None:
+            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_error
+        else:
+            cosy_instance.submit_synthesize_task.return_value = {"audio_url": "https://tts/raw.mp3"}
+    patches.append(patch("packages.application.cosyvoice_service.CosyVoiceService", return_value=cosy_instance))
+
+    # safe_download_bytes
+    if download_error is not None:
+        patches.append(patch("packages.shared.url_security.safe_download_bytes", side_effect=download_error))
+    else:
+        patches.append(patch("packages.shared.url_security.safe_download_bytes", return_value=download_bytes))
+
+    # Storage
+    if storage is None:
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.upload_file.return_value = "https://oss.example.com/tts.mp3"
+    patches.append(patch("packages.shared.storage.get_shared_storage_service", return_value=storage))
+
+    # MediaKit client
+    if mk_client is not None:
+        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=mk_client))
+    else:
+        client = MagicMock()
+        if mk_submit_error is not None:
+            client.submit_lipsync.side_effect = mk_submit_error
+        else:
+            client.submit_lipsync.return_value = mk_submit_return or {"task_id": "mk-1"}
+        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=client))
+
+    return session, patches
+
+
+class TestTtsSynthesizeAndSubmit:
+    """测试 Celery 任务 tts_synthesize_and_submit.run 的所有分支."""
+
+    def test_job_not_found_returns_early(self):
+        """Job 不存在 → 日志报错直接返回，不抛异常."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        session, patches = _apply_all_patches(job=None)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("missing-job", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        session.commit.assert_not_called()
+        session.close.assert_called_once()
+
+    def test_cancelled_job_skipped(self):
+        """Job 已 cancelled → 跳过不处理，不调用 TTS/MediaKit."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="cancelled")
+        session, patches = _apply_all_patches(job=job)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        session.commit.assert_not_called()
+        session.close.assert_called_once()
+        assert job.status == "cancelled"
+
+    def test_happy_path_tts_to_mediakit(self):
+        """正常流程：TTS 合成 → 下载 → OSS → 签名 → 提交 MediaKit → submitted."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.upload_file.return_value = "https://oss.example.com/lipsync-tts/u/j.mp3"
+
+        session, patches = _apply_all_patches(
+            job=job,
+            storage=storage,
+            mk_submit_return={"task_id": "mk-999"},
+        )
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.audio_url == "https://oss.example.com/lipsync-tts/u/j.mp3"
+        assert job.mediakit_task_id == "mk-999"
+        assert job.status == "submitted"
+        assert job.submitted_at is not None
+        session.close.assert_called_once()
+
+    def test_tts_cosyvoice_error_marks_failed(self):
+        """CosyVoiceError → 标记 failed，error_code=TTSSynthesisFailed."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        from packages.application.cosyvoice_service import CosyVoiceError
+
+        job = _make_fake_job(status="tts_processing")
+        session, patches = _apply_all_patches(
+            job=job,
+            cosyvoice_error=CosyVoiceError("TTS 服务异常"),
+        )
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSSynthesisFailed"
+        assert "TTS 合成失败" in job.error_message
+        session.close.assert_called_once()
+
+    def test_tts_value_error_marks_failed(self):
+        """ValueError → 标记 failed，error_code=TTSInvalidParam."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        session, patches = _apply_all_patches(
+            job=job,
+            cosyvoice_error=ValueError("speed 参数非法"),
+        )
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSInvalidParam"
+        assert "TTS 参数错误" in job.error_message
+        session.close.assert_called_once()
+
+    def test_tts_no_audio_url_marks_failed(self):
+        """TTS 返回空 audio_url → 标记 failed，error_code=TTSNoAudio."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        cosy = MagicMock()
+        cosy.submit_synthesize_task.return_value = {"audio_url": ""}
+        session, patches = _apply_all_patches(job=job, cosyvoice_service=cosy)
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSNoAudio"
+        session.close.assert_called_once()
+
+    def test_oss_upload_failure_falls_back_to_temp_url(self):
+        """OSS 上传失败 → 回退临时 URL，继续提交 MediaKit."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.upload_file.side_effect = Exception("OSS 上传超时")
+        session, patches = _apply_all_patches(
+            job=job,
+            storage=storage,
+            mk_submit_return={"task_id": "mk-77"},
+        )
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        # 回退到临时 URL
+        assert job.audio_url == "https://tts/raw.mp3"
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-77"
+        session.close.assert_called_once()
+
+    def test_mediakit_submit_failure_marks_failed(self):
+        """MediaKit 提交失败（MediaKitError）→ 标记 failed."""
+        from app.services.mediakit_client import MediaKitError
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        err = MediaKitError("GPU 不可用", code="MediaKitUnavailable")
+        session, patches = _apply_all_patches(
+            job=job,
+            mk_submit_error=err,
+        )
+        for p in patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "MediaKitUnavailable"
+        session.close.assert_called_once()
+
+    def test_top_level_exception_marks_async_task_error(self):
+        """顶层意外异常 → except 分支回写 failed，error_code=AsyncTaskError."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="tts_processing")
+        # 不调用 _apply_all_patches，手动构造所有 patch，让 CosyVoiceService 抛异常
+        fake_db_mod = types.ModuleType("packages.adapters.sqlalchemy_impl.database")
+        session_mock = MagicMock()
+        session_mock.query.return_value = _FakeQuery(job)
+        session_mock.commit = MagicMock()
+        session_mock.close = MagicMock()
+        fake_db_mod.SessionLocal = MagicMock(return_value=session_mock)
+
+        all_patches = [
+            patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.database": fake_db_mod}),
+            patch(
+                "app.services.lipsync_service.LipsyncService._sign_media_url",
+                side_effect=lambda url: url + "?signed" if url else url,
+            ),
+            patch(
+                "packages.application.cosyvoice_service.CosyVoiceService",
+                side_effect=RuntimeError("unexpected init failure"),
+            ),
+            patch("packages.shared.url_security.safe_download_bytes", return_value=b"AUDIO"),
+            patch("packages.shared.storage.get_shared_storage_service", return_value=MagicMock()),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=MagicMock()),
+        ]
+        for p in all_patches:
+            p.__enter__()
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(all_patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "AsyncTaskError"
+        assert "TTS 异步任务执行异常" in job.error_message
+        session_mock.close.assert_called_once()