fix(lipsync): 修复 lipsync_tts 跨容器 import 错误导致 API/Worker 启动崩溃

lipsync_tts.py 原从 apps.worker.worker_app 导入 celery_app，而 API 镜像中没有 apps.worker 目录，导致 API 容器导入 lipsync_service 时 ModuleNotFoundError，健康检查失败触发自动回滚。修复： 1. celery_app 改为从 app.core.celery_app 导入（API 侧 Celery app，API/Worker 镜像均存在） 2. _sign_media_url 改为模块内独立函数，去掉对 LipsyncService.__new__ 反射调用的依赖（Worker 镜像不复制 services 目录，反射导入会失败） 3. worker.Dockerfile 补充 apps/api/app/tasks/、mediakit_client.py 及必需的 __init__.py （Worker 进程需能 import lipsync_tts task 并注册 Celery 路由）
2026-09-10 20:48:24 +08:00
23 changed files with 417 additions and 695 deletions
@@ -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:
@@ -827,6 +827,9 @@ jobs:
        CACHE_REF="${REGISTRY}/${{ matrix.cache_name }}:develop"

        EXTRA_BUILD_ARGS="APP_VERSION=\"${GITHUB_SHA}\""
+        if [ "${{ matrix.service }}" = "web" ]; then
+          EXTRA_BUILD_ARGS="$EXTRA_BUILD_ARGS NGINX_CONF=infra/docker/nginx-staging.conf"
+        fi

        # Worker 与 API/Web 统一走持久 builder（ci-builder-persist），共享宿主机层缓存
        NO_CACHE_FLAG=""
@@ -1023,6 +1026,9 @@ jobs:
        CACHE_REF="${REGISTRY}/${{ matrix.cache_name }}:${GITHUB_REF_NAME}"

        EXTRA_BUILD_ARGS="APP_VERSION=\"${GITHUB_SHA}\""
+        if [ "${{ matrix.service }}" = "web" ]; then
+          EXTRA_BUILD_ARGS="$EXTRA_BUILD_ARGS NGINX_CONF=infra/docker/nginx-staging.conf"
+        fi

        NO_CACHE_FLAG=""
        for i in 1 2 3; do
@@ -1561,6 +1567,9 @@ jobs:
        CACHE_REF="${REGISTRY}/${{ matrix.cache_name }}:main"

        EXTRA_BUILD_ARGS="APP_VERSION=\"${TAG_NAME}\""
+        if [ "${{ matrix.service }}" = "web" ]; then
+          EXTRA_BUILD_ARGS="$EXTRA_BUILD_ARGS NGINX_CONF=infra/docker/nginx-production.conf"
+        fi

        # Docker build 带重试：失败自动重试2次，第2次重试加--no-cache
        NO_CACHE_FLAG=""
@@ -1,60 +0,0 @@
-name: "Debug: Web container v2 (mount conflict)"
-on:
-  push:
-    branches: [debug/web-crash-v2]
-  workflow_dispatch:
-jobs:
-  web-diag:
-    runs-on: runtime-builder
-    timeout-minutes: 10
-    steps:
-      - name: Setup SSH and diagnose
-        shell: bash
-        env:
-          STAGING_SSH_KEY: ${{ secrets.PREVIEW_SSH_KEY }}
-        run: |
-          set -x
-          which ssh || (apt-get update -qq && apt-get install -y -qq openssh-client)
-          mkdir -p ~/.ssh && chmod 700 ~/.ssh
-          printf "%s" "$STAGING_SSH_KEY" > ~/.ssh/id_rsa
-          chmod 600 ~/.ssh/id_rsa
-          H=47.98.113.167; P=22222
-          ssh-keyscan -p $P -H $H >> ~/.ssh/known_hosts 2>/dev/null
-          ssh -p $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@$H 'bash -s' <<'REMOTE'
-          set -x
-          echo "=== Current staging containers ==="
-          docker ps -a --filter name=xiaoxia-*-staging --format "table {{.Names}}\t{{.Status}}\t{{.Image}}"
-          echo ""
-          echo "=== Web container logs (current/current-rolledback) ==="
-          docker logs xiaoxia-web-staging 2>&1 | tail -40
-          echo ""
-          echo "=== Web inspect: env & mounts ==="
-          docker inspect xiaoxia-web-staging --format 'Entrypoint: {{.Config.Entrypoint}} Cmd: {{.Config.Cmd}}'
-          docker inspect xiaoxia-web-staging --format '{{range .Config.Env}}{{.}}{{"\n"}}{{end}}' | grep -E "APP_ENV|VERSION"
-          echo "Mounts:"
-          docker inspect xiaoxia-web-staging --format '{{range .Mounts}}{{.Type}} {{.Source}} -> {{.Destination}} (rw={{.RW}}){{"\n"}}{{end}}'
-          echo ""
-          echo "=== Reproduce: rm on read-only bind mount ==="
-          docker run --rm --name nginx-ro-test \
-            -v /var/lib/xiaoxia-saas-staging/nginx-staging.conf:/etc/nginx/conf.d/default.conf:ro \
-            git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/xiaoxia-saas-web:387514c \
-            sh -c '
-              set -x
-              echo "Before:"
-              ls -la /etc/nginx/conf.d/
-              echo "Try rm (as entrypoint does):"
-              rm -f /etc/nginx/conf.d/default.conf
-              echo "rm exitcode=$?"
-              echo "After rm:"
-              ls -la /etc/nginx/conf.d/
-              echo "Test ln:"
-              ln -s /etc/nginx/nginx-staging.conf /etc/nginx/conf.d/default.conf
-              echo "ln exitcode=$?"
-              ls -la /etc/nginx/conf.d/
-              echo "nginx -t:"
-              nginx -t 2>&1
-            ' 2>&1
-          echo ""
-          echo "=== Also test with NEW fixed image (9c0d4b1 if present) ==="
-          docker images | grep xiaoxia-saas-web | head -5
-REMOTE
@@ -1,29 +0,0 @@
-name: "Verify staging post-deploy"
-on:
-  push:
-    branches: [debug/verify-staging]
-  workflow_dispatch:
-jobs:
-  verify:
-    runs-on: runtime-builder
-    timeout-minutes: 10
-    steps:
-      - name: Setup SSH
-        shell: bash
-        env:
-          STAGING_SSH_KEY: ${{ secrets.PREVIEW_SSH_KEY }}
-        run: |
-          set -eux
-          which ssh || (apt-get update -qq && apt-get install -y -qq openssh-client)
-          mkdir -p ~/.ssh && chmod 700 ~/.ssh
-          printf "%s" "$STAGING_SSH_KEY" > ~/.ssh/id_rsa
-          chmod 600 ~/.ssh/id_rsa
-          H=47.98.113.167; P=22222
-          ssh-keyscan -p $P -H $H >> ~/.ssh/known_hosts 2>/dev/null
-      - name: Run verify
-        shell: bash
-        run: |
-          set -x
-          echo 'IyEvYmluL2Jhc2gKc2V0IC14CmVjaG8gIj09PSAxLiBDdXJyZW50IHJ1bm5pbmcgY29udGFpbmVycyA9PT0iCmRvY2tlciBwcyAtLWZpbHRlciBuYW1lPXhpYW94aWEtKi1zdGFnaW5nIC0tZm9ybWF0ICJ0YWJsZSB7ey5OYW1lc319XHR7ey5TdGF0dXN9fVx0e3suSW1hZ2V9fSIKZWNobyAiIgplY2hvICI9PT0gMi4gV2ViIGxvZ3MgKGxhc3QgMTUgbGluZXMpID09PSIKZG9ja2VyIGxvZ3MgLS10YWlsIDE1IHhpYW94aWEtd2ViLXN0YWdpbmcgMj4mMQplY2hvICIiCmVjaG8gIj09PSAzLiBXZWIgbG9jYWxob3N0IGhlYWx0aCBjaGVjayA9PT0iCmN1cmwgLXNJIC0tbWF4LXRpbWUgNSBodHRwOi8vMTI3LjAuMC4xOjMwMDEvIHwgaGVhZCAtNQplY2hvICIiCmVjaG8gIj09PSA0LiBBUEkgbG9jYWxob3N0IGhlYWx0aCBjaGVjayA9PT0iCmN1cmwgLXNmIC0tbWF4LXRpbWUgNSBodHRwOi8vMTI3LjAuMC4xOjgwMDAvaGVhbHRoCmVjaG8gIiIKZWNobyAiPT09IDUuIEFQSSB2ZXJzaW9uID09PSIKY3VybCAtcyAtLW1heC10aW1lIDUgaHR0cDovLzEyNy4wLjAuMTo4MDAwL2hlYWx0aCB8IHB5dGhvbjMgLWMgImltcG9ydCBzeXMsanNvbjsgZD1qc29uLmxvYWQoc3lzLnN0ZGluKTsgcHJpbnQoJ3ZlcnNpb246JywgZC5nZXQoJ3ZlcnNpb24nLCc/JykpIiAyPi9kZXYvbnVsbAplY2hvICIiCmVjaG8gIj09PSA2LiBXb3JrZXIgY29udGFpbmVyIHZlcnNpb24gPT09Igpkb2NrZXIgaW5zcGVjdCB4aWFveGlhLXdvcmtlci1zdGFnaW5nIC0tZm9ybWF0ICd7e3JhbmdlIC5Db25maWcuRW52fX17ey59fXt7IlxuIn19e3tlbmR9fScgfCBncmVwIEFQUF9WRVJTSU9OCmVjaG8gIiIKZWNobyAiPT09IDcuIFdvcmtlciBsb2dzIChzdGFydHVwICsgdGFzayByZWdpc3RyYXRpb24pID09PSIKZG9ja2VyIGxvZ3MgLS10YWlsIDUwIHhpYW94aWEtd29ya2VyLXN0YWdpbmcgMj4mMSB8IGdyZXAgLWlFICJyZWdpc3RlcmVkfGNlbGVyeUB8bGlwc3luY3xyZWFkeXxNb2R1bGVOb3RGb3VuZHxFcnJvcnx0YXNrIiB8IGhlYWQgLTMwCmVjaG8gIiIKZWNobyAiPT09IDguIExpc3QgYWxsIHJlZ2lzdGVyZWQgY2VsZXJ5IHRhc2tzIChXb3JrZXLlhoVpbnNwZWN0KSA9PT0iCmRvY2tlciBleGVjIHhpYW94aWEtd29ya2VyLXN0YWdpbmcgc2ggLWMgImNlbGVyeSAtQSB3b3JrZXJfYXBwLmNlbGVyeV9hcHAgaW5zcGVjdCByZWdpc3RlcmVkIDI+JjEgfCBoZWFkIC00MCIgfHwgZWNobyAiY2VsZXJ5IGluc3BlY3QgZmFpbGVkIgplY2hvICIiCmVjaG8gIj09PSA5LiBDaGVjayBpbXBvcnRzIGluc2lkZSBXb3JrZXIgKGxpcHN5bmNfdHRzIG1vZHVsZSkgPT09Igpkb2NrZXIgZXhlYyB4aWFveGlhLXdvcmtlci1zdGFnaW5nIHB5dGhvbiAtYyAiCmltcG9ydCBpbXBvcnRsaWIKbSA9IGltcG9ydGxpYi5pbXBvcnRfbW9kdWxlKCdhcHAudGFza3MubGlwc3luY190dHMnKQpwcmludCgnbGlwc3luY190dHMgbW9kdWxlIGltcG9ydGVkIE9LJykKcHJpbnQoJ3N5bnRoZXNpemVfYW5kX3N1Ym1pdDonLCBoYXNhdHRyKG0sICdzeW50aGVzaXplX2FuZF9zdWJtaXQnKSkKcHJpbnQoJ1Nlc3Npb25Mb2NhbDonLCBoYXNhdHRyKG0sICdTZXNzaW9uTG9jYWwnKSkKIiAyPiYxIHwgaGVhZCAtMTAKZWNobyAiIgplY2hvICI9PT0gMTAuIFB1YmxpYyBBUEkgaGVhbHRoIGNoZWNrIChzdGFnaW5nLWFwaSkgPT09IgpjdXJsIC1zSSAtLW1heC10aW1lIDEwIGh0dHBzOi8vc3RhZ2luZy1hcGkueGlhb3hpYWppYW5qaS5jb20vaGVhbHRoIHwgaGVhZCAtMwplY2hvICIiCmVjaG8gIj09PSAxMS4gUHVibGljIHdlYiBoZWFsdGggY2hlY2sgPT09IgpjdXJsIC1zSSAtLW1heC10aW1lIDEwIGh0dHBzOi8vc3RhZ2luZy54aWFveGlhamlhbmppLmNvbS8gfCBoZWFkIC0zCg==' | base64 -d > /tmp/verify.sh
-          chmod +x /tmp/verify.sh
-          ssh -p 22222 -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@47.98.113.167 'bash -s' < /tmp/verify.sh
@@ -211,18 +211,12 @@ class LipsyncService:
                        normalize_emotion(emotion),
                    )
                )
-            except Exception as exc:
-                # 投递失败时立即把 job 标成 failed 并写入 error_message，
-                # 前端轮询时能直接看到失败原因，不会无限卡在 tts_processing。
-                logger.exception(
-                    "Celery 任务提交失败，TTS 任务已创建但未触发执行: job_id=%s err=%s",
+            except Exception:
+                logger.warning(
+                    "Celery 任务提交失败，TTS 任务已创建但未触发执行: %s",
                    job_id,
-                    exc,
+                    exc_info=True,
                )
-                job.status = "failed"
-                job.error_message = f"Celery 任务投递失败: {exc}"
-                job.error_code = "AsyncDispatchFailed"
-                job.updated_at = datetime.now(timezone.utc)
        else:
            # 2b. 直接音频模式：同步签名并提交 MediaKit
            video_url = self._sign_media_url(video_url)
@@ -9,9 +9,6 @@
 5. 签名 URL 并提交到 MediaKit
 6. 更新 job 状态为 submitted
 7. 异常时标记 job 为 failed
-
-注意：使用 @shared_task 而非绑定到某个 celery_app 实例，
-确保任务能被 Worker 侧 celery_app 正确注册，同时 API 侧 send_task/apply_async 仍可正常调用。
 """

 import io
@@ -19,16 +16,16 @@ import logging
 from datetime import datetime, timezone
 from urllib.parse import urlparse

-from celery import shared_task
+from app.core.celery_app import celery_app

 logger = logging.getLogger(__name__)

-# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService._sign_media_url 保持一致
+# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService 保持一致
 _MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600


 def _sign_media_url(url: str) -> str:
-    """对自家 OSS 私有桶 URL 重签长有效期预签名.
+    """对自家 OSS 私有桶 URL 重签长有效期预签名（与 LipsyncService._sign_media_url 保持一致）.

    - 自家 OSS URL → 重签 7 天有效期
    - 外部临时 URL → 原样透传
@@ -54,7 +51,7 @@ def _sign_media_url(url: str) -> str:
        return url


-@shared_task(
+@celery_app.task(
    bind=True,
    name="lipsync_tts.synthesize_and_submit",
    max_retries=2,
@@ -76,21 +73,12 @@ def tts_synthesize_and_submit(
    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

-    # SessionLocal 获取：
-    # - API 容器：app.db.SessionLocal（环境变量完整，导入即建引擎）
-    # - Worker 容器：worker_app.db.SessionLocal（Worker 自己的 settings 初始化引擎）
-    # API 侧没有 worker_app 模块 → ImportError 直接回退；
-    # Worker 侧 app.db 会因缺少 API 专有环境变量抛 pydantic ValidationError，
-    # 此时也要回退到 worker_app.db。
-    try:
-        from worker_app.db import SessionLocal  # type: ignore
-    except Exception:  # noqa: BLE001
-        from app.db import SessionLocal  # type: ignore
-
    db: DBSession = SessionLocal()
    try:
        job = (
@@ -161,8 +149,6 @@ def tts_synthesize_and_submit(
                ),
                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")
@@ -178,7 +164,7 @@ def tts_synthesize_and_submit(

        db.commit()

-        # 3. 签名 URL 并提交到 MediaKit（复用模块内 _sign_media_url，避免对 LipsyncService 的耦合）
+        # 3. 签名 URL 并提交到 MediaKit
        audio_url = _sign_media_url(job.audio_url)
        video_url = _sign_media_url(job.video_url)

@@ -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
 }

@@ -34,15 +34,10 @@ 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",
-    # 注意：必须用 app.* 路径，不能用 apps.api.app.* 路径！
-    # PYTHONPATH=/app/apps/api 下，app.tasks.lipsync_tts 可直接导入且不触发 apps/api/__init__.py
-    # （apps/api/__init__.py 会 from .main import app，级联加载整个 FastAPI 栈，Worker 中不需要且会导致注册失败）
-    "app.tasks.lipsync_tts",
+    "apps.api.app.tasks.lipsync_tts",
 )

 # Celery Beat 定时任务调度
@@ -1,48 +0,0 @@
-"""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()
@@ -163,7 +163,7 @@ services:
    #   context: ../..
    #   dockerfile: ${WEB_DOCKERFILE:-infra/docker/web.Dockerfile}
    #   args:
-    #     (NGINX_CONF no longer needed - all configs baked into image)
+    #     NGINX_CONF: ${WEB_NGINX_CONF:-infra/docker/nginx.conf}
    
    container_name: xiaoxia-web-${ENV:-staging}
    restart: unless-stopped
@@ -178,12 +178,12 @@ services:
      - xiaoxia-net
    
    # =========================================
-    # Nginx 配置运行时覆盖（双保险：entrypoint 也按 APP_ENV 选择配置）
+    # Nginx 配置运行时覆盖
    # 确保容器使用正确环境的 nginx 配置，即使镜像构建时使用了默认配置
    # 注意: 只覆盖 /etc/nginx/conf.d/default.conf，不挂载 /usr/share/nginx/html
    # =========================================
    environment:
-      - APP_ENV=${ENV:-staging}
+      - NGINX_ENV=${ENV:-staging}
    volumes:
      - ./nginx-${ENV:-staging}.conf:/etc/nginx/conf.d/default.conf:ro
    
@@ -14,11 +14,11 @@ REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
 ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
 GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
 LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
-REPO_DIR="${REPO_DIR:-/var/lib/xiaoxia-saas-production/repo}"

 if [ -z "$IMAGE_TAG" ]; then
  echo "ERROR: IMAGE_TAG is required"
  exit 1
+fi

 test -f "$ENV_FILE"
 mkdir -p "$GENERATED_DIR"
@@ -31,6 +31,7 @@ if [ -n "$REGISTRY_TOKEN" ]; then
  printf %s "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || {
    echo "WARN: docker login failed, will try to pull anyway"
  }
+fi

 # ---- Pull 三镜像 ----
 REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
@@ -65,14 +66,17 @@ if docker inspect xiaoxia-web-production >/dev/null 2>&1; then
  if [ -d "$_tmpdir" ] && [ "$(ls -A "$_tmpdir" 2>/dev/null)" ]; then
    cp -an "$_tmpdir"/. "$LEGACY_ASSETS_DIR"/ 2>/dev/null || true
    echo "Legacy assets backed up: $(ls "$_tmpdir" | wc -l) files"
+  fi
  rm -rf "$_tmpdir"
 else
  echo "No existing web container, skipping legacy assets backup"
+fi

 # 清理超过 7 天的旧 assets 文件（避免无限增长）
 if [ -d "$LEGACY_ASSETS_DIR" ]; then
  find "$LEGACY_ASSETS_DIR" -type f -mtime +7 -delete 2>/dev/null || true
  echo "Legacy assets cleanup done (retain 7 days)"
+fi

 # ---- 确保基础设施容器在运行 ----
 echo "Checking infrastructure containers..."
@@ -80,10 +84,12 @@ for c in xiaoxia-postgres-production xiaoxia-redis-production; do
  if ! docker inspect "$c" >/dev/null 2>&1; then
    echo "ERROR: Required container not found: $c"
    exit 1
+  fi
  state=$(docker inspect -f '{{.State.Status}}' "$c")
  if [ "$state" != "running" ]; then
    echo "ERROR: Container not running: $c ($state)"
    exit 1
+  fi
 done

 # ---- 确保生产网络存在 ----
@@ -102,6 +108,7 @@ echo "Migrations completed."
 echo "Stopping old containers..."
 docker rm -f xiaoxia-api-production 2>/dev/null || true
 docker rm -f xiaoxia-worker-production 2>/dev/null || true
+docker rm -f xiaoxia-web-production 2>/dev/null || true

 # ---- 日志配置（所有容器共用） ----
 LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
@@ -159,16 +166,15 @@ docker run -d \
 # ---- 启动 Web ----
 # Legacy assets 挂载到 /usr/share/nginx/html/assets-legacy/assets/
 # nginx 配置中 assets location 有 fallback 逻辑
-WEB_VOLUMES=""
+LEGACY_VOLUME=""
 if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
-  WEB_VOLUMES="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+  LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
  echo "Web container: legacy assets mounted (fallback)"
 else
  echo "Web container: no legacy assets to mount"
 fi

 echo "Starting Web container..."
-docker rm -f xiaoxia-web-production 2>/dev/null || true
 docker run -d \
  --name xiaoxia-web-production \
  --network xiaoxia-net-production \
@@ -176,8 +182,7 @@ docker run -d \
  --restart unless-stopped \
  --cpus 0.5 \
  --memory 512m \
-  -e APP_ENV=production \
-  $WEB_VOLUMES \
+  $LEGACY_VOLUME \
  --health-cmd "wget --spider -q http://127.0.0.1:80" \
  --health-interval 30s \
  --health-timeout 5s \
@@ -192,6 +197,7 @@ while [ "$i" -lt 40 ]; do
  if curl -sf --max-time 5 http://127.0.0.1:8001/health >/dev/null 2>&1; then
    echo "API is healthy!"
    break
+  fi
  i=$((i + 1))
  echo "  Waiting... ($i/40)"
  sleep 3
@@ -201,6 +207,7 @@ if [ "$i" -ge 40 ]; then
  echo "ERROR: API did not become healthy within 120s"
  docker logs --tail 50 xiaoxia-api-production
  exit 1
+fi

 # ---- 等待 Web 健康 ----
 echo "Waiting for Web to become healthy..."
@@ -209,6 +216,7 @@ while [ "$i" -lt 15 ]; do
  if curl -sf --max-time 5 http://127.0.0.1:3002/ >/dev/null 2>&1; then
    echo "Web is healthy!"
    break
+  fi
  i=$((i + 1))
  echo "  Waiting... ($i/15)"
  sleep 2
@@ -218,6 +226,7 @@ if [ "$i" -ge 15 ]; then
  echo "ERROR: Web did not become healthy within 30s"
  docker logs --tail 30 xiaoxia-web-production
  exit 1
+fi

 # ---- 清理旧镜像 ----
 echo "Cleaning up old images..."
@@ -124,15 +124,23 @@ docker run -d \
  "$LOCAL_WORKER"

 # ---- 启动 Web ----
+# Web 镜像默认打包 production nginx.conf，staging 需要挂载 staging 配置
+NGINX_CONF="${NGINX_CONF:-${COMPOSE_DIR}/nginx-staging.conf}"
+if [ ! -f "$NGINX_CONF" ]; then
+  echo "WARN: nginx config not found at $NGINX_CONF, using image default"
+  NGINX_VOLUME=""
+else
+  NGINX_VOLUME="-v ${NGINX_CONF}:/etc/nginx/conf.d/default.conf:ro"
+fi
+
 echo "Starting Web container..."
-docker rm -f xiaoxia-web-staging 2>/dev/null || true
 docker run -d \
  --name xiaoxia-web-staging \
  --network xiaoxia-net-staging \
  -p 127.0.0.1:3001:80 \
  --restart unless-stopped \
  --label com.centurylinklabs.watchtower.enable=true \
-  -e APP_ENV=staging \
+  $NGINX_VOLUME \
  --health-cmd "wget --spider -q http://127.0.0.1:80" \
  --health-interval 30s \
  --health-timeout 5s \
@@ -53,6 +53,7 @@ export API_IMAGE="${API_IMAGE:-${REGISTRY}/xiaoxia-saas-api:dev}"
 export WORKER_IMAGE="${WORKER_IMAGE:-${REGISTRY}/xiaoxia-saas-worker:dev}"

 # Use staging-specific nginx config (proxy_pass → xiaoxia-api-staging:8000)
+export WEB_NGINX_CONF=infra/docker/nginx-staging.conf

 if [ "${REBUILD_BACKEND:-0}" = "1" ] || [ "${BUILD_WEB:-0}" = "1" ]; then
  if [ "${ALLOW_STAGING_BUILDS:-false}" != "true" ]; then
@@ -1,49 +0,0 @@
-#!/bin/sh
-# Select nginx config based on APP_ENV (staging/production).
-#
-# 两种运行模式：
-# 1. CI/CD 部署（staging/production）：部署脚本通过 `-v 宿主机文件:/etc/nginx/conf.d/default.conf:ro`
-#    把宿主机生成的带 resolver/docker upstream 的配置 bind mount 进来，entrypoint 不应改动。
-#    bind mount 的文件是 readonly 的，rm 会报 EBUSY ("Resource busy")，直接 exec nginx 即可。
-# 2. 本地 docker-compose / 直接 `docker run`（无外部挂载）：镜像烤入了 nginx-staging.conf 与
-#    nginx-production.conf 到 /etc/nginx/，entrypoint 根据 APP_ENV 把 default.conf 换成正确的 symlink。
-#
-# 策略：
-# - 如果 /etc/nginx/conf.d/default.conf 已经是指向目标 conf 的 symlink，什么都不做；
-# - 否则尝试 rm -f 再 ln -s；rm 失败说明是外部 bind mount（已有正确配置），不阻塞启动；
-# - 兜底：只要 conf.d 目录里有 .conf 文件（含 bind mount 来的），就直接启动 nginx。
-set -e
-
-NGINX_CONF_DIR="/etc/nginx/conf.d"
-TARGET_CONF=""
-
-case "${APP_ENV:-production}" in
-  staging)
-    TARGET_CONF="/etc/nginx/nginx-staging.conf"
-    ;;
-  *)
-    TARGET_CONF="/etc/nginx/nginx-production.conf"
-    ;;
-esac
-
-DEFAULT_CONF="$NGINX_CONF_DIR/default.conf"
-
-# 1. 已经是正确的 symlink：直接启动
-if [ -L "$DEFAULT_CONF" ] && [ "$(readlink "$DEFAULT_CONF" 2>/dev/null)" = "$TARGET_CONF" ]; then
-  exec nginx -g "daemon off;"
-fi
-
-# 2. 尝试替换为目标 symlink（无 bind mount 的场景）
-#    若 rm 失败（bind mount readonly，EBUSY/EPERM），则认为外部已注入配置，不阻塞。
-rm -f "$DEFAULT_CONF" 2>/dev/null || true
-if [ -f "$TARGET_CONF" ] && [ ! -e "$DEFAULT_CONF" ]; then
-  ln -s "$TARGET_CONF" "$DEFAULT_CONF" 2>/dev/null || true
-fi
-
-# 3. 兜底：至少要有一个 .conf 文件，否则 nginx 起不来
-if ! ls "$NGINX_CONF_DIR"/*.conf >/dev/null 2>&1; then
-  echo "ERROR: no nginx config found in $NGINX_CONF_DIR (tried $TARGET_CONF and external bind mount)" >&2
-  exit 1
-fi
-
-exec nginx -g "daemon off;"
@@ -1,11 +1,7 @@
 FROM git.xiaoxiajianji.com/xiaoxia/base/nginx:alpine AS runner
+ARG NGINX_CONF=infra/docker/nginx.conf
 WORKDIR /usr/share/nginx/html
 COPY apps/web/dist ./
-# 将所有 nginx 配置烤入镜像，entrypoint 按 APP_ENV 选择
-COPY infra/docker/nginx.conf /etc/nginx/nginx-production.conf
-COPY infra/docker/nginx-staging.conf /etc/nginx/nginx-staging.conf
-COPY infra/docker/nginx-production.conf /etc/nginx/nginx-production.conf
-COPY infra/docker/nginx-entrypoint.sh /docker-entrypoint.sh
-RUN chmod +x /docker-entrypoint.sh
+COPY ${NGINX_CONF} /etc/nginx/conf.d/default.conf
 EXPOSE 80
-ENTRYPOINT ["/docker-entrypoint.sh"]
+CMD ["nginx", "-g", "daemon off;"]
@@ -28,13 +28,9 @@ RUN --mount=type=cache,target=/app/apps/web/.tscache,sharing=locked \

 # Production stage with nginx
 FROM git.xiaoxiajianji.com/xiaoxia/base/nginx:alpine AS runner
+ARG NGINX_CONF=infra/docker/nginx.conf
 WORKDIR /usr/share/nginx/html
 COPY --from=builder /app/apps/web/dist ./
-# 将所有 nginx 配置烤入镜像，entrypoint 按 APP_ENV 选择
-COPY infra/docker/nginx.conf /etc/nginx/nginx-production.conf
-COPY infra/docker/nginx-staging.conf /etc/nginx/nginx-staging.conf
-COPY infra/docker/nginx-production.conf /etc/nginx/nginx-production.conf
-COPY infra/docker/nginx-entrypoint.sh /docker-entrypoint.sh
-RUN chmod +x /docker-entrypoint.sh
+COPY ${NGINX_CONF} /etc/nginx/conf.d/default.conf
 EXPOSE 80
-ENTRYPOINT ["/docker-entrypoint.sh"]
+CMD ["nginx", "-g", "daemon off;"]
@@ -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,15 @@ 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/
+COPY apps/api/app/config.py /app/apps/api/app/config.py
+COPY apps/api/app/core/ /app/apps/api/app/core/
+# API 侧 Celery 任务（lipsync_tts 等）在 worker 进程中执行，需复制任务文件、依赖及 __init__.py
+RUN mkdir -p /app/apps && touch /app/apps/__init__.py
+COPY apps/api/__init__.py /app/apps/api/__init__.py
+COPY apps/api/app/__init__.py /app/apps/api/app/__init__.py
+COPY apps/api/app/services/__init__.py /app/apps/api/app/services/__init__.py
+COPY apps/api/app/services/mediakit_client.py /app/apps/api/app/services/mediakit_client.py
+COPY apps/api/app/tasks/ /app/apps/api/app/tasks/

 # Worker 启动脚本
 COPY infra/docker/entrypoint-worker.sh /usr/local/bin/entrypoint-worker.sh
@@ -1 +0,0 @@
-# xiaoxia-saas shared packages namespace
@@ -1 +0,0 @@
-# adapter implementations namespace
@@ -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}")
@@ -143,8 +143,8 @@ class TestCreateJobAsyncTTS:
        assert args[2] == "v-1"  # voice_id
        assert args[3] == "测试文本"  # script_text

-    def test_tts_mode_celery_dispatch_failure_marks_job_failed(self):
-        """Celery dispatch 失败时，job 标为 failed 并写入 error_message，前端轮询能直接看到错误."""
+    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:
@@ -157,11 +157,9 @@ class TestCreateJobAsyncTTS:
                script_text="测试文本",
            )

-        # job 已创建且状态标为 failed
+        # job 已创建
        assert job is not None
-        assert job.status == "failed"
-        assert "Celery 任务投递失败" in job.error_message
-        assert job.error_code == "AsyncDispatchFailed"
+        assert job.status == "tts_processing"
        # MediaKit 未被调用
        client.submit_lipsync.assert_not_called()

@@ -284,3 +282,342 @@ class TestCancelJobTtsProcessing:

        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()
@@ -1,389 +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。
-    """
-    # SessionLocal 通过懒探测获取（Worker 用 worker_app.db，API 用 app.db），
-    # 测试环境里两个模块都能被真实导入，必须同时 mock 保证用的是 fake session。
-    fake_app_db = ModuleType("app.db")
-    fake_worker_db = ModuleType("worker_app.db")
-    session, factory = _build_session(job)
-    fake_app_db.SessionLocal = factory
-    fake_worker_db.SessionLocal = factory
-
-    patches = [
-        patch.dict(sys.modules, {"app.db": fake_app_db, "worker_app.db": fake_worker_db}),
-        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_app_db = ModuleType("app.db")
-        fake_worker_db = ModuleType("worker_app.db")
-        session, factory = _build_session(job)
-        fake_app_db.SessionLocal = factory
-        fake_worker_db.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,
-            {
-                "app.db": fake_app_db,
-                "worker_app.db": fake_worker_db,
-                "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()