debug(web): use docker run instead of docker create

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

.gitea/workflows/ci-pipeline.yml

@@ -196,7 +196,7 @@ jobs:
     - name: Run style checks
       shell: bash
       run: bash scripts/ci/validate_style.sh
-    - name: Auto-fix formatting (black + isort)
+    - name: Auto-fix formatting (black + isort + ruff)
       if: failure()
       shell: sh
       env:
@@ -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-web-diag.yml

@@ -0,0 +1,37 @@
+name: "Debug: Web container crash diag"
+
+on:
+  push:
+    branches:
+      - 'debug/web-crash-diag'
+  workflow_dispatch:
+
+jobs:
+  web-diag:
+    name: Web Crash Diagnostics
+    runs-on: runtime-builder
+    timeout-minutes: 10
+    steps:
+      - 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: Run web diag
+        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-web-diag.sh root@$H:/tmp/server-web-diag.sh
+          ssh -p $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@$H \
+            "bash /tmp/server-web-diag.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-web-diag.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+set +e
+echo "=== 1. Check base nginx image conf.d ==="
+docker run --rm --entrypoint sh git.xiaoxiajianji.com/xiaoxia/base/nginx:alpine -c \
+  "ls -la /etc/nginx/conf.d/ 2>&1; echo '---'; file /etc/nginx/conf.d/default.conf 2>&1; echo '---'; ls -la /etc/nginx/nginx*.conf 2>&1"
+echo ""
+echo "=== 2. Check staging web image (currently running) ==="
+WEB_IMG=$(docker inspect xiaoxia-web-staging --format '{{.Config.Image}}' 2>/dev/null)
+echo "Staging web image: $WEB_IMG"
+docker run --rm --entrypoint sh "$WEB_IMG" -c \
+  "ls -la /etc/nginx/conf.d/ 2>&1; echo '---'; cat /docker-entrypoint.sh 2>&1; echo '---'; ls -la /etc/nginx/nginx-*.conf 2>&1"
+echo ""
+echo "=== 3. Run entrypoint with APP_ENV=staging to reproduce ==="
+docker run --rm -e APP_ENV=staging --entrypoint sh "$WEB_IMG" -c \
+  "set -x; sh -x /docker-entrypoint.sh 2>&1 | head -30"
+echo ""
+echo "=== 4. What if we rm -f default.conf first then ln? ==="
+docker run --rm -e APP_ENV=staging --entrypoint sh "$WEB_IMG" -c \
+  "rm -f /etc/nginx/conf.d/default.conf && ln -sf /etc/nginx/nginx-staging.conf /etc/nginx/conf.d/default.conf && ls -la /etc/nginx/conf.d/default.conf && echo OK"
+echo ""
+echo "=== 5. Check running web logs for crash reason ==="
+docker logs xiaoxia-web-staging 2>&1 | tail -30
+echo ""
+echo "=== Done ==="

scripts/ci/auto_fix_formatting.py

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

tests/unit/test_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()