fix(ai-avatar): 素材预览不叠加标题 + 封面接口传title_config + 渲染后封面避免双标题

1. PanelVideoSelector: 移除素材视频上的标题叠加预览，标题只在对口型预览和最终成片上展示 2. generateSmartCover API: 传入title_config参数，让后端在对口型视频帧上用drawtext叠加标题 3. AiAvatarPage: 调用generateSmartCover时传入state.titleConfig；移除传给PanelVideoSelector的titleConfig prop 4. ai_avatar_render_service: 渲染完成后自动抽封面时不传title_config，避免最终视频已有标题再叠加导致双重标题
chore(ai-avatar): generateSmartCover 接口超时 60s → 120s (#1861 )
2026-09-11 18:29:23 +08:00 · 2026-09-11 15:58:19 +08:00 · 2026-09-11 15:52:55 +08:00 · 2026-09-11 15:12:55 +08:00 · 2026-09-11 15:00:22 +08:00 · 2026-09-11 06:27:13 +00:00
31 changed files with 1044 additions and 516 deletions
@@ -196,7 +196,7 @@ jobs:
    - name: Run style checks
      shell: bash
      run: bash scripts/ci/validate_style.sh
-    - name: Auto-fix formatting (black + isort)
+    - 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=""
@@ -0,0 +1,60 @@
+name: "Debug: Web container v2 (mount conflict)"
+on:
+  push:
+    branches: [debug/web-crash-v2]
+  workflow_dispatch:
+jobs:
+  web-diag:
+    runs-on: runtime-builder
+    timeout-minutes: 10
+    steps:
+      - name: Setup SSH and diagnose
+        shell: bash
+        env:
+          STAGING_SSH_KEY: ${{ secrets.PREVIEW_SSH_KEY }}
+        run: |
+          set -x
+          which ssh || (apt-get update -qq && apt-get install -y -qq openssh-client)
+          mkdir -p ~/.ssh && chmod 700 ~/.ssh
+          printf "%s" "$STAGING_SSH_KEY" > ~/.ssh/id_rsa
+          chmod 600 ~/.ssh/id_rsa
+          H=47.98.113.167; P=22222
+          ssh-keyscan -p $P -H $H >> ~/.ssh/known_hosts 2>/dev/null
+          ssh -p $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@$H 'bash -s' <<'REMOTE'
+          set -x
+          echo "=== Current staging containers ==="
+          docker ps -a --filter name=xiaoxia-*-staging --format "table {{.Names}}\t{{.Status}}\t{{.Image}}"
+          echo ""
+          echo "=== Web container logs (current/current-rolledback) ==="
+          docker logs xiaoxia-web-staging 2>&1 | tail -40
+          echo ""
+          echo "=== Web inspect: env & mounts ==="
+          docker inspect xiaoxia-web-staging --format 'Entrypoint: {{.Config.Entrypoint}} Cmd: {{.Config.Cmd}}'
+          docker inspect xiaoxia-web-staging --format '{{range .Config.Env}}{{.}}{{"\n"}}{{end}}' | grep -E "APP_ENV|VERSION"
+          echo "Mounts:"
+          docker inspect xiaoxia-web-staging --format '{{range .Mounts}}{{.Type}} {{.Source}} -> {{.Destination}} (rw={{.RW}}){{"\n"}}{{end}}'
+          echo ""
+          echo "=== Reproduce: rm on read-only bind mount ==="
+          docker run --rm --name nginx-ro-test \
+            -v /var/lib/xiaoxia-saas-staging/nginx-staging.conf:/etc/nginx/conf.d/default.conf:ro \
+            git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/xiaoxia-saas-web:387514c \
+            sh -c '
+              set -x
+              echo "Before:"
+              ls -la /etc/nginx/conf.d/
+              echo "Try rm (as entrypoint does):"
+              rm -f /etc/nginx/conf.d/default.conf
+              echo "rm exitcode=$?"
+              echo "After rm:"
+              ls -la /etc/nginx/conf.d/
+              echo "Test ln:"
+              ln -s /etc/nginx/nginx-staging.conf /etc/nginx/conf.d/default.conf
+              echo "ln exitcode=$?"
+              ls -la /etc/nginx/conf.d/
+              echo "nginx -t:"
+              nginx -t 2>&1
+            ' 2>&1
+          echo ""
+          echo "=== Also test with NEW fixed image (9c0d4b1 if present) ==="
+          docker images | grep xiaoxia-saas-web | head -5
+REMOTE
@@ -11,6 +11,7 @@
 from __future__ import annotations

 import logging
+from datetime import datetime, timezone

 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_db_session
@@ -77,10 +78,16 @@ def create_render_job(
        from app.tasks.ai_avatar_render import execute_ai_avatar_render

        execute_ai_avatar_render.delay(job.id)
-    except Exception:
-        logger.warning("Celery 任务提交失败，渲染任务已创建但未触发执行: %s", job.id)
+    except Exception as exc:
+        logger.exception("Celery 任务投递失败（创建）: job_id=%s err=%s", job.id, exc)
+        job.status = "failed"
+        job.error_message = f"任务提交失败：{exc}"
+        job.updated_at = datetime.now(timezone.utc)
+        svc.db.commit()
+        svc.db.refresh(job)
+        return AiAvatarRenderJobResponse.model_validate(job)

-    return job
+    return AiAvatarRenderJobResponse.model_validate(job)


 # ── GET /jobs — 任务列表 ─────────────────────────────────────────────────
@@ -172,10 +179,16 @@ def retry_render_job(
        from app.tasks.ai_avatar_render import execute_ai_avatar_render

        execute_ai_avatar_render.delay(job.id)
-    except Exception:
-        logger.warning("Celery 任务提交失败，重试任务已重置但未触发执行: %s", job.id)
+    except Exception as exc:
+        logger.exception("Celery 任务投递失败（重试）: job_id=%s err=%s", job.id, exc)
+        job.status = "failed"
+        job.error_message = f"任务提交失败：{exc}"
+        job.updated_at = datetime.now(timezone.utc)
+        svc.db.commit()
+        svc.db.refresh(job)
+        return AiAvatarRenderJobResponse.model_validate(job)

-    return job
+    return AiAvatarRenderJobResponse.model_validate(job)



@@ -199,7 +212,11 @@ def generate_avatar_smart_cover(
        raise HTTPException(status_code=400, detail="video_url 必须是合法的 HTTP/HTTPS URL")

    try:
-        cover_url = generate_smart_cover(video_url, max_frames=body.max_frames)
+        cover_url = generate_smart_cover(
+            video_url,
+            max_frames=body.max_frames,
+            title_config=getattr(body, "title_config", None),
+        )
    except Exception as exc:
        logger.error(
            "智能封面生成异常: user=%s video_url=%s err=%s",
@@ -110,10 +110,13 @@ class AiAvatarRenderProgressResponse(BaseModel):


 class SmartCoverRequest(BaseModel):
-    """智能封面请求 — MediaKit 抽帧 + 质量评分选最佳帧."""
+    """智能封面请求 — MediaKit 抽帧 + 质量评分选最佳帧 + 可选标题 drawtext 叠加."""

    video_url: str = Field(..., description="数字人视频 URL（对口型/渲染成片）")
    max_frames: int = Field(5, ge=1, le=10, description="抽帧数量（默认 5）")
+    title_config: Optional[dict[str, Any]] = Field(
+        None, description="标题配置；传入时在封面上用 drawtext 叠加标题（竖屏 720x1280）"
+    )


 class SmartCoverResponse(BaseModel):
@@ -11,6 +11,8 @@
 from __future__ import annotations

 import logging
+import os
+import subprocess
 import tempfile
 import uuid
 from pathlib import Path
@@ -156,13 +158,79 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
        return ""


-def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-avatar/covers") -> str:
+def apply_title_to_cover(local_frame: str, *, title_config: dict | None) -> str:
+    """用 ffmpeg drawtext 在封面图上叠加标题，返回叠加后图片的本地路径.
+
+    ffmpeg 失败时回退返回原始 local_frame。竖屏封面按 720x1280 计算位置。
+    """
+    if not title_config or not isinstance(title_config, dict):
+        return local_frame
+    text = (title_config.get("text") or title_config.get("content") or "").strip()
+    if not text:
+        return local_frame
+    enabled = title_config.get("enabled", True)
+    if not enabled:
+        return local_frame
+
+    try:
+        from packages.domain.video_filter_builder import build_title_drawtext_filter
+
+        drawtext_filter = build_title_drawtext_filter(
+            title_config,
+            output_width=720,
+            output_height=1280,
+        )
+        if not drawtext_filter:
+            return local_frame
+
+        base, ext = os.path.splitext(local_frame)
+        titled_path = f"{base}_titled{ext or '.jpg'}"
+        cmd = [
+            "ffmpeg",
+            "-i",
+            local_frame,
+            "-vf",
+            drawtext_filter,
+            "-y",
+            titled_path,
+        ]
+        logger.info("[数字人封面] 叠加标题: text=%s", text[:30])
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        if result.returncode != 0:
+            logger.warning(
+                "[数字人封面] drawtext 失败，回退无标题: exit=%s stderr=%s",
+                result.returncode,
+                (result.stderr or "")[-300:],
+            )
+            return local_frame
+        if not os.path.exists(titled_path) or os.path.getsize(titled_path) == 0:
+            logger.warning("[数字人封面] drawtext 输出为空，回退无标题")
+            return local_frame
+        return titled_path
+    except Exception as exc:
+        logger.warning("[数字人封面] 标题叠加异常，回退无标题: %s", exc, exc_info=True)
+        return local_frame
+
+
+def persist_cover_to_oss(
+    frame_url: str,
+    *,
+    job_id: str = "",
+    prefix: str = "ai-avatar/covers",
+    title_config: dict | None = None,
+) -> str:
    """下载帧图并转存到 OSS，返回公网封面 URL.

    Args:
        frame_url: MediaKit 返回的临时帧图 URL
        job_id: 关联任务 ID（用于 OSS key 命名）
        prefix: OSS key 前缀
+        title_config: 可选标题配置；传入时用 drawtext 叠加标题（竖屏 720x1280）

    Returns:
        OSS 公网 URL；失败回退原始 frame_url
@@ -170,6 +238,7 @@ def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-
    if not frame_url:
        return ""
    tmp_path: Optional[str] = None
+    titled_path: Optional[str] = None
    try:
        import httpx

@@ -189,12 +258,21 @@ def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-
        storage = get_shared_storage_service()
        token = job_id or uuid.uuid4().hex[:12]
        cover_key = f"{prefix}/{token}/cover_{uuid.uuid4().hex[:8]}.jpg"
+
+        upload_path = apply_title_to_cover(tmp_path, title_config=title_config)
+        if upload_path != tmp_path:
+            titled_path = upload_path
+
        public_url = storage.upload_file(
-            file_or_path=tmp_path,
+            file_or_path=upload_path,
            storage_key=cover_key,
            content_type="image/jpeg",
        )
-        logger.info("[数字人封面] 封面已转存 OSS: key=%s", cover_key)
+        logger.info(
+            "[数字人封面] 封面已转存 OSS: key=%s titled=%s",
+            cover_key,
+            bool(titled_path),
+        )
        # 私有桶：返回预签名 URL（前端才能加载）
        if public_url:
            signed = storage.get_download_url(cover_key, expires_seconds=86400)
@@ -204,19 +282,32 @@ def persist_cover_to_oss(frame_url: str, *, job_id: str = "", prefix: str = "ai-
        logger.warning("[数字人封面] 封面转存 OSS 失败，返回原始 URL", exc_info=True)
        return frame_url
    finally:
-        if tmp_path:
-            try:
-                Path(tmp_path).unlink(missing_ok=True)
-            except Exception:
-                pass
+        for p in (tmp_path, titled_path):
+            if p:
+                try:
+                    Path(p).unlink(missing_ok=True)
+                except Exception:
+                    pass


-def generate_smart_cover(video_url: str, *, job_id: str = "", max_frames: int = 5) -> str:
-    """一站式：MediaKit 智能抽帧选最佳 → 转存 OSS，返回封面公网 URL.
+def generate_smart_cover(
+    video_url: str,
+    *,
+    job_id: str = "",
+    max_frames: int = 5,
+    title_config: dict | None = None,
+) -> str:
+    """一站式：MediaKit 智能抽帧选最佳 → （可选）drawtext 叠加标题 → 转存 OSS.

    供独立封面接口与渲染管线复用。失败返回空字符串。
+
+    Args:
+        video_url: 可公网访问的视频 URL
+        job_id: 关联任务 ID
+        max_frames: 抽帧数量
+        title_config: 可选标题配置；传入时在封面上叠加 drawtext 标题（竖屏 720x1280）
    """
    best_frame = select_best_cover_frame(video_url, max_frames=max_frames)
    if not best_frame:
        return ""
-    return persist_cover_to_oss(best_frame, job_id=job_id)
+    return persist_cover_to_oss(best_frame, job_id=job_id, title_config=title_config)
@@ -11,6 +11,7 @@ from __future__ import annotations

 import logging
 import os
+import subprocess
 import tempfile
 import uuid
 from datetime import datetime, timezone
@@ -24,7 +25,6 @@ from packages.adapters.sqlalchemy_impl.models import (
    ScriptModel,
 )
 from packages.domain.video_filter_builder import (
-    build_cover_extract_command,
    build_title_drawtext_filter,
 )
 from packages.shared.storage import get_shared_storage_service
@@ -257,7 +257,7 @@ class AiAvatarRenderService:
            with tempfile.TemporaryDirectory() as tmpdir:
                output_video_path = os.path.join(tmpdir, "output.mp4")

-                cmd = self._build_ffmpeg_command(
+                cmd_list = self._build_ffmpeg_command(
                    input_video=input_video_path,
                    b_roll_segments=job.b_roll_segments,
                    filter_complex=filter_complex,
@@ -265,9 +265,25 @@ class AiAvatarRenderService:
                    output_path=output_video_path,
                )

-                exit_code = os.system(cmd)
-                if exit_code != 0:
-                    raise AiAvatarRenderError(f"FFmpeg 渲染失败，退出码: {exit_code}", code="FFmpegFailed")
+                try:
+                    render_result = subprocess.run(
+                        cmd_list,
+                        capture_output=True,
+                        text=True,
+                        timeout=600,
+                    )
+                except subprocess.TimeoutExpired as exc:
+                    raise AiAvatarRenderError(
+                        "FFmpeg 渲染超时（600s）",
+                        code="FFmpegTimeout",
+                    ) from exc
+
+                if render_result.returncode != 0:
+                    stderr_tail = (render_result.stderr or "").strip()[-800:]
+                    raise AiAvatarRenderError(
+                        f"FFmpeg 渲染失败，退出码: {render_result.returncode}, stderr: {stderr_tail}",
+                        code="FFmpegFailed",
+                    )

                job.progress = 80
                self.db.commit()
@@ -276,11 +292,27 @@ class AiAvatarRenderService:
                cover_path = ""
                if job.cover_config:
                    cover_path = os.path.join(tmpdir, "cover.jpg")
-                    cover_cmd = build_cover_extract_command(job.cover_config, cover_path)
-                    cover_cmd = cover_cmd.replace("INPUT_VIDEO", output_video_path)
-                    cover_exit = os.system(cover_cmd)
-                    if cover_exit != 0:
-                        logger.warning("封面提取失败，跳过: %s", cover_cmd)
+                    cover_cmd = self._build_cover_extract_cmd(
+                        cover_config=job.cover_config,
+                        input_video=output_video_path,
+                        output_path=cover_path,
+                    )
+                    try:
+                        cover_result = subprocess.run(
+                            cover_cmd,
+                            capture_output=True,
+                            text=True,
+                            timeout=60,
+                        )
+                        if cover_result.returncode != 0:
+                            logger.warning(
+                                "封面提取失败（非致命），跳过: exit=%s stderr=%s",
+                                cover_result.returncode,
+                                (cover_result.stderr or "")[-300:],
+                            )
+                            cover_path = ""
+                    except Exception as cover_err:
+                        logger.warning("封面提取异常（非致命），跳过: %s", cover_err)
                        cover_path = ""

                job.progress = 90
@@ -290,7 +322,7 @@ class AiAvatarRenderService:
                output_video_url = self._upload_to_oss(output_video_path, f"ai-avatar/{job_id}/output.mp4")
                job.output_video_url = output_video_url

-                # 封面：优先复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧；
+                # 封面：优先复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧（支持 drawtext 标题叠加）；
                # MediaKit 不可用时回退到 FFmpeg 已按 cover_config 抽取的 cover_path
                smart_cover_url = ""
                if output_video_url:
@@ -299,7 +331,13 @@ class AiAvatarRenderService:
                            generate_smart_cover,
                        )

-                        smart_cover_url = generate_smart_cover(output_video_url, job_id=job_id, max_frames=5)
+                        smart_cover_url = generate_smart_cover(
+                            output_video_url,
+                            job_id=job_id,
+                            max_frames=5,
+                            # 注意：不传 title_config —— 最终输出视频已经通过 drawtext 叠加了标题，
+                            # 再传会导致封面标题双重叠加
+                        )
                    except Exception:
                        logger.warning("智能封面(MediaKit)失败，回退 FFmpeg 封面 job_id=%s", job_id, exc_info=True)

@@ -360,12 +398,14 @@ class AiAvatarRenderService:
            job.updated_at = datetime.now(timezone.utc)
            self.db.commit()
            logger.error("渲染任务失败 [%s]: %s", job_id, exc)
+            raise
        except Exception as exc:
            job.status = "failed"
            job.error_message = f"渲染异常: {str(exc)}"
            job.updated_at = datetime.now(timezone.utc)
            self.db.commit()
            logger.exception("渲染任务异常 [%s]", job_id)
+            raise

    def _download_video(self, url: str) -> str:
        """下载视频到临时文件."""
@@ -391,24 +431,73 @@ class AiAvatarRenderService:
        filter_complex: str,
        final_label: Optional[str],
        output_path: str,
-    ) -> str:
-        """构建 FFmpeg 命令."""
-        # 输入文件
-        inputs = f"-i {input_video}"
+    ) -> list[str]:
+        """构建 FFmpeg 命令（list 形式，shell=False）.
+
+        根因修复 #1798 P0：OSS 预签名 URL 含 `&Expires=...&Signature=...` 特殊字符，
+        os.system(shell=True) 会把 `&` 解释为后台命令分隔符，导致 -filter_complex 被
+        当成独立命令报 sh: -filter_complex: not found（exit 127 → Python 32512）。
+        list + shell=False 彻底规避 shell 转义问题。
+        """
+        cmd: list[str] = ["ffmpeg", "-i", input_video]
        for seg in b_roll_segments:
            asset_url = seg.get("asset_url", "")
            if asset_url:
-                inputs += f" -i {asset_url}"
+                cmd.extend(["-i", asset_url])

-        # 滤镜
        if filter_complex and final_label:
-            filter_arg = f'-filter_complex "{filter_complex}" -map "[{final_label}]"'
+            cmd.extend(["-filter_complex", filter_complex, "-map", f"[{final_label}]"])
        elif filter_complex:
-            filter_arg = f'-filter_complex "{filter_complex}"'
-        else:
-            filter_arg = ""
+            cmd.extend(["-filter_complex", filter_complex])

-        return f"ffmpeg {inputs} {filter_arg} -c:v libx264 -preset veryfast -crf 23 -y {output_path}"
+        cmd.extend(
+            [
+                "-c:v",
+                "libx264",
+                "-preset",
+                "veryfast",
+                "-crf",
+                "23",
+                "-y",
+                output_path,
+            ]
+        )
+        return cmd
+
+    def _build_cover_extract_cmd(
+        self,
+        *,
+        cover_config: dict[str, Any],
+        input_video: str,
+        output_path: str,
+    ) -> list[str]:
+        """构建封面截帧 FFmpeg 命令（list 形式，shell=False）."""
+        if not cover_config or not isinstance(cover_config, dict):
+            timestamp = 0.0
+            width = 0
+            height = 0
+        else:
+            timestamp = cover_config.get("timestamp", 0.0)
+            width = cover_config.get("width", 0)
+            height = cover_config.get("height", 0)
+
+        cmd: list[str] = [
+            "ffmpeg",
+            "-ss",
+            str(timestamp),
+            "-i",
+            input_video,
+            "-frames:v",
+            "1",
+        ]
+        if width > 0 and height > 0:
+            vf = (
+                f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
+                f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2"
+            )
+            cmd.extend(["-vf", vf])
+        cmd.extend(["-y", output_path])
+        return cmd

    def _upload_to_oss(self, local_path: str, oss_key: str) -> str:
        """上传文件到 OSS，返回 URL.
@@ -211,12 +211,18 @@ class LipsyncService:
                        normalize_emotion(emotion),
                    )
                )
-            except Exception:
-                logger.warning(
-                    "Celery 任务提交失败，TTS 任务已创建但未触发执行: %s",
+            except Exception as exc:
+                # 投递失败时立即把 job 标成 failed 并写入 error_message，
+                # 前端轮询时能直接看到失败原因，不会无限卡在 tts_processing。
+                logger.exception(
+                    "Celery 任务提交失败，TTS 任务已创建但未触发执行: job_id=%s err=%s",
                    job_id,
-                    exc_info=True,
+                    exc,
                )
+                job.status = "failed"
+                job.error_message = f"Celery 任务投递失败: {exc}"
+                job.error_code = "AsyncDispatchFailed"
+                job.updated_at = datetime.now(timezone.utc)
        else:
            # 2b. 直接音频模式：同步签名并提交 MediaKit
            video_url = self._sign_media_url(video_url)
@@ -9,6 +9,9 @@
 5. 签名 URL 并提交到 MediaKit
 6. 更新 job 状态为 submitted
 7. 异常时标记 job 为 failed
+
+注意：使用 @shared_task 而非绑定到某个 celery_app 实例，
+确保任务能被 Worker 侧 celery_app 正确注册，同时 API 侧 send_task/apply_async 仍可正常调用。
 """

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

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

 logger = logging.getLogger(__name__)

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


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

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


-@celery_app.task(
+@shared_task(
    bind=True,
    name="lipsync_tts.synthesize_and_submit",
    max_retries=2,
@@ -73,12 +76,21 @@ def tts_synthesize_and_submit(
    from app.services.mediakit_client import MediaKitError, get_mediakit_client
    from sqlalchemy.orm import Session as DBSession

-    from packages.adapters.sqlalchemy_impl.database import SessionLocal
    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
    from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
-    from packages.shared.storage import get_shared_storage_service
    from packages.shared.url_security import safe_download_bytes

+    # SessionLocal 获取：
+    # - API 容器：app.db.SessionLocal（环境变量完整，导入即建引擎）
+    # - Worker 容器：worker_app.db.SessionLocal（Worker 自己的 settings 初始化引擎）
+    # API 侧没有 worker_app 模块 → ImportError 直接回退；
+    # Worker 侧 app.db 会因缺少 API 专有环境变量抛 pydantic ValidationError，
+    # 此时也要回退到 worker_app.db。
+    try:
+        from worker_app.db import SessionLocal  # type: ignore
+    except Exception:  # noqa: BLE001
+        from app.db import SessionLocal  # type: ignore
+
    db: DBSession = SessionLocal()
    try:
        job = (
@@ -144,11 +156,14 @@ def tts_synthesize_and_submit(
                    "audio/mpeg",
                    "audio/mp3",
                    "audio/wav",
+                    "audio/x-wav",  # CosyVoice 部分接口返回 audio/x-wav，与 audio/wav 等价（RIFF/WAVE）
                    "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")
@@ -164,7 +179,7 @@ def tts_synthesize_and_submit(

        db.commit()

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

@@ -552,7 +552,7 @@
  max-width: 240px;
  aspect-ratio: 9/16;
  background: #f0f0f5;
-  border-radius: 8px;
+  border-radius: 12px;
  overflow: hidden;
  display: flex;
  align-items: center;
@@ -564,8 +564,10 @@
 .aa-cover-preview img {
  width: 100%;
  height: 100%;
+  aspect-ratio: 9/16;
  object-fit: cover;
  display: block;
+  border-radius: 12px;
 }

 .aa-cover-preview__placeholder {
@@ -573,6 +575,19 @@
  color: #8c8ca1;
 }

+.aa-cover-preview__loading {
+  position: absolute;
+  inset: 0;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  background: rgba(0, 0, 0, 0.45);
+  color: #fff;
+  font-size: 13px;
+  backdrop-filter: blur(4px);
+  -webkit-backdrop-filter: blur(4px);
+}
+
 .aa-cover-actions {
  display: flex;
  gap: 8px;
@@ -279,7 +279,7 @@ const AiAvatarPage: React.FC = () => {
    }
    setSmartCoverLoading(true)
    try {
-      const res = await generateSmartCover(videoUrl, 5)
+      const res = await generateSmartCover(videoUrl, state.titleConfig, 5)
      if (res.cover_url) {
        state.setCoverConfig((prev) => ({
          ...prev,
@@ -342,7 +342,6 @@ const AiAvatarPage: React.FC = () => {
                  selectedVideo={state.selectedVideo}
                  onSelectVideo={() => state.setShowAssetPicker(true)}
                  onRemoveVideo={state.removeVideo}
-                  titleConfig={state.titleConfig}
                />
              </div>
            </div>
@@ -441,6 +440,7 @@ const AiAvatarPage: React.FC = () => {
              <div className="aa-panel__body">
                <PanelCoverAndGenerate
                  coverConfig={state.coverConfig}
+                  titleConfig={state.titleConfig}
                  onCoverConfigChange={(partial) =>
                    state.setCoverConfig((prev) => ({ ...prev, ...partial }))
                  }
@@ -2,7 +2,7 @@
 * AI数字人 — API 封装（#1822 契约对齐）
 */
 import apiClient from "@/api/client"
-import type { Script, LipsyncJob, RenderJob, BRollSegment } from "../types"
+import type { Script, LipsyncJob, RenderJob, BRollSegment, AiAvatarTitleConfig } from "../types"

 /* ── 文案库 ── */
 export const getScripts = async (): Promise<Script[]> => {
@@ -54,19 +54,21 @@ 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
 }

-/* ── 智能封面（MediaKit 抽帧 + 质量评分选最佳帧，独立于渲染任务） ── */
+/* ── 智能封面（MediaKit 抽帧 + 质量评分选最佳帧 + 可选 drawtext 标题叠加） ── */
 export const generateSmartCover = async (
  video_url: string,
+  title_config?: AiAvatarTitleConfig | null,
  max_frames = 5,
 ): Promise<{ cover_url: string; status: string; message: string }> => {
  const response = await apiClient.post<{ cover_url: string; status: string; message: string }>(
    "/ai-avatar/render/smart-cover",
-    { video_url, max_frames },
-    { timeout: 60000 },
+    { video_url, max_frames, title_config: title_config ?? null },
+    // smart-cover 链路：下载视频+抽帧+drawtext 加标题+上传 OSS，需要较长时间，120s 超时
+    { timeout: 120000 },
  )
  return response.data
 }
@@ -85,7 +87,7 @@ export const submitRender = async (data: {
 }

 export const getRenderJob = async (jobId: string): Promise<RenderJob> => {
-  const response = await apiClient.get<RenderJob>(`/ai-avatar/render/${jobId}`)
+  const response = await apiClient.get<RenderJob>(`/ai-avatar/render/${jobId}`, { timeout: 60000 })
  return response.data
 }

@@ -1,17 +1,18 @@
 /**
 * AI数字人 — 面板5：封面 & 生成
- * - 竖屏 9:16 封面预览（从视频截取 / 自定义上传）
+ * - 竖屏 9:16 封面预览（从视频截取 / 自定义上传）+ 标题文字实时叠加预览
 * - 分辨率选择（720p / 1080p / 4K）
 * - 配置汇总卡片（出镜视频/音色/文案/对口型/B-roll/标题/封面）
 * - 渐变紫色生成按钮
 *
 * 注意：v3 已删除"画面插入模式"，本面板不包含该选项。
 */
-import React, { useRef } from "react"
-import type { AiAvatarCoverConfig } from "../types"
+import React, { useMemo, useRef } from "react"
+import type { AiAvatarCoverConfig, AiAvatarTitleConfig } from "../types"

 interface PanelCoverAndGenerateProps {
  coverConfig: AiAvatarCoverConfig
+  titleConfig: AiAvatarTitleConfig
  onCoverConfigChange: (partial: Partial<AiAvatarCoverConfig>) => void
  resolution: string
  onResolutionChange: (r: string) => void
@@ -47,8 +48,19 @@ const LIPSYNC_STATUS_LABEL: Record<string, { text: string; cls: string }> = {
  failed: { text: "失败", cls: "aa-status-badge--failed" },
 }

+/** 字体名 → CSS font-family 映射（与后端 drawtext 对齐） */
+const FONT_FAMILY_MAP: Record<string, string> = {
+  思源黑体: "'Noto Sans SC', 'Source Han Sans SC', 'PingFang SC', 'Microsoft YaHei', sans-serif",
+  思源宋体: "'Noto Serif SC', 'Source Han Serif SC', 'SimSun', serif",
+  楷体: "KaiTi, 'STKaiti', serif",
+  黑体: "'Heiti SC', 'SimHei', 'Microsoft YaHei', sans-serif",
+}
+
+const getFontFamily = (font: string): string => FONT_FAMILY_MAP[font] || FONT_FAMILY_MAP["思源黑体"]
+
 const PanelCoverAndGenerate: React.FC<PanelCoverAndGenerateProps> = ({
  coverConfig,
+  titleConfig,
  onCoverConfigChange,
  resolution,
  onResolutionChange,
@@ -86,15 +98,85 @@ const PanelCoverAndGenerate: React.FC<PanelCoverAndGenerateProps> = ({

  const canGenerate = summary.lipsyncStatus === "completed" && !isGenerating

+  /** 封面图实际展示的 url：智能封面 > 自定义上传 > 空 */
+  const coverUrl =
+    coverConfig.smart_cover_url || coverConfig.thumbnail_url || coverConfig.upload_url
+  const hasCoverImage = Boolean(coverUrl)
+
+  /** 是否显示标题叠加层：有图、有文字、非加载中 */
+  const showTitleOverlay =
+    hasCoverImage && !smartCoverLoading && titleConfig.title.trim().length > 0
+
+  /** 计算标题叠加层的 inline 样式 */
+  const titleOverlayStyle = useMemo<React.CSSProperties>(() => {
+    const style: React.CSSProperties = {
+      position: "absolute",
+      left: "50%",
+      width: "90%",
+      transform: "translateX(-50%)",
+      textAlign: "center",
+      boxSizing: "border-box",
+      padding: "0 4px",
+      wordBreak: "break-word",
+      whiteSpace: "pre-wrap",
+      color: titleConfig.color || "#ffffff",
+      fontSize: `${titleConfig.size}px`,
+      fontFamily: getFontFamily(titleConfig.font),
+      fontWeight: titleConfig.bold ? "bold" : "normal",
+      fontStyle: titleConfig.italic ? "italic" : "normal",
+      lineHeight: 1.3,
+      pointerEvents: "none",
+    }
+
+    // 位置
+    const pos = titleConfig.position || "bottom"
+    if (pos === "top") {
+      style.top = "40px"
+    } else if (pos === "center") {
+      style.top = "50%"
+      style.transform = "translate(-50%, -50%)"
+    } else if (pos === "custom" && titleConfig.pos_x != null && titleConfig.pos_y != null) {
+      // pos_x/pos_y 是相对预览容器的百分比坐标
+      style.left = `${titleConfig.pos_x}%`
+      style.top = `${titleConfig.pos_y}%`
+      style.transform = "translate(-50%, -50%)"
+    } else {
+      style.bottom = "40px"
+    }
+
+    // 描边优先于阴影（二者互斥，与 drawtext 对齐）
+    if (titleConfig.stroke) {
+      // 描边宽度按字号估算，保证视觉一致
+      const strokeWidth = Math.max(1, Math.round(titleConfig.size / 18))
+      ;(style as React.CSSProperties)["WebkitTextStroke"] = `${strokeWidth}px rgba(0,0,0,0.75)`
+      style.textShadow = "none"
+    } else if (titleConfig.shadow) {
+      style.textShadow = "0 2px 8px rgba(0,0,0,0.7), 0 0 2px rgba(0,0,0,0.5)"
+    } else {
+      // 默认给轻微阴影保证白字在亮背景可读
+      style.textShadow = "0 2px 6px rgba(0,0,0,0.6)"
+    }
+
+    return style
+  }, [titleConfig])
+
  return (
    <div className="aa-cover-generate">
      {/* 封面预览（竖屏 9:16） */}
      <div className="aa-cover-preview">
-        {coverConfig.thumbnail_url ? (
-          <img src={coverConfig.thumbnail_url} alt="封面预览" />
+        {hasCoverImage ? (
+          <img src={coverUrl!} alt="封面预览" draggable={false} />
        ) : (
          <span className="aa-cover-preview__placeholder">暂无封面</span>
        )}
+        {/* 智能封面加载遮罩 */}
+        {smartCoverLoading && <div className="aa-cover-preview__loading">⏳ 智能选帧中…</div>}
+        {/* 标题文字叠加层（实时预览，仅前端视觉参考，最终由后端 ffmpeg drawtext 叠加） */}
+        {showTitleOverlay && (
+          <div style={titleOverlayStyle} aria-hidden="true">
+            {titleConfig.title}
+          </div>
+        )}
      </div>

      <div className="aa-cover-actions">
@@ -2,17 +2,16 @@
 * AI数字人 — 出镜视频选择面板
 * - 未选视频：虚线上传区，点击打开素材库弹窗
 * - 已选视频：竖屏 9:16 预览播放器 + 视频信息卡片 + 移除按钮
+ *
+ * 注意：本面板只展示原始素材视频，不叠加标题（标题在对口型预览和最终成片上展示）
 */
 import type { AssetItem } from "@/api/assets"
-import type { AiAvatarTitleConfig } from "../types"
-import { getFontFamily } from "@/pages/generate/constants"

 export interface PanelVideoSelectorProps {
  selectedVideo: AssetItem | null
  /** 触发打开素材库弹窗 */
  onSelectVideo: () => void
  onRemoveVideo: () => void
-  titleConfig?: AiAvatarTitleConfig
 }

 /** 格式化时长（秒 → mm:ss） */
@@ -27,7 +26,6 @@ export function PanelVideoSelector({
  selectedVideo,
  onSelectVideo,
  onRemoveVideo,
-  titleConfig,
 }: PanelVideoSelectorProps) {
  /* 未选视频：虚线上传区，点击打开素材库弹窗 */
  if (!selectedVideo) {
@@ -57,42 +55,13 @@ export function PanelVideoSelector({

  return (
    <div>
-      {/* 竖屏 9:16 视频预览播放器 + 标题实时预览 */}
-      <div className="aa-video-preview" style={{ position: "relative" }}>
+      {/* 竖屏 9:16 视频预览播放器（纯素材预览，不叠加标题） */}
+      <div className="aa-video-preview">
        {fileUrl ? (
          <video src={fileUrl} poster={selectedVideo.thumbnail_url} controls playsInline />
        ) : (
          <div className="aa-video-preview__placeholder">视频暂不可预览</div>
        )}
-        {titleConfig?.title && (
-          <div
-            style={{
-              position: "absolute",
-              left: "50%",
-              transform: "translateX(-50%)",
-              ...(titleConfig.position === "top"
-                ? { top: "10%" }
-                : titleConfig.position === "bottom"
-                  ? { bottom: "10%" }
-                  : { top: "50%", transform: "translate(-50%, -50%)" }),
-              fontSize: Math.max(titleConfig.size, 32),
-              fontFamily: getFontFamily(titleConfig.font),
-              color: titleConfig.color,
-              fontWeight: titleConfig.bold ? 700 : 400,
-              fontStyle: titleConfig.italic ? "italic" : "normal",
-              textShadow: "0 2px 4px rgba(0,0,0,0.5)",
-              WebkitTextStroke: "2px #000",
-              pointerEvents: "none",
-              zIndex: 10,
-              maxWidth: "90%",
-              textAlign: "center",
-              whiteSpace: "pre-wrap",
-              lineHeight: 1.3,
-            }}
-          >
-            {titleConfig.title}
-          </div>
-        )}
      </div>

      {/* 视频信息卡片：文件名 / 时长 / 分辨率 */}
@@ -34,10 +34,15 @@ celery_app.conf.imports = (
    "worker_app.tasks.tts_synthesis",
    "worker_app.tasks.batch_download",
    "worker_app.tasks.duplication_check",
+    # #1798 AI 数字人渲染：必须在 Worker 实例上注册同名任务，否则消息无人消费（渲染卡 0%）
+    "worker_app.tasks.ai_avatar_render",
    "worker_app.tasks._startup",
    "apps.worker.video_processing.dedup",
    "worker_app.tasks.cleanup",
-    "apps.api.app.tasks.lipsync_tts",
+    # 注意：必须用 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 定时任务调度
@@ -0,0 +1,48 @@
+"""AI 数字人渲染任务 — Worker 侧 Celery 任务注册.
+
+#1798 渲染进度卡在 0% 的根因：渲染任务定义在 API 侧（`app.tasks.ai_avatar_render`），
+装饰在 API 自己的 Celery 实例（`xiaoxia-saas-api`）上；而 Worker 用的是
+`worker_app.celery_app` 实例，`conf.imports` 从未导入该任务，Worker 的任务
+注册表里没有 `ai_avatar_render.execute`，消息被路由到默认 `celery` 队列后
+无人消费，任务永远停在 0%。
+
+修复：在 Worker 侧用 `worker_app.celery_app` 注册同名任务，直接调用与 API
+服务一致的 `AiAvatarRenderService.execute_render` 核心管线（业务逻辑在
+`apps.api.app.services`，worker 镜像已复制 `apps/api/app`）。任务名保持
+`ai_avatar_render.execute`，与 API 生产端 `.delay()` 的消息路由一致；未在
+task_routes 显式配置，走默认 `celery` 队列，由 transcode worker 消费。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, name="ai_avatar_render.execute", max_retries=2)
+def execute_ai_avatar_render(self, job_id: str) -> dict:
+    """执行 AI 数字人渲染管线（Worker 侧入口）.
+
+    进度由 service 直接写入 DB（AiAvatarRenderJob.progress:
+    0→5→20→40→80→90→95→100），API 通过轮询 progress 字段展示。
+    """
+    logger.info("开始执行渲染任务: %s", job_id)
+    self.update_state(state="PROCESSING", meta={"progress": 0, "job_id": job_id})
+
+    session = SessionLocal()
+    try:
+        from app.services.ai_avatar_render_service import AiAvatarRenderService
+
+        service = AiAvatarRenderService(session)
+        service.execute_render(job_id)
+        return {"status": "completed", "job_id": job_id}
+    except Exception as exc:
+        logger.exception("渲染任务执行异常 [%s]: %s", job_id, exc)
+        self.update_state(state="FAILED", meta={"progress": 0, "error": str(exc)})
+        raise
+    finally:
+        session.close()
@@ -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
    
@@ -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..."
@@ -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 \
@@ -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
@@ -0,0 +1,49 @@
+#!/bin/sh
+# Select nginx config based on APP_ENV (staging/production).
+#
+# 两种运行模式：
+# 1. CI/CD 部署（staging/production）：部署脚本通过 `-v 宿主机文件:/etc/nginx/conf.d/default.conf:ro`
+#    把宿主机生成的带 resolver/docker upstream 的配置 bind mount 进来，entrypoint 不应改动。
+#    bind mount 的文件是 readonly 的，rm 会报 EBUSY ("Resource busy")，直接 exec nginx 即可。
+# 2. 本地 docker-compose / 直接 `docker run`（无外部挂载）：镜像烤入了 nginx-staging.conf 与
+#    nginx-production.conf 到 /etc/nginx/，entrypoint 根据 APP_ENV 把 default.conf 换成正确的 symlink。
+#
+# 策略：
+# - 如果 /etc/nginx/conf.d/default.conf 已经是指向目标 conf 的 symlink，什么都不做；
+# - 否则尝试 rm -f 再 ln -s；rm 失败说明是外部 bind mount（已有正确配置），不阻塞启动；
+# - 兜底：只要 conf.d 目录里有 .conf 文件（含 bind mount 来的），就直接启动 nginx。
+set -e
+
+NGINX_CONF_DIR="/etc/nginx/conf.d"
+TARGET_CONF=""
+
+case "${APP_ENV:-production}" in
+  staging)
+    TARGET_CONF="/etc/nginx/nginx-staging.conf"
+    ;;
+  *)
+    TARGET_CONF="/etc/nginx/nginx-production.conf"
+    ;;
+esac
+
+DEFAULT_CONF="$NGINX_CONF_DIR/default.conf"
+
+# 1. 已经是正确的 symlink：直接启动
+if [ -L "$DEFAULT_CONF" ] && [ "$(readlink "$DEFAULT_CONF" 2>/dev/null)" = "$TARGET_CONF" ]; then
+  exec nginx -g "daemon off;"
+fi
+
+# 2. 尝试替换为目标 symlink（无 bind mount 的场景）
+#    若 rm 失败（bind mount readonly，EBUSY/EPERM），则认为外部已注入配置，不阻塞。
+rm -f "$DEFAULT_CONF" 2>/dev/null || true
+if [ -f "$TARGET_CONF" ] && [ ! -e "$DEFAULT_CONF" ]; then
+  ln -s "$TARGET_CONF" "$DEFAULT_CONF" 2>/dev/null || true
+fi
+
+# 3. 兜底：至少要有一个 .conf 文件，否则 nginx 起不来
+if ! ls "$NGINX_CONF_DIR"/*.conf >/dev/null 2>&1; then
+  echo "ERROR: no nginx config found in $NGINX_CONF_DIR (tried $TARGET_CONF and external bind mount)" >&2
+  exit 1
+fi
+
+exec nginx -g "daemon off;"
@@ -1,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"]
@@ -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"]
@@ -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,15 +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/
-# API 侧 Celery 任务（lipsync_tts 等）在 worker 进程中执行，需复制任务文件、依赖及 __init__.py
-RUN mkdir -p /app/apps && touch /app/apps/__init__.py
-COPY apps/api/__init__.py /app/apps/api/__init__.py
-COPY apps/api/app/__init__.py /app/apps/api/app/__init__.py
-COPY apps/api/app/services/__init__.py /app/apps/api/app/services/__init__.py
-COPY apps/api/app/services/mediakit_client.py /app/apps/api/app/services/mediakit_client.py
-COPY apps/api/app/tasks/ /app/apps/api/app/tasks/
+# 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
@@ -0,0 +1 @@
+# xiaoxia-saas shared packages namespace
@@ -0,0 +1 @@
+# adapter implementations namespace
@@ -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}")
@@ -547,7 +547,7 @@ class TestAiAvatarRenderService:
        with (
            patch.object(svc, "_download_video", return_value="/tmp/video.mp4"),
            patch.object(svc, "_upload_to_oss", side_effect=lambda path, key: f"https://oss/{key}"),
-            patch("os.system", return_value=0),
+            patch("subprocess.run") as mock_run,
            patch("tempfile.TemporaryDirectory") as tmpdir_mock,
            patch(
                "app.services.ai_avatar_cover_service.generate_smart_cover", return_value="https://oss/smart_cover.jpg"
@@ -557,6 +557,9 @@ class TestAiAvatarRenderService:
                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository"
            ) as repo_cls,
        ):
+            import subprocess as _sp
+
+            mock_run.return_value = _sp.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
            import tempfile as _tf

            tmpdir_mock.return_value.__enter__ = MagicMock(return_value="/tmp/testdir")
@@ -605,10 +608,13 @@ class TestAiAvatarRenderService:
        with (
            patch.object(svc, "_download_video", return_value="/tmp/video.mp4"),
            patch.object(svc, "_upload_to_oss", side_effect=lambda path, key: f"https://oss/{key}"),
-            patch("os.system", return_value=0),
+            patch("subprocess.run") as mock_run,
            patch("tempfile.TemporaryDirectory") as tmpdir_mock,
            patch("app.services.ai_avatar_cover_service.generate_smart_cover", side_effect=RuntimeError("DB error")),
        ):
+            import subprocess as _sp
+
+            mock_run.return_value = _sp.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
            tmpdir_mock.return_value.__enter__ = MagicMock(return_value="/tmp/testdir")
            tmpdir_mock.return_value.__exit__ = MagicMock(return_value=False)
            svc.execute_render("render-clip-fail")
@@ -35,7 +35,10 @@ class TestFFmpegPresetOptimization:
            final_label=None,
            output_path="/tmp/output.mp4",
        )
-        assert "-preset veryfast" in cmd, f"期望 -preset veryfast，实际命令: {cmd}"
+        # cmd 现在是 list[str]；preset 与值是相邻两个元素
+        assert "-preset" in cmd, f"期望包含 -preset，实际命令: {cmd}"
+        preset_idx = cmd.index("-preset")
+        assert cmd[preset_idx + 1] == "veryfast", f"期望 veryfast，实际: {cmd}"

    def test_preset_veryfast_with_filter(self):
        """带滤镜场景下也必须使用 veryfast."""
@@ -49,7 +52,8 @@ class TestFFmpegPresetOptimization:
            final_label="[v]",
            output_path="/tmp/output.mp4",
        )
-        assert "-preset veryfast" in cmd
+        assert "-preset" in cmd
+        assert cmd[cmd.index("-preset") + 1] == "veryfast"
        assert "-filter_complex" in cmd

    def test_preset_not_fast(self):
@@ -65,11 +69,11 @@ class TestFFmpegPresetOptimization:
            output_path="/tmp/output.mp4",
        )
        # 确保是 veryfast 而不是 fast
-        assert "-preset veryfast" in cmd
-        # 排除 "fast" 单独出现（veryfast 包含 fast 子串，需精确判断）
-        parts = cmd.split()
-        preset_idx = parts.index("-preset")
-        assert parts[preset_idx + 1] == "veryfast"
+        assert "-preset" in cmd
+        preset_idx = cmd.index("-preset")
+        assert cmd[preset_idx + 1] == "veryfast"
+        # 禁止 fast 单独作为 preset 值（veryfast 包含 "fast" 子串，不影响）
+        assert cmd[preset_idx + 1] != "fast"


 # ═══════════════════════════════════════════════════════════════════════════════
@@ -143,8 +147,8 @@ class TestCreateJobAsyncTTS:
        assert args[2] == "v-1"  # voice_id
        assert args[3] == "测试文本"  # script_text

-    def test_tts_mode_celery_dispatch_failure_still_creates_job(self):
-        """Celery dispatch 失败时，job 记录已创建，状态保持 tts_processing."""
+    def test_tts_mode_celery_dispatch_failure_marks_job_failed(self):
+        """Celery dispatch 失败时，job 标为 failed 并写入 error_message，前端轮询能直接看到错误."""
        svc, client, cosy = _make_service_with_mocks()

        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
@@ -157,9 +161,11 @@ class TestCreateJobAsyncTTS:
                script_text="测试文本",
            )

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

@@ -282,342 +288,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()
@@ -0,0 +1,389 @@
+"""AI 数字人口型 TTS Celery 异步任务 — 单元测试.
+
+覆盖 lipsync_tts.py 的全部主要分支：
+- Job 不存在/cancelled/正常/异常路径
+- TTS 合成、音频下载、OSS 上传、MediaKit 提交
+- CosyVoiceError/ValueError/MediaKitError/顶层异常等错误码
+"""
+
+import os
+import sys
+import types
+from types import ModuleType
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
+
+
+class _FakeQuery:
+    """模拟 SQLAlchemy query.filter().first() 链式调用."""
+
+    def __init__(self, job):
+        self._job = job
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def first(self):
+        return self._job
+
+
+def _make_fake_job(**kwargs):
+    """构造可 setattr 的 job 记录."""
+    job = MagicMock()
+    job.id = kwargs.get("job_id", "job-1")
+    job.user_id = kwargs.get("user_id", "user-1")
+    job.status = kwargs.get("status", "tts_processing")
+    job.audio_url = kwargs.get("audio_url", "")
+    job.video_url = kwargs.get("video_url", "https://oss/video.mp4")
+    job.mediakit_task_id = kwargs.get("mediakit_task_id", "")
+    job.enable_video_loop = kwargs.get("enable_video_loop", False)
+    job.error_code = ""
+    job.error_message = ""
+    job.submitted_at = None
+    job.updated_at = None
+    return job
+
+
+def _build_session(job):
+    """构造 mock DB session + factory. 返回 (session, factory)."""
+    session = MagicMock()
+    session.query.return_value = _FakeQuery(job)
+    session.commit = MagicMock()
+    session.close = MagicMock()
+    factory = MagicMock(return_value=session)
+    return session, factory
+
+
+def _apply_all_patches(
+    *,
+    job=None,
+    cosyvoice_service=None,
+    cosyvoice_side_effect=None,
+    cosyvoice_error=None,
+    download_bytes=b"AUDIO",
+    download_error=None,
+    storage=None,
+    mk_client=None,
+    mk_submit_return=None,
+    mk_submit_error=None,
+):
+    """统一构造测试需要的 patch 列表.
+
+    lipsync_tts.run() 在函数体内部懒 import 多个模块，通过 sys.modules 注入
+    伪造包路径避免真实导入；对存在的模块用 patch() 替换返回值/side_effect。
+    """
+    # SessionLocal 通过懒探测获取（Worker 用 worker_app.db，API 用 app.db），
+    # 测试环境里两个模块都能被真实导入，必须同时 mock 保证用的是 fake session。
+    fake_app_db = ModuleType("app.db")
+    fake_worker_db = ModuleType("worker_app.db")
+    session, factory = _build_session(job)
+    fake_app_db.SessionLocal = factory
+    fake_worker_db.SessionLocal = factory
+
+    patches = [
+        patch.dict(sys.modules, {"app.db": fake_app_db, "worker_app.db": fake_worker_db}),
+        patch(
+            "app.tasks.lipsync_tts._sign_media_url",
+            side_effect=lambda url: url + "?signed" if url else url,
+        ),
+    ]
+
+    # CosyVoice
+    if cosyvoice_service is not None:
+        cosy_instance = cosyvoice_service
+    else:
+        cosy_instance = MagicMock()
+        if cosyvoice_side_effect is not None:
+            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_side_effect
+        elif cosyvoice_error is not None:
+            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_error
+        else:
+            cosy_instance.submit_synthesize_task.return_value = {"audio_url": "https://tts/raw.mp3"}
+    patches.append(patch("packages.application.cosyvoice_service.CosyVoiceService", return_value=cosy_instance))
+
+    # safe_download_bytes
+    if download_error is not None:
+        patches.append(patch("packages.shared.url_security.safe_download_bytes", side_effect=download_error))
+    else:
+        patches.append(patch("packages.shared.url_security.safe_download_bytes", return_value=download_bytes))
+
+    # Storage
+    if storage is None:
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.upload_file.return_value = "https://oss.example.com/tts.mp3"
+    patches.append(patch("packages.shared.storage.get_shared_storage_service", return_value=storage))
+
+    # MediaKit client
+    if mk_client is not None:
+        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=mk_client))
+    else:
+        client = MagicMock()
+        if mk_submit_error is not None:
+            client.submit_lipsync.side_effect = mk_submit_error
+        else:
+            client.submit_lipsync.return_value = mk_submit_return or {"task_id": "mk-1"}
+        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=client))
+
+    return session, patches
+
+
+class TestTtsSynthesizeAndSubmit:
+    """测试 Celery 任务 tts_synthesize_and_submit.run 的所有分支."""
+
+    def test_job_not_found_returns_early(self):
+        """Job 不存在 → 日志报错直接返回，不抛异常."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        session, patches = _apply_all_patches(job=None)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("missing-job", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        session.commit.assert_not_called()
+        session.close.assert_called_once()
+
+    def test_cancelled_job_skipped(self):
+        """Job 已 cancelled → 跳过不处理，不调用 TTS/MediaKit."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job(status="cancelled")
+        session, patches = _apply_all_patches(job=job)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        # cancelled 不应 commit，不应触发 TTS/MediaKit
+        session.commit.assert_not_called()
+        session.close.assert_called_once()
+
+    def test_happy_path_tts_to_mediakit(self):
+        """完整正常流程：TTS 合成 → OSS 上传 → 签名 → 提交 MediaKit → submitted."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        mk_client = MagicMock()
+        mk_client.submit_lipsync.return_value = {"task_id": "mk-999"}
+        session, patches = _apply_all_patches(job=job, mk_client=mk_client)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好世界", 1.0, "happy")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-999"
+        assert job.error_code == ""
+        mk_client.submit_lipsync.assert_called_once()
+        call_kwargs = mk_client.submit_lipsync.call_args.kwargs
+        assert call_kwargs["client_token"] == "job-1"
+        assert call_kwargs["audio_url"].endswith("?signed")
+        session.commit.assert_called()
+        session.close.assert_called_once()
+
+    def test_cosyvoice_error_marks_tts_synthesis_failed(self):
+        """CosyVoiceError → failed, error_code=TTSSynthesisFailed."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        from packages.application.cosyvoice_service import CosyVoiceError
+
+        job = _make_fake_job()
+        session, patches = _apply_all_patches(job=job, cosyvoice_error=CosyVoiceError("tts boom"))
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSSynthesisFailed"
+        session.close.assert_called_once()
+
+    def test_value_error_marks_tts_invalid_param(self):
+        """ValueError（参数错误）→ failed, error_code=TTSInvalidParam."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        session, patches = _apply_all_patches(job=job, cosyvoice_side_effect=ValueError("bad param"))
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", -1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSInvalidParam"
+        session.close.assert_called_once()
+
+    def test_no_audio_url_marks_tts_no_audio(self):
+        """TTS 返回空 audio_url → failed, error_code=TTSNoAudio."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        cosy = MagicMock()
+        cosy.submit_synthesize_task.return_value = {"audio_url": ""}
+        session, patches = _apply_all_patches(job=job, cosyvoice_service=cosy)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "TTSNoAudio"
+        session.close.assert_called_once()
+
+    def test_oss_upload_failure_falls_back_to_temp_url(self):
+        """OSS 上传失败 → 回退临时 URL，仍然 submitted."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        storage = MagicMock()
+        storage.public_url = "https://oss.example.com"
+        storage.upload_file.side_effect = RuntimeError("oss down")
+        mk_client = MagicMock()
+        mk_client.submit_lipsync.return_value = {"task_id": "mk-7"}
+        session, patches = _apply_all_patches(job=job, storage=storage, mk_client=mk_client)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        # 上传失败后 audio_url 回退为临时 TTS URL，仍继续提交到 MediaKit
+        assert job.audio_url == "https://tts/raw.mp3"
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-7"
+        mk_client.submit_lipsync.assert_called_once()
+        session.close.assert_called_once()
+
+    def test_mediakit_error_marks_mediakit_unavailable(self):
+        """MediaKit 提交失败 → failed, error_code=MediaKitUnavailable."""
+        from app.services.mediakit_client import MediaKitError
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        mk_err = MediaKitError("mk down", code="MediaKitUnavailable")
+        session, patches = _apply_all_patches(job=job, mk_submit_error=mk_err)
+        entered = [p.__enter__() for p in patches]
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+
+        assert job.status == "failed"
+        assert job.error_code == "MediaKitUnavailable"
+        session.close.assert_called_once()
+
+    def test_top_level_exception_marks_async_task_error(self):
+        """顶层未预期异常 → failed, error_code=AsyncTaskError."""
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        job = _make_fake_job()
+        fake_app_db = ModuleType("app.db")
+        fake_worker_db = ModuleType("worker_app.db")
+        session, factory = _build_session(job)
+        fake_app_db.SessionLocal = factory
+        fake_worker_db.SessionLocal = factory
+
+        # CosyVoiceService 在 __init__ 抛 RuntimeError（非 CosyVoiceError/ValueError）
+        fake_cosy_mod = ModuleType("packages.application.cosyvoice_service")
+
+        class _CosyVoiceErrorForTest(Exception):
+            pass
+
+        class _BoomService:
+            def __init__(self):
+                raise RuntimeError("top-level boom")
+
+        fake_cosy_mod.CosyVoiceError = _CosyVoiceErrorForTest
+        fake_cosy_mod.CosyVoiceService = _BoomService
+
+        with patch.dict(
+            sys.modules,
+            {
+                "app.db": fake_app_db,
+                "worker_app.db": fake_worker_db,
+                "packages.application.cosyvoice_service": fake_cosy_mod,
+            },
+        ):
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+
+        assert job.status == "failed"
+        assert job.error_code == "AsyncTaskError"
+        session.close.assert_called()
+
+
+class TestSignMediaUrl:
+    """覆盖模块内 _sign_media_url 的所有分支（CI 增量覆盖率需要）."""
+
+    def test_empty_url_returns_empty(self):
+        from app.tasks.lipsync_tts import _sign_media_url
+
+        assert _sign_media_url("") == ""
+        assert _sign_media_url(None) is None
+
+    def test_own_oss_url_signed(self):
+        """自家 OSS URL → 调用 storage.get_download_url 签名."""
+        from app.tasks.lipsync_tts import _sign_media_url
+
+        fake_storage = MagicMock()
+        fake_storage.public_url = "https://oss.example.com/"
+        fake_storage.get_download_url.return_value = "https://oss.example.com/a?sig=xyz"
+
+        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
+            result = _sign_media_url("https://oss.example.com/lipsync/a.mp3")
+
+        assert result == "https://oss.example.com/a?sig=xyz"
+        fake_storage.get_download_url.assert_called_once()
+
+    def test_external_url_passthrough(self):
+        """外部 URL（不是自家 OSS host）→ 原样透传，不签名."""
+        from app.tasks.lipsync_tts import _sign_media_url
+
+        fake_storage = MagicMock()
+        fake_storage.public_url = "https://oss.example.com/"
+
+        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
+            result = _sign_media_url("https://tts.example.com/raw.mp3")
+
+        assert result == "https://tts.example.com/raw.mp3"
+        fake_storage.get_download_url.assert_not_called()
+
+    def test_storage_exception_falls_back(self):
+        """storage 调用异常 → 降级原样返回，不抛错."""
+        from app.tasks.lipsync_tts import _sign_media_url
+
+        with patch(
+            "packages.shared.storage.get_shared_storage_service",
+            side_effect=RuntimeError("storage down"),
+        ):
+            result = _sign_media_url("https://oss.example.com/a.mp3")
+
+        assert result == "https://oss.example.com/a.mp3"
+
+    def test_no_public_url_passthrough(self):
+        """storage.public_url 为空 → 原样透传."""
+        from app.tasks.lipsync_tts import _sign_media_url
+
+        fake_storage = MagicMock()
+        fake_storage.public_url = ""
+
+        with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage):
+            result = _sign_media_url("https://anything.example.com/a.mp3")
+
+        assert result == "https://anything.example.com/a.mp3"
+        fake_storage.get_download_url.assert_not_called()