debug(lipsync): v3 - docker cp python scripts, check nginx/DB

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

.gitea/workflows/ci-pipeline.yml

@@ -827,9 +827,6 @@ 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=""
@@ -1026,9 +1023,6 @@ 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
@@ -1567,9 +1561,6 @@ 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=""

.gitea/workflows/debug-lipsync-diag.yml

@@ -0,0 +1,43 @@
+name: Debug Lipsync SSH Diag
+on:
+  workflow_dispatch:
+    inputs:
+      reason:
+        description: "debug reason"
+        required: false
+        default: "lipsync tts_processing stuck diagnosis"
+  push:
+    branches:
+      - debug/lipsync-ssh-diag
+permissions:
+  contents: read
+jobs:
+  ssh-diag:
+    name: SSH Staging Diagnostics
+    runs-on: runtime-builder
+    timeout-minutes: 15
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Setup SSH
+        shell: bash
+        env:
+          STAGING_SSH_KEY: ${{ secrets.PREVIEW_SSH_KEY }}
+        run: |
+          set -euo pipefail
+          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 "echo SSH_OK; hostname; date"
+      - name: Upload diag script and run
+        shell: bash
+        run: |
+          set -euo pipefail
+          H=47.98.113.167; P=22222
+          scp -P $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no \
+            infra/scripts/server-lipsync-diag-v3.sh root@$H:/tmp/server-lipsync-diag-v3.sh
+          ssh -p $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@$H \
+            "bash /tmp/server-lipsync-diag-v3.sh 2>&1"

apps/api/app/tasks/lipsync_tts.py

@@ -9,18 +9,52 @@
 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 apps.worker.worker_app.celery_app import celery_app
+from celery import shared_task
 
 logger = logging.getLogger(__name__)
 
+# MediaKit 预签名 URL 有效期（7天，秒），与 LipsyncService._sign_media_url 保持一致
+_MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600
 
-@celery_app.task(
+
+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,
@@ -45,7 +79,6 @@ def tts_synthesize_and_submit(
     from packages.adapters.sqlalchemy_impl.database import SessionLocal
     from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
     from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
-    from packages.shared.storage import get_shared_storage_service
     from packages.shared.url_security import safe_download_bytes
 
     db: DBSession = SessionLocal()
@@ -109,26 +142,35 @@ def tts_synthesize_and_submit(
             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"),
+                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)
+            logger.warning(
+                "[lipsync_tts] TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s",
+                job_id,
+                exc,
+            )
             job.audio_url = temp_url
 
         db.commit()
 
-        # 3. 签名 URL 并提交到 MediaKit
-        from app.services.lipsync_service import LipsyncService
-
-        temp_service = LipsyncService.__new__(LipsyncService)
-        audio_url = temp_service._sign_media_url(job.audio_url)
-        video_url = temp_service._sign_media_url(job.video_url)
+        # 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:
@@ -141,7 +183,11 @@ def tts_synthesize_and_submit(
             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"])
+            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)

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}`)
+  const response = await apiClient.get<LipsyncJob>(`/lipsync/jobs/${id}`, { timeout: 60000 })
   return response.data
 }
 

apps/worker/worker_app/celery_app.py

@@ -37,7 +37,10 @@ celery_app.conf.imports = (
     "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",
     "worker_app.tasks.cleanup",
-    "apps.api.app.tasks.lipsync_tts",
+    # 注意：必须用 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",
 )
 
 # Celery Beat 定时任务调度

infra/docker/compose.yml

@@ -163,7 +163,7 @@ services:
     #   context: ../..
     #   dockerfile: ${WEB_DOCKERFILE:-infra/docker/web.Dockerfile}
     #   args:
-    #     NGINX_CONF: ${WEB_NGINX_CONF:-infra/docker/nginx.conf}
+    #     (NGINX_CONF no longer needed - all configs baked into image)
     
     container_name: xiaoxia-web-${ENV:-staging}
     restart: unless-stopped
@@ -178,12 +178,12 @@ services:
       - xiaoxia-net
     
     # =========================================
-    # Nginx 配置运行时覆盖
+    # Nginx 配置运行时覆盖（双保险：entrypoint 也按 APP_ENV 选择配置）
     # 确保容器使用正确环境的 nginx 配置，即使镜像构建时使用了默认配置
     # 注意: 只覆盖 /etc/nginx/conf.d/default.conf，不挂载 /usr/share/nginx/html
     # =========================================
     environment:
-      - NGINX_ENV=${ENV:-staging}
+      - APP_ENV=${ENV:-staging}
     volumes:
       - ./nginx-${ENV:-staging}.conf:/etc/nginx/conf.d/default.conf:ro
     

infra/docker/deploy-production-registry.sh

@@ -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,7 +31,6 @@ 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}"
@@ -66,17 +65,14 @@ 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..."
@@ -84,12 +80,10 @@ 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
 
 # ---- 确保生产网络存在 ----
@@ -108,7 +102,6 @@ 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"
@@ -166,15 +159,16 @@ docker run -d \
 # ---- 启动 Web ----
 # Legacy assets 挂载到 /usr/share/nginx/html/assets-legacy/assets/
 # nginx 配置中 assets location 有 fallback 逻辑
-LEGACY_VOLUME=""
+WEB_VOLUMES=""
 if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
-  LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+  WEB_VOLUMES="-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 \
@@ -182,7 +176,8 @@ docker run -d \
   --restart unless-stopped \
   --cpus 0.5 \
   --memory 512m \
-  $LEGACY_VOLUME \
+  -e APP_ENV=production \
+  $WEB_VOLUMES \
   --health-cmd "wget --spider -q http://127.0.0.1:80" \
   --health-interval 30s \
   --health-timeout 5s \
@@ -197,7 +192,6 @@ 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
@@ -207,7 +201,6 @@ 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..."
@@ -216,7 +209,6 @@ 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
@@ -226,7 +218,6 @@ 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..."

infra/docker/deploy-staging-registry.sh

@@ -124,23 +124,15 @@ 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 \
-  $NGINX_VOLUME \
+  -e APP_ENV=staging \
   --health-cmd "wget --spider -q http://127.0.0.1:80" \
   --health-interval 30s \
   --health-timeout 5s \

infra/docker/deploy-staging.sh

@@ -53,7 +53,6 @@ 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

infra/docker/nginx-entrypoint.sh

@@ -0,0 +1,18 @@
+#!/bin/sh
+# Select nginx config based on APP_ENV (staging/production).
+# Both configs are baked into the image at well-known paths.
+# nginx reads config only at startup, so symlink before exec.
+set -e
+
+NGINX_CONF_DIR="/etc/nginx/conf.d"
+
+case "${APP_ENV:-production}" in
+  staging)
+    ln -sf /etc/nginx/nginx-staging.conf "$NGINX_CONF_DIR/default.conf"
+    ;;
+  *)
+    ln -sf /etc/nginx/nginx-production.conf "$NGINX_CONF_DIR/default.conf"
+    ;;
+esac
+
+exec nginx -g "daemon off;"

infra/docker/web-artifact.Dockerfile

@@ -1,7 +1,11 @@
 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 ./
-COPY ${NGINX_CONF} /etc/nginx/conf.d/default.conf
+# 将所有 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
 EXPOSE 80
-CMD ["nginx", "-g", "daemon off;"]
+ENTRYPOINT ["/docker-entrypoint.sh"]

infra/docker/web.Dockerfile

@@ -28,9 +28,13 @@ 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 ./
-COPY ${NGINX_CONF} /etc/nginx/conf.d/default.conf
+# 将所有 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
 EXPOSE 80
-CMD ["nginx", "-g", "daemon off;"]
+ENTRYPOINT ["/docker-entrypoint.sh"]

infra/docker/worker.Dockerfile

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

infra/scripts/server-lipsync-diag-v2.sh

@@ -0,0 +1,152 @@
+#!/bin/bash
+# Lipsync 深度诊断脚本 v2 - 重点排查broker连通/消息投递/DB状态
+set +e
+echo "#########################################################"
+echo "#  Lipsync 深度诊断 v2"
+echo "#  Date: $(date)"
+echo "#########################################################"
+
+W=$(docker ps --format '{{.Names}}' | grep -E 'worker' | head -1)
+A=$(docker ps --format '{{.Names}}' | grep -E 'api' | grep -v 'web' | head -1)
+R=$(docker ps --format '{{.Names}}' | grep -E 'redis' | head -1)
+echo "Containers: W=$W  A=$A  R=$R"
+
+echo ""
+echo "=== 1. 容器 uptime + image tag ==="
+for c in $W $A; do
+  echo "--- $c ---"
+  docker inspect "$c" --format 'Image={{.Config.Image}} Created={{.Created}} Started={{.State.StartedAt}} Restarts={{.RestartCount}}'
+done
+
+echo ""
+echo "=== 2. API+Worker 的 BROKER/REDIS 地址（隐藏密码） ==="
+echo "--- API BROKER ---"
+docker exec "$A" env | grep -iE "broker|celery_broker|redis_url|backend" | sed -E 's|(:)?//[^:]+:([^@]+)@|\1//***:***@|g'
+echo "--- Worker BROKER ---"
+docker exec "$W" env | grep -iE "broker|celery_broker|redis_url|backend" | sed -E 's|(:)?//[^:]+:([^@]+)@|\1//***:***@|g'
+
+echo ""
+echo "=== 3. API容器 完整日志（近1000行） lipsync/TTS 相关 ==="
+docker logs --tail=1000 "$A" 2>&1 | grep -iE "lipsync|tts_synth|tts_processing|celery.*task|apply_async|NotRegistered|OperationalError|ConnectionError|error.*submit|traceback|submitted" | tail -200
+
+echo ""
+echo "=== 4. API 容器最近的 ERROR/Exception ==="
+docker logs --tail=2000 "$A" 2>&1 | grep -iE "error|exception|traceback|critical" | grep -v "health\|/health" | tail -80
+
+echo ""
+echo "=== 5. 测试1：在API容器内实际投递一条测试消息，看Worker是否消费 ==="
+# 投递后立即检查队列和Worker日志
+docker exec "$A" python - <<'PYEOF'
+import sys, time, traceback
+try:
+    from app.tasks.lipsync_tts import tts_synthesize_and_submit
+    from app.core.celery_app import celery_app as api_app
+    # 检查broker连接
+    conn = api_app.connection()
+    conn.ensure_connection(max_retries=2)
+    print("API celery broker connected:", conn.as_uri())
+    # 投递到celery默认队列
+    result = tts_synthesize_and_submit.apply_async(
+        args=["diag-test-job-id", "diag-user-id", "diag-voice", "diagnostic script text", 1.0, "neutral"],
+        queue="celery",
+    )
+    print("APPLY_ASYNC_OK task_id:", result.id)
+    print("task name:", result.name)
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+echo ""
+sleep 3
+echo "--- After apply_async: queue lengths ---"
+for q in celery generation transcode; do
+    echo "  $q: $(docker exec "$R" redis-cli LLEN $q)"
+done
+
+echo ""
+echo "=== 6. 等8秒看Worker是否收到并消费 ==="
+sleep 8
+echo "--- Queue lengths after 8s ---"
+for q in celery generation transcode; do
+    echo "  $q: $(docker exec "$R" redis-cli LLEN $q)"
+done
+
+echo ""
+echo "=== 7. Worker 日志最新记录（查看是否消费了测试消息） ==="
+docker logs --since=30s "$W" 2>&1 | tail -60
+
+echo ""
+echo "=== 8. Worker 内 Python 直接连接broker测试 ==="
+docker exec "$W" python - <<'PYEOF'
+import traceback
+try:
+    from worker_app.celery_app import celery_app
+    conn = celery_app.connection()
+    conn.ensure_connection(max_retries=2)
+    print("Worker celery broker connected:", conn.as_uri())
+    insp = celery_app.control.inspect(timeout=3)
+    reg = insp.registered()
+    print("Registered from worker inspect (via broker):")
+    for node, tasks in (reg or {}).items():
+        has_l = any('lipsync' in t for t in tasks)
+        print(f"  {node}: {len(tasks)} tasks, lipsync registered: {has_l}")
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+
+echo ""
+echo "=== 9. DB 查询最近10条lipsync_jobs状态 ==="
+docker exec "$A" python - <<'PYEOF' 2>&1
+import os, traceback
+db_url = os.environ.get("DATABASE_URL","")
+print("DATABASE_URL prefix:", (db_url[:50]+"...") if db_url else "(empty)")
+try:
+    # 通过现有代码路径
+    from app.db.session import SessionLocal
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+    db = SessionLocal()
+    jobs = db.query(LipsyncJobModel).order_by(LipsyncJobModel.created_at.desc()).limit(10).all()
+    print(f"Found {len(jobs)} recent lipsync jobs:")
+    for j in jobs:
+        err = getattr(j, 'error_message', '') or ''
+        t_id = getattr(j, 'celery_task_id', '') or ''
+        print(f"  id={j.id} status={j.status} mode={getattr(j,'mode','?')} "
+              f"created={j.created_at} celery_task_id={t_id} "
+              f"error={(err[:120]+'...') if len(err)>120 else err!r}")
+    db.close()
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+
+echo ""
+echo "=== 10. Worker 启动完整日志（前60行）==="
+docker logs "$W" 2>&1 | head -60
+
+echo ""
+echo "=== 11. lipsync_service.py 完整 create_job 相关段 ==="
+docker exec "$A" sed -n '170,240p' /app/apps/api/app/services/lipsync_service.py 2>&1
+
+echo ""
+echo "=== 12. celery_app task_routes / queue 配置（API侧）==="
+docker exec "$A" python - <<'PYEOF'
+from app.core.celery_app import celery_app
+print("task_routes:", getattr(celery_app.conf, 'task_routes', None))
+print("task_default_queue:", celery_app.conf.task_default_queue)
+print("task_queues:", celery_app.conf.task_queues)
+print("broker_url prefix:", celery_app.conf.broker_url[:60])
+print("result_backend prefix:", str(celery_app.conf.result_backend)[:60] if celery_app.conf.result_backend else None)
+PYEOF
+
+echo ""
+echo "=== 13. Worker celery_app queue 配置 ==="
+docker exec "$W" python - <<'PYEOF'
+from worker_app.celery_app import celery_app
+print("task_routes:", getattr(celery_app.conf, 'task_routes', None))
+print("task_default_queue:", celery_app.conf.task_default_queue)
+print("task_queues:", celery_app.conf.task_queues)
+print("include/imports count:", len(celery_app.conf.imports))
+PYEOF
+
+echo ""
+echo "#########################################################"
+echo "#  v2 诊断完成"
+echo "#########################################################"

infra/scripts/server-lipsync-diag-v3.sh

@@ -0,0 +1,162 @@
+#!/bin/bash
+# Lipsync 诊断脚本 v3 - 用独立文件避免heredoc问题
+set +e
+echo "#########################################################"
+echo "#  Lipsync 诊断 v3"
+echo "#  Date: $(date)"
+echo "#########################################################"
+
+W=$(docker ps --format '{{.Names}}' | grep -E 'worker' | head -1)
+A=$(docker ps --format '{{.Names}}' | grep -E 'api' | grep -v 'web' | head -1)
+R=$(docker ps --format '{{.Names}}' | grep -E 'redis' | head -1)
+NGINX=$(docker ps --format '{{.Names}}' | grep -iE 'nginx|web' | head -1)
+echo "Containers:"
+echo "  W=$W"
+echo "  A=$A"
+echo "  R=$R"
+echo "  NGINX=$NGINX"
+
+echo ""
+echo "=== A1. API 容器近 2000 行日志全部（看请求是否进来） ==="
+docker logs --tail=2000 "$A" 2>&1 | tail -300
+
+echo ""
+echo "=== A2. API容器启动命令 & 网络 ==="
+docker inspect "$A" --format 'Cmd={{.Config.Cmd}} Entrypoint={{.Config.Entrypoint}} NetworkMode={{.HostConfig.NetworkMode}}'
+docker inspect "$A" --format '{{range .NetworkSettings.Networks}}{{.NetworkID}} {{.IPAddress}}{{end}}'
+
+echo ""
+echo "=== A3. Nginx 容器（如果有）日志 ==="
+if [ -n "$NGINX" ]; then
+  docker logs --tail=200 "$NGINX" 2>&1 | grep -iE "lipsync|ai-avatar|error|upstream" | tail -100
+  echo "--- nginx config ---"
+  docker exec "$NGINX" cat /etc/nginx/conf.d/default.conf 2>&1 | head -80
+fi
+
+echo ""
+echo "=== B1. 写测试脚本到/tmp，再docker cp到容器里执行（避免heredoc问题） ==="
+cat > /tmp/api_diag.py << 'PYEOF'
+import sys, traceback, os
+print("=== PYTHON OK in API container, cwd=", os.getcwd())
+print("sys.path[0:3]:", sys.path[:3])
+print("PYTHONPATH:", os.environ.get("PYTHONPATH",""))
+try:
+    from app.tasks.lipsync_tts import tts_synthesize_and_submit
+    from app.core.celery_app import celery_app as api_app
+    print("api_app.main:", api_app.main)
+    print("api_app.conf.broker_url:", api_app.conf.broker_url)
+    print("task name:", tts_synthesize_and_submit.name)
+    print("task.app.main:", tts_synthesize_and_submit.app.main)
+    print("task in api_app.tasks:", tts_synthesize_and_submit.name in api_app.tasks)
+    print("api_app.conf.task_routes:", api_app.conf.task_routes)
+    print("api_app.conf.task_default_queue:", api_app.conf.task_default_queue)
+    # 测试broker连接
+    conn = api_app.connection()
+    conn.ensure_connection(max_retries=2)
+    print("broker connected OK:", conn.as_uri())
+    # 投递测试任务
+    result = tts_synthesize_and_submit.apply_async(
+        args=["diag-v3-job", "diag-user", "diag-voice", "hello v3 diag", 1.0, "neutral"],
+        queue="celery",
+    )
+    print("APPLY_ASYNC_OK id=", result.id, "name=", result.name, "queue=celery")
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+docker cp /tmp/api_diag.py "$A:/tmp/api_diag.py"
+echo "--- docker exec python /tmp/api_diag.py in API ---"
+docker exec "$A" python /tmp/api_diag.py 2>&1
+
+echo ""
+echo "=== B2. 投递3秒后查队列 ==="
+sleep 3
+for q in celery generation transcode; do
+    echo "  queue $q: $(docker exec "$R" redis-cli LLEN $q)"
+done
+echo "--- peek celery ---"
+docker exec "$R" redis-cli LRANGE celery 0 2 2>&1 | head -20
+
+echo ""
+echo "=== B3. 等10秒再查队列和Worker日志 ==="
+sleep 7
+for q in celery generation transcode; do
+    echo "  queue $q: $(docker exec "$R" redis-cli LLEN $q)"
+done
+echo "--- Worker logs since 20s ago ---"
+docker logs --since=20s "$W" 2>&1 | tail -50
+
+echo ""
+echo "=== C1. Worker侧执行诊断脚本 ==="
+cat > /tmp/worker_diag.py << 'PYEOF'
+import sys, os, traceback
+print("=== PYTHON OK in Worker container ===")
+try:
+    from worker_app.celery_app import celery_app
+    print("worker celery_app.main:", celery_app.main)
+    print("broker_url:", celery_app.conf.broker_url)
+    print("task_default_queue:", celery_app.conf.task_default_queue)
+    print("task_routes:", celery_app.conf.task_routes)
+    print("task_queues:", [(q.name, [b.name for b in q.bindings]) for q in (celery_app.conf.task_queues or [])])
+    # inspect via broker
+    insp = celery_app.control.inspect(timeout=3)
+    act = insp.active() or {}
+    reg = insp.registered() or {}
+    res = insp.reserved() or {}
+    for node, tasks in act.items():
+        print(f"ACTIVE on {node}: {len(tasks)} tasks")
+        for t in tasks:
+            print(f"  - {t.get('name')} args={str(t.get('args',''))[:60]}")
+    for node, tasks in res.items():
+        print(f"RESERVED on {node}: {len(tasks)} tasks")
+        for t in tasks:
+            print(f"  - {t.get('name')}")
+    print("registered lipsync count per node:")
+    for node, tasks in reg.items():
+        has = [t for t in tasks if 'lipsync' in t]
+        print(f"  {node}: total={len(tasks)}, lipsync={has}")
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+docker cp /tmp/worker_diag.py "$W:/tmp/worker_diag.py"
+docker exec "$W" python /tmp/worker_diag.py 2>&1
+
+echo ""
+echo "=== D1. DB最近20条lipsync jobs ==="
+cat > /tmp/db_diag.py << 'PYEOF'
+import sys, os, traceback
+print("=== DB diag in API container ===")
+print("DATABASE_URL prefix:", os.environ.get("DATABASE_URL","")[:60])
+try:
+    from app.db.session import SessionLocal
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+    from datetime import datetime, timezone
+    db = SessionLocal()
+    jobs = db.query(LipsyncJobModel).order_by(LipsyncJobModel.created_at.desc()).limit(20).all()
+    print(f"Found {len(jobs)} recent lipsync jobs:")
+    for j in jobs:
+        err = getattr(j, 'error_message', None) or ''
+        t_id = getattr(j, 'celery_task_id', None) or ''
+        c_at = j.created_at.isoformat() if j.created_at else '?'
+        print(f"  id={j.id} status={j.status} mode={'tts' if (j.voice_id and not j.audio_url) else 'audio'} "
+              f"created={c_at} celery_task_id={t_id} err={(err[:100]+'...') if len(err)>100 else err!r}")
+    db.close()
+except Exception as e:
+    traceback.print_exc()
+PYEOF
+docker cp /tmp/db_diag.py "$A:/tmp/db_diag.py"
+docker exec "$A" python /tmp/db_diag.py 2>&1
+
+echo ""
+echo "=== D2. API容器监听端口 ==="
+docker exec "$A" sh -c "ss -tlnp 2>/dev/null || netstat -tlnp 2>/dev/null || cat /proc/net/tcp | head" 2>&1 | head -20
+
+echo ""
+echo "=== E1. 外部HTTPS健康检查 ==="
+curl -sk --max-time 8 https://staging-api.xiaoxiajianji.com/api/v1/health 2>&1 | head -3
+echo ""
+curl -sk --max-time 8 https://staging-api.xiaoxiajianji.com/api/v1/lipsync/jobs 2>&1 | head -3
+
+echo ""
+echo "#########################################################"
+echo "#  v3 诊断完成"
+echo "#########################################################"

infra/scripts/server-lipsync-diag.sh

@@ -0,0 +1,117 @@
+#!/bin/bash
+# Lipsync staging 诊断脚本（在staging服务器本机执行）
+set +e
+echo "#########################################################"
+echo "#  Lipsync TTS async 诊断报告"
+echo "#  Host: $(hostname)"
+echo "#  Date: $(date)"
+echo "#########################################################"
+
+echo ""
+echo "=== 1. docker ps (worker/api/redis) ==="
+docker ps --format 'table {{.Names}}\t{{.Image}}\t{{.Status}}\t{{.CreatedAt}}' | grep -E "worker|api|redis|nginx|NAMES"
+
+W=$(docker ps --format '{{.Names}}' | grep -E 'worker' | head -1)
+A=$(docker ps --format '{{.Names}}' | grep -E 'api' | grep -v 'web' | head -1)
+R=$(docker ps --format '{{.Names}}' | grep -E 'redis' | head -1)
+
+echo ""
+echo "=== Worker container: $W ==="
+echo "=== API container: $A ==="
+echo "=== Redis container: $R ==="
+
+echo ""
+echo "=== 2. Worker container image + created time ==="
+docker inspect "$W" --format 'Image={{.Config.Image}} Created={{.Created}}'
+docker inspect "$A" --format 'Image={{.Config.Image}} Created={{.Created}}'
+
+echo ""
+echo "=== 3. Worker 启动日志：celery ready / registered / lipsync 相关 ==="
+docker logs --tail=500 "$W" 2>&1 | grep -iE "lipsync|tts_synth|synthesize|celery@|ready|registered|register|error|traceback|import|critical|not registered|Consumer|mingle|tasks" | tail -120
+
+echo ""
+echo "=== 4. celery -A inspect registered ==="
+docker exec "$W" celery -A worker_app.celery_app inspect registered 2>&1 | tail -80
+
+echo ""
+echo "=== 5. celery -A inspect active/reserved ==="
+echo "--- active ---"
+docker exec "$W" celery -A worker_app.celery_app inspect active 2>&1 | tail -30
+echo "--- reserved ---"
+docker exec "$W" celery -A worker_app.celery_app inspect reserved 2>&1 | tail -30
+
+echo ""
+echo "=== 6. Worker 内 Python: import lipsync_tts task，看绑定 ==="
+docker exec "$W" python -c "
+from worker_app.celery_app import celery_app
+print('worker celery_app.main:', celery_app.main)
+print('worker celery_app.conf.imports:')
+for m in celery_app.conf.imports:
+    print('  ', m)
+import app.tasks.lipsync_tts as m
+t = m.tts_synthesize_and_submit
+print('task name:', t.name)
+print('task.app.main:', t.app.main)
+print('task.app is worker celery_app:', t.app is celery_app)
+print('task.name in worker celery_app.tasks:', t.name in celery_app.tasks)
+" 2>&1
+
+echo ""
+echo "=== 7. API 容器日志：lipsync/celery/apply_async 相关错误 ==="
+docker logs --tail=800 "$A" 2>&1 | grep -iE "lipsync|tts_|celery|apply_async|NotRegistered|traceback|error.*task" | tail -120
+
+echo ""
+echo "=== 8. API 内 Python: import task 看绑定 + apply_async 试投递（不消费） ==="
+docker exec "$A" python -c "
+import traceback
+try:
+    from app.tasks.lipsync_tts import tts_synthesize_and_submit
+    from app.core.celery_app import celery_app as api_app
+    print('api celery_app.main:', api_app.main)
+    print('task name:', tts_synthesize_and_submit.name)
+    print('task.app.main:', tts_synthesize_and_submit.app.main)
+    print('task.app is api_app:', tts_synthesize_and_submit.app is api_app)
+    print('task.name in api_app.tasks:', tts_synthesize_and_submit.name in api_app.tasks)
+    # 测试 send_task 是否能路由（不发送真任务）
+    print('api_app tasks includes lipsync_tts.synthesize_and_submit:', 'lipsync_tts.synthesize_and_submit' in api_app.tasks)
+except Exception as e:
+    traceback.print_exc()
+" 2>&1
+
+echo ""
+echo "=== 9. Redis 队列长度 ==="
+for q in celery generation transcode; do
+    len=$(docker exec "$R" redis-cli LLEN $q 2>&1)
+    echo "  $q length: $len"
+done
+
+echo ""
+echo "=== 10. Redis celery 队列 peek（最多5条） ==="
+docker exec "$R" redis-cli LRANGE celery 0 4 2>&1 | head -40
+
+echo ""
+echo "=== 11. Worker 关键文件内容校验 ==="
+echo "--- worker_app/celery_app.py imports 段 ---"
+docker exec "$W" grep -n "lipsync\|imports\s*=\|app.tasks\|apps.api" /app/apps/worker/worker_app/celery_app.py 2>&1
+echo ""
+echo "--- lipsync_tts.py 前20行（应该是 shared_task） ---"
+docker exec "$W" head -25 /app/apps/api/app/tasks/lipsync_tts.py 2>&1
+
+echo ""
+echo "=== 12. API lipsync_service.py apply_async 上下文 ==="
+docker exec "$A" grep -n -B3 -A10 "apply_async\|tts_synthesize" /app/apps/api/app/services/lipsync_service.py 2>&1 | head -80
+
+echo ""
+echo "=== 13. Worker /app 目录结构（app/tasks） ==="
+docker exec "$W" ls -la /app/apps/api/app/tasks/ 2>&1 | head -30
+echo ""
+echo "--- /app/apps/api/app/__init__.py? ---"
+docker exec "$W" ls -la /app/apps/api/app/__init__.py /app/apps/api/app/core/__init__.py /app/apps/api/app/api/__init__.py 2>&1
+echo ""
+echo "--- PYTHONPATH inside worker ---"
+docker exec "$W" env | grep PYTHONPATH
+
+echo ""
+echo "#########################################################"
+echo "#  诊断完成"
+echo "#########################################################"

tests/unit/test_lipsync_speed_optimization.py

@@ -282,342 +282,3 @@ 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()

tests/unit/test_lipsync_tts.py

@@ -0,0 +1,382 @@
+"""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()