fix(worker): 注册 ai_avatar_render Celery 任务到 Worker，修复渲染卡 0%

渲染进度卡 0% 根因：ai_avatar_render.execute 任务定义在 API 侧
(app.tasks.ai_avatar_render)，装饰在 API 自己的 Celery 实例上；Worker
用 worker_app.celery_app 实例，conf.imports 从未导入该任务，Worker 任务
注册表无此 task，消息路由到默认 celery 队列后无人消费。

修复：
- 新增 worker_app/tasks/ai_avatar_render.py，用 worker celery_app 注册
  同名任务 ai_avatar_render.execute，复用 AiAvatarRenderService.execute_render
- celery_app.conf.imports 注册 worker_app.tasks.ai_avatar_render
- worker.Dockerfile 复制完整 apps/api/app 到 /app/app，使 Worker 可
  import app.services（原先只复制 config.py + core/，缺 services）

任务名保持不变，与 API 端 .delay() 消息一致；走默认 celery 队列由
transcode worker 消费。

.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 + ruff)
+    - name: Auto-fix formatting (black + isort)
       if: failure()
       shell: sh
       env:

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

@@ -14,6 +14,7 @@ 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,
 )
@@ -23,6 +24,8 @@ 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()
@@ -31,11 +34,13 @@ 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 解析
-    # TTS 合成已移至 Celery 异步任务，无需同步注入 cosyvoice_service
+    # voice_clone_repo 用于克隆音色 profile 解析；cosyvoice_service 用于 TTS 直生
+    # （TTS 合成、音色解析、错误码归一化都在 LipsyncService 内部完成）
     return LipsyncService(
         db,
+        cosyvoice_service=cosyvoice_service,
         voice_clone_repo=voice_clone_repo,
     )
 
@@ -52,8 +57,8 @@ def create_lipsync_job(
     """提交对口型任务.
 
     #1809/#1822: 前端传 {video_url, voice_id, script_text, speed?, emotion?}，
-    后端创建任务记录（状态 tts_processing），dispatch Celery 异步任务执行 TTS 合成 + MediaKit 提交；
-    也支持直接传 {video_url, audio_url}（同步提交 MediaKit）。
+    后端内部解析音色、调 TTS 合成音频、转存 OSS，再提交 MediaKit；
+    也支持直接传 {video_url, audio_url}。
     """
     try:
         job = svc.create_job(
@@ -70,13 +75,21 @@ 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:
-        # 音色无权访问 → 403；参数无效 → 400；MediaKit 提交失败 → 502
+        # TTS 合成失败 / 音色无权访问 → 400/403；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={
@@ -174,13 +187,13 @@ def cancel_lipsync_job(
     current_user: AuthenticatedUser = Depends(get_current_user),
     svc: LipsyncService = Depends(_get_service),
 ):
-    """取消对口型任务（仅 pending/tts_processing/submitted 状态可取消）."""
+    """取消对口型任务（仅 pending/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/tts_processing/submitted 可取消",
+            detail=f"任务状态 {job.status} 不可取消，仅 pending/submitted 可取消",
         )
     return job

apps/api/app/services/lipsync_service.py

@@ -25,9 +25,6 @@ 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
@@ -157,31 +154,41 @@ class LipsyncService:
         enable_video_loop: bool = False,
         project_id: str = "",
     ) -> LipsyncJobModel:
-        """创建对口型任务.
+        """创建对口型任务并提交到 MediaKit.
 
         两种输入模式：
-        - TTS 直生：voice_id + script_text（audio_url 留空）
-          → 先创建 DB 记录（状态 tts_processing），再 dispatch Celery 异步任务
-            执行 TTS 合成 + MediaKit 提交。API 响应 <1s。
+        - TTS 直生：voice_id + script_text（audio_url 留空），后端先合成音频
         - 直接音频：提供 audio_url
-          → 同步提交 MediaKit，状态直接设为 submitted。
 
         Raises:
-            MediaKitError: 参数校验失败或 MediaKit 提交失败（仅直接音频模式）
+            MediaKitError: TTS 合成或 MediaKit 提交失败
         """
-        # 0. 输入校验
+        # 0. TTS 直生模式：先合成音频（在创建 DB 记录之前完成，失败直接抛出）
         if not audio_url:
             if not (voice_id and script_text):
                 raise MediaKitError(
                     "必须提供 audio_url 或 voice_id+script_text",
                     code="InvalidInput",
                 )
-            # TTS 模式：在 HTTP 请求中同步校验音色归属，快速失败
-            self._resolve_voice_id(voice_id, user_id)
+            # 预合成：用临时 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)
 
         # 1. 创建数据库记录
         job_id = str(uuid.uuid4())
-        is_tts_mode = not bool(audio_url)
         job = LipsyncJobModel(
             id=job_id,
             user_id=user_id,
@@ -193,53 +200,28 @@ class LipsyncService:
             script_text=script_text or "",
             speed=speed,
             emotion=normalize_emotion(emotion),
-            status="tts_processing" if is_tts_mode else "pending",
+            status="pending",
         )
         self.db.add(job)
         self.db.flush()
 
-        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
+        # 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
 
         self.db.commit()
         self.db.refresh(job)
@@ -378,12 +360,12 @@ class LipsyncService:
     # ── 取消任务 ──────────────────────────────────────────────────────────
 
     def cancel_job(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
-        """取消任务（仅 pending/tts_processing/submitted 状态可取消）."""
+        """取消任务（仅 pending/submitted 状态可取消）."""
         job = self.get_job(job_id, user_id)
         if job is None:
             return None
 
-        if job.status in ("pending", "tts_processing", "submitted"):
+        if job.status in ("pending", "submitted"):
             job.status = "cancelled"
             job.updated_at = datetime.now(timezone.utc)
             self.db.commit()

apps/api/app/tasks/lipsync_tts.py

@@ -1,212 +0,0 @@
-"""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
-
-注意：使用 @shared_task 而非绑定到某个 celery_app 实例，
-确保任务能被 Worker 侧 celery_app 正确注册，同时 API 侧 send_task/apply_async 仍可正常调用。
-"""
-
-import io
-import logging
-from datetime import datetime, timezone
-from urllib.parse import urlparse
-
-from celery import shared_task
-
-logger = logging.getLogger(__name__)
-
-# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService._sign_media_url 保持一致
-_MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600
-
-
-def _sign_media_url(url: str) -> str:
-    """对自家 OSS 私有桶 URL 重签长有效期预签名.
-
-    - 自家 OSS URL → 重签 7 天有效期
-    - 外部临时 URL → 原样透传
-    - 任何异常降级原样返回，不阻断主流程
-    """
-    if not url:
-        return 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 url
-        own_host = urlparse(public_base).netloc.lower()
-        host = urlparse(url).netloc.lower()
-        if not own_host or host != own_host:
-            return url
-        signed = storage.get_download_url(url, expires_seconds=_MEDIAKIT_URL_TTL_SECONDS)
-        return signed or url
-    except Exception as exc:  # noqa: BLE001
-        logger.warning("[lipsync_tts] URL 重签失败，原样返回: url_prefix=%s err=%s", url[:80], exc)
-        return url
-
-
-@shared_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.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,
-            )
-            from packages.shared.storage import get_shared_storage_service
-
-            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（复用模块内 _sign_media_url，避免对 LipsyncService 的耦合）
-        audio_url = _sign_media_url(job.audio_url)
-        video_url = _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

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

apps/web/src/pages/ai-avatar/api/aiAvatar.ts

@@ -54,7 +54,7 @@ export const createLipsyncJob = async (data: {
 }
 
 export const getLipsyncJob = async (id: string): Promise<LipsyncJob> => {
-  const response = await apiClient.get<LipsyncJob>(`/lipsync/jobs/${id}`, { timeout: 60000 })
+  const response = await apiClient.get<LipsyncJob>(`/lipsync/jobs/${id}`)
   return response.data
 }
 

apps/worker/worker_app/celery_app.py

@@ -34,10 +34,11 @@ celery_app.conf.imports = (
     "worker_app.tasks.tts_synthesis",
     "worker_app.tasks.batch_download",
     "worker_app.tasks.duplication_check",
+    # #1798 AI 数字人渲染：必须在 Worker 实例上注册同名任务，否则消息无人消费（渲染卡 0%）
+    "worker_app.tasks.ai_avatar_render",
     "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",
     "worker_app.tasks.cleanup",
-    "apps.api.app.tasks.lipsync_tts",
 )
 
 # Celery Beat 定时任务调度

apps/worker/worker_app/tasks/ai_avatar_render.py

@@ -0,0 +1,48 @@
+"""AI 数字人渲染任务 — Worker 侧 Celery 任务注册.
+
+#1798 渲染进度卡在 0% 的根因：渲染任务定义在 API 侧（`app.tasks.ai_avatar_render`），
+装饰在 API 自己的 Celery 实例（`xiaoxia-saas-api`）上；而 Worker 用的是
+`worker_app.celery_app` 实例，`conf.imports` 从未导入该任务，Worker 的任务
+注册表里没有 `ai_avatar_render.execute`，消息被路由到默认 `celery` 队列后
+无人消费，任务永远停在 0%。
+
+修复：在 Worker 侧用 `worker_app.celery_app` 注册同名任务，直接调用与 API
+服务一致的 `AiAvatarRenderService.execute_render` 核心管线（业务逻辑在
+`apps.api.app.services`，worker 镜像已复制 `apps/api/app`）。任务名保持
+`ai_avatar_render.execute`，与 API 生产端 `.delay()` 的消息路由一致；未在
+task_routes 显式配置，走默认 `celery` 队列，由 transcode worker 消费。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, name="ai_avatar_render.execute", max_retries=2)
+def execute_ai_avatar_render(self, job_id: str) -> dict:
+    """执行 AI 数字人渲染管线（Worker 侧入口）.
+
+    进度由 service 直接写入 DB（AiAvatarRenderJob.progress:
+    0→5→20→40→80→90→95→100），API 通过轮询 progress 字段展示。
+    """
+    logger.info("开始执行渲染任务: %s", job_id)
+    self.update_state(state="PROCESSING", meta={"progress": 0, "job_id": job_id})
+
+    session = SessionLocal()
+    try:
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        service = AiAvatarRenderService(session)
+        service.execute_render(job_id)
+        return {"status": "completed", "job_id": job_id}
+    except Exception as exc:
+        logger.exception("渲染任务执行异常 [%s]: %s", job_id, exc)
+        self.update_state(state="FAILED", meta={"progress": 0, "error": str(exc)})
+        raise
+    finally:
+        session.close()

infra/docker/worker.Dockerfile

@@ -20,7 +20,7 @@ WORKDIR /app
 
 # 设置 Python 环境变量
 ENV PATH="/opt/venv/bin:$PATH"
-ENV PYTHONPATH=/app:/app/apps/api:/app/packages
+ENV PYTHONPATH=/app:/app/packages
 ENV PYTHONUNBUFFERED=1
 ENV APP_VERSION=$APP_VERSION
 
@@ -28,10 +28,8 @@ ENV APP_VERSION=$APP_VERSION
 COPY alembic.ini /app/alembic.ini
 COPY migrations/ /app/migrations/
 COPY packages/ /app/packages/
-# 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/
+# #1798 复制完整 API app 包到 /app/app：Worker 的 AI 数字人渲染任务需要 import app.services / app.config / app.core
+COPY apps/api/app/ /app/app/
 
 # Worker 启动脚本
 COPY infra/docker/entrypoint-worker.sh /usr/local/bin/entrypoint-worker.sh

scripts/ci/auto_fix_formatting.py

@@ -1,13 +1,10 @@
 #!/usr/bin/env python3
-"""CI中自动修复代码格式（Python: black + isort + ruff | Frontend: prettier），并推送回原分支。
+"""CI中自动修复代码格式（Python: black + isort | Frontend: prettier），并推送回原分支。
 
 - PR事件：所有PR只要Code Quality因格式问题失败，自动修复并push回源分支
 - Push事件（develop/main）：自动修复并push回原分支，保持主干格式永远正确
 - 防循环：修复commit带 [skip ci-format-check] 标记，检测到该标记则跳过修复
-- black/isort/prettier 修格式；ruff check --fix --unsafe-fixes 自动修复
-  ruff 可修复的 lint 规则（含 F401 未使用 import 等 unsafe fix）
-- ruff 目标范围与 validate_style.sh 的检查范围对齐：apps packages tests
-  （alembic/scripts 不在 ruff 检查范围内，不做修复）
+- 只修格式（black/isort/prettier），ruff逻辑类错误不动
 当code quality检查因格式问题失败时触发。
 """
 
@@ -138,7 +135,7 @@ def get_pr_head_branch(pr_number, api_url, token):
 
 
 def fix_python(target_py_files, scan_mode):
-    """修复 Python 文件 (black 格式化 + isort 排序 + ruff lint 自动修复)"""
+    """修复 Python 文件格式 (black + isort)"""
     if not target_py_files:
         print("没有需要修复的 Python 文件，跳过")
         return
@@ -149,48 +146,14 @@ 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/ruff", file=sys.stderr)
+        print("black执行失败，但继续尝试isort", 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执行失败，继续尝试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 自动修复")
+        print("isort执行失败", file=sys.stderr)
 
 
 def fix_frontend(target_fe_files, scan_mode, repo_root):
@@ -371,7 +334,7 @@ def main():
     # 提交修复
     run("git clean -fd")
     run("git add -u")
-    run('git commit -m "style: auto-format with black + isort + ruff + prettier [skip ci-format-check]"')
+    run('git commit -m "style: auto-format with black + isort + 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,8 +132,14 @@ 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.tts_synthesize_and_submit") as mock_task:
-        mock_task.apply_async.return_value = MagicMock(id="celery-task-123")
+    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"
 
         job = svc.create_job(
             user_id="user-1",
@@ -144,16 +150,19 @@ def test_create_job_tts_direct_mode_synthesizes_audio():
             emotion="兴奋",
         )
 
-    # v4: TTS 模式下 create_job 返回 tts_processing 状态，dispatch Celery 任务
-    assert job.status == "tts_processing"
+    # 调了 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 字段
     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():
@@ -171,25 +180,23 @@ def test_create_job_direct_audio_mode_skips_tts():
 
 
 def test_create_job_tts_failure_raises():
-    """v4: TTS 模式下 create_job 不再同步失败，而是 dispatch Celery 任务。
-    TTS 合成失败由 Celery 任务内部处理并更新 job 状态。"""
+    from app.services.mediakit_client import MediaKitError
+
+    from packages.application.cosyvoice_service import CosyVoiceError
+
     svc, client, cosy = _lipsync_service_with_mocks()
+    cosy.submit_synthesize_task.side_effect = CosyVoiceError("Arrearage")
 
-    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(
+    with pytest.raises(MediaKitError) as exc:
+        svc.create_job(
             user_id="user-1",
             video_url="https://oss/person.mp4",
             voice_id="v-1",
             script_text="文本",
         )
-
-    # create_job 成功返回 tts_processing，不直接调用 TTS
-    assert job.status == "tts_processing"
-    cosy.submit_synthesize_task.assert_not_called()
+    assert exc.value.code == "TTSSynthesisFailed"
+    # TTS 失败不应提交 MediaKit
     client.submit_lipsync.assert_not_called()
-    mock_task.apply_async.assert_called_once()
 
 
 # ── refresh 同步中间状态 ────────────────────────────────────────────────

tests/unit/test_lipsync_routes.py

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

@@ -1,284 +0,0 @@
-"""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"

tests/unit/test_lipsync_tts.py

@@ -1,382 +0,0 @@
-"""AI 数字人口型 TTS Celery 异步任务 — 单元测试.
-
-覆盖 lipsync_tts.py 的全部主要分支：
-- Job 不存在/cancelled/正常/异常路径
-- TTS 合成、音频下载、OSS 上传、MediaKit 提交
-- CosyVoiceError/ValueError/MediaKitError/顶层异常等错误码
-"""
-
-import os
-import sys
-import types
-from types import ModuleType
-from unittest.mock import MagicMock, patch
-
-os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
-
-
-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)."""
-    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。
-    """
-    fake_db_mod = 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.tasks.lipsync_tts._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)
-
-        # cancelled 不应 commit，不应触发 TTS/MediaKit
-        session.commit.assert_not_called()
-        session.close.assert_called_once()
-
-    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()
-        mk_client = MagicMock()
-        mk_client.submit_lipsync.return_value = {"task_id": "mk-999"}
-        session, patches = _apply_all_patches(job=job, mk_client=mk_client)
-        entered = [p.__enter__() for p in patches]
-        try:
-            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好世界", 1.0, "happy")
-        finally:
-            for p in reversed(patches):
-                p.__exit__(None, None, None)
-
-        assert job.status == "submitted"
-        assert job.mediakit_task_id == "mk-999"
-        assert job.error_code == ""
-        mk_client.submit_lipsync.assert_called_once()
-        call_kwargs = mk_client.submit_lipsync.call_args.kwargs
-        assert call_kwargs["client_token"] == "job-1"
-        assert call_kwargs["audio_url"].endswith("?signed")
-        session.commit.assert_called()
-        session.close.assert_called_once()
-
-    def test_cosyvoice_error_marks_tts_synthesis_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()
-        session, patches = _apply_all_patches(job=job, cosyvoice_error=CosyVoiceError("tts boom"))
-        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)
-
-        assert job.status == "failed"
-        assert job.error_code == "TTSSynthesisFailed"
-        session.close.assert_called_once()
-
-    def test_value_error_marks_tts_invalid_param(self):
-        """ValueError（参数错误）→ failed, error_code=TTSInvalidParam."""
-        from app.tasks.lipsync_tts import tts_synthesize_and_submit
-
-        job = _make_fake_job()
-        session, patches = _apply_all_patches(job=job, cosyvoice_side_effect=ValueError("bad param"))
-        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)
-
-        assert job.status == "failed"
-        assert job.error_code == "TTSInvalidParam"
-        session.close.assert_called_once()
-
-    def test_no_audio_url_marks_tts_no_audio(self):
-        """TTS 返回空 audio_url → failed, error_code=TTSNoAudio."""
-        from app.tasks.lipsync_tts import tts_synthesize_and_submit
-
-        job = _make_fake_job()
-        cosy = MagicMock()
-        cosy.submit_synthesize_task.return_value = {"audio_url": ""}
-        session, patches = _apply_all_patches(job=job, cosyvoice_service=cosy)
-        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)
-
-        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，仍然 submitted."""
-        from app.tasks.lipsync_tts import tts_synthesize_and_submit
-
-        job = _make_fake_job()
-        storage = MagicMock()
-        storage.public_url = "https://oss.example.com"
-        storage.upload_file.side_effect = RuntimeError("oss down")
-        mk_client = MagicMock()
-        mk_client.submit_lipsync.return_value = {"task_id": "mk-7"}
-        session, patches = _apply_all_patches(job=job, storage=storage, mk_client=mk_client)
-        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)
-
-        # 上传失败后 audio_url 回退为临时 TTS URL，仍继续提交到 MediaKit
-        assert job.audio_url == "https://tts/raw.mp3"
-        assert job.status == "submitted"
-        assert job.mediakit_task_id == "mk-7"
-        mk_client.submit_lipsync.assert_called_once()
-        session.close.assert_called_once()
-
-    def test_mediakit_error_marks_mediakit_unavailable(self):
-        """MediaKit 提交失败 → failed, error_code=MediaKitUnavailable."""
-        from app.services.mediakit_client import MediaKitError
-        from app.tasks.lipsync_tts import tts_synthesize_and_submit
-
-        job = _make_fake_job()
-        mk_err = MediaKitError("mk down", code="MediaKitUnavailable")
-        session, patches = _apply_all_patches(job=job, mk_submit_error=mk_err)
-        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)
-
-        assert job.status == "failed"
-        assert job.error_code == "MediaKitUnavailable"
-        session.close.assert_called_once()
-
-    def test_top_level_exception_marks_async_task_error(self):
-        """顶层未预期异常 → failed, error_code=AsyncTaskError."""
-        from app.tasks.lipsync_tts import tts_synthesize_and_submit
-
-        job = _make_fake_job()
-        fake_db_mod = ModuleType("packages.adapters.sqlalchemy_impl.database")
-        session, factory = _build_session(job)
-        fake_db_mod.SessionLocal = factory
-
-        # CosyVoiceService 在 __init__ 抛 RuntimeError（非 CosyVoiceError/ValueError）
-        fake_cosy_mod = ModuleType("packages.application.cosyvoice_service")
-
-        class _CosyVoiceErrorForTest(Exception):
-            pass
-
-        class _BoomService:
-            def __init__(self):
-                raise RuntimeError("top-level boom")
-
-        fake_cosy_mod.CosyVoiceError = _CosyVoiceErrorForTest
-        fake_cosy_mod.CosyVoiceService = _BoomService
-
-        with patch.dict(
-            sys.modules,
-            {
-                "packages.adapters.sqlalchemy_impl.database": fake_db_mod,
-                "packages.application.cosyvoice_service": fake_cosy_mod,
-            },
-        ):
-            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
-
-        assert job.status == "failed"
-        assert job.error_code == "AsyncTaskError"
-        session.close.assert_called()
-
-
-class TestSignMediaUrl:
-    """覆盖模块内 _sign_media_url 的所有分支（CI 增量覆盖率需要）."""
-
-    def test_empty_url_returns_empty(self):
-        from app.tasks.lipsync_tts import _sign_media_url
-
-        assert _sign_media_url("") == ""
-        assert _sign_media_url(None) is None
-
-    def test_own_oss_url_signed(self):
-        """自家 OSS URL → 调用 storage.get_download_url 签名."""
-        from app.tasks.lipsync_tts import _sign_media_url
-
-        fake_storage = MagicMock()
-        fake_storage.public_url = "https://oss.example.com/"
-        fake_storage.get_download_url.return_value = "https://oss.example.com/a?sig=xyz"
-
-        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
-            result = _sign_media_url("https://oss.example.com/lipsync/a.mp3")
-
-        assert result == "https://oss.example.com/a?sig=xyz"
-        fake_storage.get_download_url.assert_called_once()
-
-    def test_external_url_passthrough(self):
-        """外部 URL（不是自家 OSS host）→ 原样透传，不签名."""
-        from app.tasks.lipsync_tts import _sign_media_url
-
-        fake_storage = MagicMock()
-        fake_storage.public_url = "https://oss.example.com/"
-
-        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
-            result = _sign_media_url("https://tts.example.com/raw.mp3")
-
-        assert result == "https://tts.example.com/raw.mp3"
-        fake_storage.get_download_url.assert_not_called()
-
-    def test_storage_exception_falls_back(self):
-        """storage 调用异常 → 降级原样返回，不抛错."""
-        from app.tasks.lipsync_tts import _sign_media_url
-
-        with patch(
-            "packages.shared.storage.get_shared_storage_service",
-            side_effect=RuntimeError("storage down"),
-        ):
-            result = _sign_media_url("https://oss.example.com/a.mp3")
-
-        assert result == "https://oss.example.com/a.mp3"
-
-    def test_no_public_url_passthrough(self):
-        """storage.public_url 为空 → 原样透传."""
-        from app.tasks.lipsync_tts import _sign_media_url
-
-        fake_storage = MagicMock()
-        fake_storage.public_url = ""
-
-        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
-            result = _sign_media_url("https://anything.example.com/a.mp3")
-
-        assert result == "https://anything.example.com/a.mp3"
-        fake_storage.get_download_url.assert_not_called()