feat(gpu): #1978 AI数字人口型同步接入MuseTalk GPU Worker

业务侧（LipsyncService）集成 GpuLipsyncService：
- 新增开关 USE_GPU_LIPSYNC（env，默认false；staging默认true方便联调，生产默认false待稳定后开）
- _submit_audio_direct 在签名提交 MediaKit 前判断：
  * 开关关闭 → 直接走现有 MediaKit 云端 lipsync
  * 开关开但无可用 Worker（last_heartbeat_at 超5分钟窗口） → 回退 MediaKit
  * 开关开+Worker可用 → 创建GPU任务→同步轮询等待结果
- GPU路径成功：直接标记 job.status=completed、output_video_url=签名后的结果URL
- GPU超时（默认1200s）/终态failed/异常 → 全部回退 MediaKit，对用户透明
- 轮询间隔默认5s，可配 GPU_LIPSYNC_POLL_INTERVAL
- Worker心跳新鲜度窗口可配 GPU_WORKER_STALE_SECONDS（默认300s）

GpuLipsyncService新增方法：
- has_available_worker()：判断5分钟内有心跳的Worker存在
- wait_for_result(task_id, timeout, poll_interval)：同步轮询DB等待终态，
  期间顺手 _recover_timed_out_tasks；超时返回None让调用方兜底

env/CI配置：
- .env.example 新增4个配置项注释
- deploy/configs/.env.staging USE_GPU_LIPSYNC=true（联调）
- deploy/configs/.env.production USE_GPU_LIPSYNC=false（暂不开）

单元测试：tests/unit/test_lipsync_gpu_integration.py 9个用例覆盖：
- 开关关不调用GPU、无Worker回退、GPU成功completed、超时回退、
  failed回退、异常回退、has_available_worker三种场景
- 全部 30个GPU相关测试全绿（含此前11+10个）

.env.example

@@ -213,3 +213,20 @@ TIKHUB_API_KEY=
 # apizero.cn API Key (https://v1.apizero.cn) — 国内抖音解析服务
 APIZERO_API_KEY=
 
+# ==================== GPU MuseTalk Worker（反向轮询口型同步）====================
+# GPU Worker 长期鉴权 Token，Worker 端 .env 的 GPU_WORKER_TOKEN 必须与此一致
+# 留空时 development 环境允许匿名访问（仅本地调试），staging/production 必须配置
+GPU_WORKER_TOKEN=
+# 单任务超时（秒），processing 超过此时长无任务心跳才回退 pending 或标记 failed
+# #1970：RTX2060 6G 推理 720p 长视频需 5 分钟以上，默认 900
+GPU_TASK_TIMEOUT_SECONDS=900
+# 是否启用 GPU 口型同步（开关）。开启后需同时有 Worker 在心跳窗口内（5分钟）才会走 GPU 路径；
+# 开关关闭 / 无可用 Worker / GPU 任务失败或超时 → 自动回退现有 MediaKit 云端 lipsync
+USE_GPU_LIPSYNC=false
+# 业务侧轮询 GPU 任务结果的间隔（秒）
+GPU_LIPSYNC_POLL_INTERVAL=5
+# 业务侧等待 GPU 任务总超时（秒）；超时回退 MediaKit
+GPU_LIPSYNC_WAIT_TIMEOUT=1200
+# Worker 心跳新鲜度窗口（秒），last_heartbeat_at 在此窗口内视为在线
+GPU_WORKER_STALE_SECONDS=300
+

.gitea/workflows/ci-pipeline.yml

@@ -1186,10 +1186,12 @@ jobs:
         DOUBAO_API_KEY: "${{ secrets.DOUBAO_API_KEY }}"
         DOUBAO_MODEL: "${{ secrets.DOUBAO_MODEL }}"
         DOUBAO_BASE_URL: "${{ secrets.DOUBAO_BASE_URL }}"
+        DOUBAO_VISION_MODEL: "${{ secrets.DOUBAO_VISION_MODEL }}"
         WECHAT_APP_ID: "${{ secrets.WECHAT_APP_ID }}"
         WECHAT_APP_SECRET: "${{ secrets.WECHAT_APP_SECRET }}"
         TIKHUB_API_KEY: "${{ secrets.TIKHUB_API_KEY }}"
         APIZERO_API_KEY: "${{ secrets.APIZERO_API_KEY }}"
+        GPU_WORKER_TOKEN: "${{ secrets.GPU_WORKER_TOKEN }}"
       run: |
         set -eu
         echo "Rendering .env from template + secrets..."
@@ -1640,10 +1642,12 @@ jobs:
         DOUBAO_API_KEY: "${{ secrets.DOUBAO_API_KEY }}"
         DOUBAO_MODEL: "${{ secrets.DOUBAO_MODEL }}"
         DOUBAO_BASE_URL: "${{ secrets.DOUBAO_BASE_URL }}"
+        DOUBAO_VISION_MODEL: "${{ secrets.DOUBAO_VISION_MODEL }}"
         WECHAT_APP_ID: "${{ secrets.WECHAT_APP_ID }}"
         WECHAT_APP_SECRET: "${{ secrets.WECHAT_APP_SECRET }}"
         TIKHUB_API_KEY: "${{ secrets.TIKHUB_API_KEY }}"
         APIZERO_API_KEY: "${{ secrets.APIZERO_API_KEY }}"
+        GPU_WORKER_TOKEN: "${{ secrets.GPU_WORKER_TOKEN }}"
       run: |
         set -eu
         echo "Rendering .env from template + secrets..."

alembic/versions/081_add_gpu_lipsync_tasks.py

@@ -0,0 +1,58 @@
+"""add gpu_lipsync_tasks and gpu_workers tables for MuseTalk reverse-poll worker
+
+Revision ID: 081_add_gpu_lipsync
+Revises: 080_edit_plan_clips_atom_clip_id
+Create Date: 2026-09-18
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "081_add_gpu_lipsync"
+down_revision = "080_edit_plan_clips_atom_clip_id"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # GPU Worker 注册表
+    op.create_table(
+        "gpu_workers",
+        sa.Column("worker_id", sa.String(100), primary_key=True),
+        sa.Column("hostname", sa.String(200), nullable=False, server_default=""),
+        sa.Column("gpu_name", sa.String(200), nullable=False, server_default=""),
+        sa.Column("free_vram_mb", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("capabilities", sa.String(500), nullable=False, server_default=""),
+        sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True, index=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+    )
+
+    # GPU 口型同步任务表
+    op.create_table(
+        "gpu_lipsync_tasks",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("lipsync_job_id", sa.String(36), nullable=False, server_default="", index=True),
+        sa.Column("user_id", sa.String(36), nullable=False, server_default="", index=True),
+        sa.Column("project_id", sa.String(36), nullable=False, server_default="", index=True),
+        sa.Column("video_url", sa.Text(), nullable=False),
+        sa.Column("audio_url", sa.Text(), nullable=False),
+        sa.Column("result_url", sa.Text(), nullable=False, server_default=""),
+        sa.Column("result_duration", sa.Float(), nullable=False, server_default=sa.text("0.0")),
+        sa.Column("status", sa.String(20), nullable=False, server_default="pending", index=True),
+        sa.Column("worker_id", sa.String(100), nullable=False, server_default="", index=True),
+        sa.Column("attempt", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("error_msg", sa.Text(), nullable=False, server_default=""),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("started_at", sa.DateTime(), nullable=True),
+        sa.Column("finished_at", sa.DateTime(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True),
+    )
+    op.create_index("ix_gpu_lipsync_status_created", "gpu_lipsync_tasks", ["status", "created_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_gpu_lipsync_status_created", table_name="gpu_lipsync_tasks")
+    op.drop_table("gpu_lipsync_tasks")
+    op.drop_table("gpu_workers")

alembic/versions/082_atom_clip_ai_tags.py

@@ -0,0 +1,26 @@
+"""add ai_tags to asset_atom_clips for #1970 fragment-level AI tagging
+
+Revision ID: 082_atom_clip_ai_tags
+Revises: 081_add_gpu_lipsync
+Create Date: 2026-09-18
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "082_atom_clip_ai_tags"
+down_revision = "081_add_gpu_lipsync"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "asset_atom_clips",
+        sa.Column("ai_tags", sa.JSON(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("asset_atom_clips", "ai_tags")

apps/api/app/api/router.py

@@ -14,6 +14,7 @@ from app.api.routes.generation_cover import router as generation_cover_router
 from app.api.routes.generation_preview import router as generation_preview_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.generation_variant_plans import router as generation_variant_plans_router
+from app.api.routes.gpu_lipsync import router as gpu_lipsync_router
 from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
 from app.api.routes.internal_render import router as internal_render_router
@@ -211,3 +212,8 @@ api_router.include_router(
     prefix="/usage",
     tags=["Usage"],
 )
+api_router.include_router(
+    gpu_lipsync_router,
+    prefix="/gpu",
+    tags=["GPU Worker"],
+)

apps/api/app/api/routes/gpu_lipsync.py

@@ -0,0 +1,231 @@
+"""GPU MuseTalk Worker 反向轮询路由 — /api/v1/gpu/lipsync/*.
+
+仅面向部署在用户 RTX2060 本地的 GPU Worker 脚本，不面向前端用户。
+鉴权方式：长期 API Token（`Authorization: Bearer <GPU_WORKER_TOKEN>`），不走用户 JWT。
+
+接口：
+  POST  /api/v1/gpu/register              Worker 注册/心跳
+  GET   /api/v1/gpu/lipsync/poll          Worker 轮询拉任务（无任务返回 204）
+  POST  /api/v1/gpu/lipsync/result        Worker multipart 上传结果视频/上报失败
+  GET   /api/v1/gpu/lipsync/status/{id}   业务侧查询任务状态（内部接口，暂开放给登录用户）
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Optional
+
+import requests
+from app.core.storage import get_storage_service
+from app.dependencies import get_db_session
+from app.schemas.gpu_lipsync import (
+    GpuLipsyncPollResponse,
+    GpuLipsyncResultResponse,
+    GpuLipsyncStatusResponse,
+    GpuLipsyncTaskPayload,
+    GpuWorkerRegisterRequest,
+    GpuWorkerRegisterResponse,
+)
+from app.services.gpu_lipsync_service import GpuLipsyncService
+from fastapi import (
+    APIRouter,
+    Depends,
+    File,
+    Form,
+    HTTPException,
+    Query,
+    Request,
+    UploadFile,
+    status,
+)
+from fastapi.responses import Response
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from packages.config import get_api_settings
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# 复用 bearer scheme 抽 Token，但不校验用户 JWT
+_gpu_bearer = HTTPBearer(auto_error=False)
+
+
+def _verify_gpu_token(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_gpu_bearer),
+) -> str:
+    """校验 GPU Worker Token，返回 worker 提供的 token 串（仅用于日志，不做身份识别）.
+
+    - development 且未配置 token → 直接放行（方便本地调试）。
+    - production/staging 未配置 token → 拒绝（避免裸奔）。
+    - token 不匹配 → 401。
+    """
+    settings = get_api_settings()
+    expected = (settings.gpu_worker_token or "").strip()
+    is_dev = settings.environment == "development"
+    if not expected:
+        if is_dev:
+            return credentials.credentials if credentials else ""
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="GPU_WORKER_TOKEN not configured on server",
+        )
+    if credentials is None or credentials.scheme.lower() != "bearer":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
+    if credentials.credentials != expected:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid GPU worker token")
+    return credentials.credentials
+
+
+def _get_svc(db=Depends(get_db_session)) -> GpuLipsyncService:
+    return GpuLipsyncService(db)
+
+
+# ── POST /register — Worker 注册/心跳 ──────────────────────────────
+
+
+@router.post("/register", response_model=GpuWorkerRegisterResponse)
+def register_worker(
+    body: GpuWorkerRegisterRequest,
+    svc: GpuLipsyncService = Depends(_get_svc),
+    _token: str = Depends(_verify_gpu_token),
+):
+    svc.register_worker(
+        worker_id=body.worker_id,
+        hostname=body.hostname,
+        gpu_name=body.gpu_name,
+        free_vram_mb=body.free_vram_mb,
+        capabilities=body.capabilities,
+        task_id=body.task_id,
+    )
+    return GpuWorkerRegisterResponse(ok=True, server_time=datetime.now(UTC), message="ok")
+
+
+# ── GET /lipsync/poll — Worker 轮询拉任务 ─────────────────────────
+
+
+@router.get("/lipsync/poll")
+def poll_task(
+    worker_id: str = Query(..., min_length=1, max_length=100, description="Worker 唯一 ID"),
+    svc: GpuLipsyncService = Depends(_get_svc),
+    _token: str = Depends(_verify_gpu_token),
+):
+    task = svc.poll_task(worker_id=worker_id)
+    if task is None:
+        return Response(status_code=status.HTTP_204_NO_CONTENT)
+    payload = GpuLipsyncTaskPayload(
+        task_id=task.id,
+        video_url=getattr(task, "_signed_video_url", task.video_url),
+        audio_url=getattr(task, "_signed_audio_url", task.audio_url),
+        lipsync_job_id=task.lipsync_job_id or "",
+        user_id=task.user_id or "",
+        project_id=task.project_id or "",
+        created_at=task.created_at,
+        upload_url=getattr(task, "_signed_upload_url", ""),
+        upload_method="PUT",
+        expires_at=getattr(task, "_upload_expires_at", datetime.now(UTC)),
+    )
+    return GpuLipsyncPollResponse(task=payload)
+
+
+# ── POST /lipsync/result — Worker 上报结果（multipart） ─────────────
+
+
+@router.post("/lipsync/result", response_model=GpuLipsyncResultResponse)
+async def report_result(
+    request: Request,
+    task_id: str = Form(...),
+    worker_id: str = Form(...),
+    success: bool = Form(True),
+    duration_seconds: float = Form(0.0),
+    error_msg: str = Form(""),
+    result: Optional[UploadFile] = File(None),
+    svc: GpuLipsyncService = Depends(_get_svc),
+    _token: str = Depends(_verify_gpu_token),
+):
+    # 参数校验：
+    # - success=true + result 文件 → API 代为上传到 OSS（方便 Worker 端实现）
+    # - success=true + 无文件 → Worker 已经自己 PUT 到预签名 upload_url，直接确认
+    # - success=false → 不上传文件，错误信息通过 error_msg 传递
+    if success and result is not None:
+        # 把文件落盘到临时目录，然后 PUT 到预签名 URL
+        storage = get_storage_service()
+        result_key = svc._result_key(task_id)
+        upload_url = storage.get_upload_url(result_key, expires_seconds=3600, content_type="video/mp4")
+        try:
+            with tempfile.TemporaryDirectory(prefix="gpu_result_") as tmpdir:
+                tmp_path = Path(tmpdir) / "result.mp4"
+                content = await result.read()
+                if not content:
+                    raise HTTPException(status_code=400, detail="上传的 result 文件为空")
+                tmp_path.write_bytes(content)
+                headers = {"Content-Type": "video/mp4"}
+                with open(tmp_path, "rb") as f:
+                    resp = requests.put(upload_url, data=f, headers=headers, timeout=300)
+                if resp.status_code >= 400:
+                    logger.error(
+                        "上传 GPU 结果到 OSS 失败: status=%d body=%s",
+                        resp.status_code,
+                        resp.text[:500],
+                    )
+                    raise HTTPException(
+                        status_code=502,
+                        detail=f"上传结果视频到 OSS 失败 (HTTP {resp.status_code})",
+                    )
+        except HTTPException:
+            raise
+        except Exception as exc:
+            logger.exception("上传 GPU 结果视频异常: %s", exc)
+            raise HTTPException(status_code=500, detail=f"上传结果视频异常: {exc}") from exc
+    elif not success:
+        # 失败时忽略 result 文件（即便传了也没用）
+        pass
+    # 其他情况：success=true 且无文件 → Worker 已自行 PUT 到预签名 URL，直接标记完成
+
+    try:
+        task = svc.report_result(
+            task_id=task_id,
+            worker_id=worker_id,
+            success=success,
+            duration_seconds=duration_seconds,
+            error_msg=error_msg,
+        )
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    return GpuLipsyncResultResponse(
+        ok=True,
+        task_id=task.id,
+        status=task.status,
+        message="ok",
+    )
+
+
+# ── GET /lipsync/status/{task_id} — 业务侧查询状态 ─────────────────
+# 说明：此接口会被 lipsync_service 内部在业务流程里直接读 DB，不通过 HTTP。
+# 但仍暴露一个简单查询接口，方便调试和前端轮询（如后续需要）。暂不做用户权限校验，
+# task_id 本身是 UUID，不可枚举。
+
+
+@router.get("/lipsync/status/{task_id}", response_model=GpuLipsyncStatusResponse)
+def get_task_status(
+    task_id: str,
+    svc: GpuLipsyncService = Depends(_get_svc),
+):
+    task = svc.get_task(task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="task not found")
+    return GpuLipsyncStatusResponse(
+        task_id=task.id,
+        status=task.status,
+        result_url=task.result_url,
+        result_duration=task.result_duration,
+        error_msg=task.error_msg,
+        worker_id=task.worker_id,
+        attempt=task.attempt,
+        created_at=task.created_at,
+        started_at=task.started_at,
+        finished_at=task.finished_at,
+    )

apps/api/app/schemas/gpu_lipsync.py

@@ -0,0 +1,111 @@
+"""GPU MuseTalk 反向轮询 API Schema 定义.
+
+面向部署在用户 RTX2060 本地的 GPU Worker 脚本，不面向前端用户。
+Worker 用长期 GPU_WORKER_TOKEN 鉴权（不是用户 JWT）。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+# ── Worker 注册/心跳 ──────────────────────────────────────────────
+
+
+class GpuWorkerRegisterRequest(BaseModel):
+    """Worker 启动/心跳时上报自身信息."""
+
+    worker_id: str = Field(..., min_length=1, max_length=100, description="Worker 唯一 ID（机器名+UUID 等）")
+    hostname: str = Field("", max_length=200, description="主机名，用于运维排查")
+    gpu_name: str = Field("", max_length=200, description="GPU 型号，如 'NVIDIA GeForce RTX 2060'")
+    free_vram_mb: int = Field(0, ge=0, description="当前空闲显存（MB）")
+    capabilities: str = Field("musetalk", max_length=500, description="能力列表，逗号分隔，如 'musetalk'")
+    task_id: Optional[str] = Field(
+        None,
+        max_length=64,
+        description=(
+            "当前正在处理的任务 ID。Worker 推理期间定期心跳时携带，"
+            "服务端同步刷新该任务 last_heartbeat_at，防止长推理被误判超时；空闲时不传"
+        ),
+    )
+
+
+class GpuWorkerRegisterResponse(BaseModel):
+    ok: bool = True
+    server_time: datetime
+    message: str = "ok"
+
+
+# ── 轮询任务 ────────────────────────────────────────────────────
+
+
+class GpuLipsyncTaskPayload(BaseModel):
+    """下发给 Worker 的任务载荷（含预签名下载 URL）."""
+
+    task_id: str
+    video_url: str = Field(..., description="人物视频预签名下载 URL（GET）")
+    audio_url: str = Field(..., description="驱动音频预签名下载 URL（GET）")
+    lipsync_job_id: str = ""
+    user_id: str = ""
+    project_id: str = ""
+    created_at: datetime
+    upload_url: str = Field(..., description="结果视频预签名上传 URL（PUT, video/mp4）")
+    upload_method: str = Field("PUT", description="上传方式，目前只支持 PUT")
+    expires_at: datetime
+
+
+class GpuLipsyncPollResponse(BaseModel):
+    """Worker poll 的返回：200 带任务，204 无任务."""
+
+    task: Optional[GpuLipsyncTaskPayload] = None
+
+
+# ── Worker 上报结果 ──────────────────────────────────────────────
+
+
+class GpuLipsyncResultRequest(BaseModel):
+    """Worker 通过 multipart 上传结果时携带的字段（非文件字段）."""
+
+    task_id: str = Field(..., min_length=1, max_length=64)
+    worker_id: str = Field(..., min_length=1, max_length=100)
+    success: bool = Field(True, description="true=成功（此时必须上传 result 视频文件）；false=失败")
+    duration_seconds: float = Field(0.0, ge=0, description="合成后视频时长（秒），成功时应填入")
+    error_msg: str = Field("", max_length=2000, description="失败原因，success=false 时必填")
+
+
+class GpuLipsyncResultResponse(BaseModel):
+    ok: bool = True
+    task_id: str
+    status: str  # done / failed
+    message: str = "ok"
+
+
+# ── 业务侧查询任务状态 ────────────────────────────────────────────
+
+
+class GpuLipsyncStatusResponse(BaseModel):
+    task_id: str
+    status: str
+    result_url: str = ""
+    result_duration: float = 0.0
+    error_msg: str = ""
+    worker_id: str = ""
+    attempt: int = 0
+    created_at: datetime
+    started_at: Optional[datetime] = None
+    finished_at: Optional[datetime] = None
+
+
+# ── 创建任务（内部服务调用） ──────────────────────────────────────
+
+
+class GpuLipsyncCreateRequest(BaseModel):
+    """服务层内部创建 GPU 任务用（不通过 HTTP 暴露给 Worker/前端）."""
+
+    video_url: str  # 已可访问的 OSS key 或公网 URL（API 侧会转预签名）
+    audio_url: str
+    lipsync_job_id: str = ""
+    user_id: str = ""
+    project_id: str = ""

apps/api/app/services/gpu_lipsync_service.py

@@ -0,0 +1,382 @@
+"""GPU MuseTalk 口型同步服务 — 反向轮询模式.
+
+职责：
+1. 创建任务（由 lipsync 业务流程调用），为输入/输出生成预签名 URL，任务入队；
+2. Worker 心跳注册（register）：登记/刷新 worker 状态；
+3. Worker 轮询拉任务（poll）：原子地 CLAIM 一条 pending 任务，返回预签名 URL；
+4. Worker 上报结果（report_result）：标记 done/failed，失败可重试；
+5. 业务侧查询状态（get_status）。
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import UTC, datetime, timedelta
+from typing import Optional
+
+from app.core.storage import get_storage_service
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel, GpuWorkerModel
+from packages.config import get_api_settings
+
+logger = logging.getLogger(__name__)
+
+# 任务在 processing 超过此时长仍未完成 → 超时回退 pending 或置 failed
+MAX_ATTEMPTS = 3
+
+
+class GpuLipsyncService:
+    """GPU 口型同步服务（无状态方法，每次调用从 DI 拿 db/storage）."""
+
+    RESULT_PREFIX = "gpu-lipsync/results/"
+    INPUT_SIGN_EXPIRES_PAD = 600  # 输入预签名 URL 在任务超时基础上再加 10min 余量
+
+    # ── 公共入口 ────────────────────────────────────────────────────
+
+    def __init__(self, db: Session):
+        self.db = db
+        self.settings = get_api_settings()
+        self.storage = get_storage_service()
+
+    # ── Worker 注册/心跳 ────────────────────────────────────────────
+
+    def register_worker(
+        self,
+        worker_id: str,
+        hostname: str = "",
+        gpu_name: str = "",
+        free_vram_mb: int = 0,
+        capabilities: str = "musetalk",
+        task_id: Optional[str] = None,
+    ) -> GpuWorkerModel:
+        """Worker 注册/心跳。
+
+        task_id 非空时（Worker 推理期间的任务级心跳），同步把对应 processing
+        任务的 last_heartbeat_at 续到当前时间，使长推理不会被
+        ``_recover_timed_out_tasks`` 误回退。任务已结束 / 不属于该 worker
+        （如已被超时回收重新派发）时忽略，不报错。
+        """
+        now = datetime.now(UTC)
+        worker = self.db.query(GpuWorkerModel).filter(GpuWorkerModel.worker_id == worker_id).one_or_none()
+        if worker is None:
+            worker = GpuWorkerModel(
+                worker_id=worker_id,
+                hostname=hostname,
+                gpu_name=gpu_name,
+                free_vram_mb=free_vram_mb,
+                capabilities=capabilities,
+                last_heartbeat_at=now,
+                created_at=now,
+            )
+            self.db.add(worker)
+        else:
+            worker.hostname = hostname or worker.hostname
+            worker.gpu_name = gpu_name or worker.gpu_name
+            worker.free_vram_mb = free_vram_mb
+            worker.capabilities = capabilities or worker.capabilities
+            worker.last_heartbeat_at = now
+        if task_id:
+            self._touch_task_heartbeat(task_id, worker_id, now)
+        self.db.commit()
+        return worker
+
+    # ── 轮询拉任务（Worker 调用） ──────────────────────────────────
+
+    def poll_task(self, worker_id: str) -> Optional[GpuLipsyncTaskModel]:
+        """原子地认领一条最早的 pending 任务，返回给 worker；无任务返回 None.
+
+        同时会：
+        - 把 processing 状态且真正超时（任务心跳停滞超过
+          gpu_task_timeout_seconds；Worker 推理期会通过 register(task_id=...)
+          续心跳，长推理不会误判）的任务回退为 pending（attempt++，超过
+          MAX_ATTEMPTS 置 failed），让其它 worker 认领。
+        - 刷新 worker 心跳。
+        """
+        now = datetime.now(UTC)
+        self._recover_timed_out_tasks(now)
+        # 更新 worker 心跳
+        self._touch_worker(worker_id, now)
+
+        # 选一条最早 pending 任务（FOR UPDATE SKIP LOCKED 语义：简单起见先查再锁状态）
+        task = (
+            self.db.query(GpuLipsyncTaskModel)
+            .filter(GpuLipsyncTaskModel.status == "pending")
+            .order_by(GpuLipsyncTaskModel.created_at.asc())
+            .first()
+        )
+        if task is None:
+            self.db.commit()
+            return None
+
+        # 原子 claim：用 UPDATE WHERE status=pending 避免并发
+        upd_rows = (
+            self.db.query(GpuLipsyncTaskModel)
+            .filter(
+                GpuLipsyncTaskModel.id == task.id,
+                GpuLipsyncTaskModel.status == "pending",
+            )
+            .update(
+                {
+                    GpuLipsyncTaskModel.status: "processing",
+                    GpuLipsyncTaskModel.worker_id: worker_id,
+                    GpuLipsyncTaskModel.started_at: now,
+                    GpuLipsyncTaskModel.last_heartbeat_at: now,
+                    GpuLipsyncTaskModel.attempt: GpuLipsyncTaskModel.attempt + 1,
+                    GpuLipsyncTaskModel.updated_at: now,
+                },
+                synchronize_session=False,
+            )
+        )
+        self.db.commit()
+        if upd_rows == 0:
+            # 被其它 worker 抢先了
+            return None
+        self.db.refresh(task)
+        # 生成预签名输入/输出 URL（在 claim 时动态生成，避免长时间过期）
+        expires = self.settings.gpu_task_timeout_seconds + self.INPUT_SIGN_EXPIRES_PAD
+        task._signed_video_url = self.storage.get_download_url(task.video_url, expires_seconds=expires)
+        task._signed_audio_url = self.storage.get_download_url(task.audio_url, expires_seconds=expires)
+        task._signed_upload_url = self.storage.get_upload_url(
+            self._result_key(task.id),
+            expires_seconds=expires,
+            content_type="video/mp4",
+        )
+        task._upload_expires_at = now + timedelta(seconds=expires)
+        return task
+
+    # ── 上报结果 ──────────────────────────────────────────────────
+
+    def report_result(
+        self,
+        task_id: str,
+        worker_id: str,
+        success: bool,
+        duration_seconds: float = 0.0,
+        error_msg: str = "",
+    ) -> GpuLipsyncTaskModel:
+        task = self.db.get(GpuLipsyncTaskModel, task_id)
+        if task is None:
+            raise KeyError(f"task {task_id} not found")
+        now = datetime.now(UTC)
+        if success:
+            task.status = "done"
+            task.result_url = self._result_key(task_id)
+            task.result_duration = duration_seconds or 0.0
+            task.error_msg = ""
+            task.finished_at = now
+        else:
+            # 失败：若仍可重试（已尝试次数 < MAX_ATTEMPTS）→ 回退 pending；否则 → failed
+            if task.attempt < MAX_ATTEMPTS:
+                task.status = "pending"
+                task.worker_id = ""
+                task.started_at = None
+                task.error_msg = error_msg[:2000]
+                logger.warning(
+                    "GPU 任务 %s 在 worker %s 上失败，回退 pending 等待重试（attempt=%d）: %s",
+                    task_id,
+                    worker_id,
+                    task.attempt,
+                    error_msg[:200],
+                )
+            else:
+                task.status = "failed"
+                task.error_msg = error_msg[:2000]
+                task.finished_at = now
+                logger.error(
+                    "GPU 任务 %s 失败达到最大重试次数 %d，置为 failed: %s",
+                    task_id,
+                    MAX_ATTEMPTS,
+                    error_msg[:200],
+                )
+        task.updated_at = now
+        task.last_heartbeat_at = now
+        self._touch_worker(worker_id, now)
+        self.db.commit()
+        self.db.refresh(task)
+        return task
+
+    # ── 业务侧查询 ────────────────────────────────────────────────
+
+    def get_task(self, task_id: str) -> Optional[GpuLipsyncTaskModel]:
+        return self.db.get(GpuLipsyncTaskModel, task_id)
+
+    def get_by_lipsync_job(self, lipsync_job_id: str) -> Optional[GpuLipsyncTaskModel]:
+        return (
+            self.db.query(GpuLipsyncTaskModel)
+            .filter(GpuLipsyncTaskModel.lipsync_job_id == lipsync_job_id)
+            .order_by(GpuLipsyncTaskModel.created_at.desc())
+            .first()
+        )
+
+    # ── 创建任务（业务侧调用） ────────────────────────────────────
+
+    def create_task(
+        self,
+        video_url: str,
+        audio_url: str,
+        lipsync_job_id: str = "",
+        user_id: str = "",
+        project_id: str = "",
+    ) -> GpuLipsyncTaskModel:
+        task_id = str(uuid.uuid4())
+        now = datetime.now(UTC)
+        task = GpuLipsyncTaskModel(
+            id=task_id,
+            lipsync_job_id=lipsync_job_id,
+            user_id=user_id,
+            project_id=project_id,
+            video_url=video_url,
+            audio_url=audio_url,
+            status="pending",
+            attempt=0,
+            created_at=now,
+            updated_at=now,
+        )
+        self.db.add(task)
+        self.db.commit()
+        self.db.refresh(task)
+        logger.info(
+            "创建 GPU 口型任务 %s (lipsync_job=%s, user=%s)",
+            task_id,
+            lipsync_job_id,
+            user_id,
+        )
+        return task
+
+    # ── 内部辅助 ──────────────────────────────────────────────────
+
+    def _result_key(self, task_id: str) -> str:
+        return f"{self.RESULT_PREFIX}{task_id}.mp4"
+
+    def _touch_task_heartbeat(self, task_id: str, worker_id: str, now: datetime) -> None:
+        """Worker 推理期间的任务级心跳：只刷新属于该 worker 且仍在 processing 的任务。
+
+        任务不存在 / 已被超时回收重新派发 / 已完成 → 静默忽略（此时旧 worker 的
+        结果上报会被结果接口按最终态处理）。
+        """
+        task = self.db.get(GpuLipsyncTaskModel, task_id)
+        if task is None:
+            return
+        if task.status != "processing" or task.worker_id != worker_id:
+            logger.info(
+                "忽略过期任务心跳 task=%s worker=%s（status=%s owner=%s）",
+                task_id,
+                worker_id,
+                task.status,
+                task.worker_id,
+            )
+            return
+        task.last_heartbeat_at = now
+        task.updated_at = now
+        self.db.flush()
+
+    def _touch_worker(self, worker_id: str, now: datetime) -> None:
+        if not worker_id:
+            return
+        worker = self.db.query(GpuWorkerModel).filter(GpuWorkerModel.worker_id == worker_id).one_or_none()
+        if worker is not None:
+            worker.last_heartbeat_at = now
+            self.db.flush()
+        else:
+            # 自注册（poll 时允许自动建一个空 worker 记录，运维可见）
+            worker = GpuWorkerModel(
+                worker_id=worker_id,
+                hostname="",
+                gpu_name="",
+                free_vram_mb=0,
+                capabilities="musetalk",
+                last_heartbeat_at=now,
+                created_at=now,
+            )
+            self.db.add(worker)
+            self.db.flush()
+
+    def _recover_timed_out_tasks(self, now: datetime) -> None:
+        """扫描 processing 状态且真正超时的任务，回退 pending 或失败。
+
+        判定只看任务自身 last_heartbeat_at：claim 时写入，Worker 推理期间通过
+        /gpu/register(task_id=...) 每 30s 续期。因此仅在 Worker 崩溃/断网
+        （任务心跳停滞超过 gpu_task_timeout_seconds）时才回收，
+        不会因 Worker 主循环忙于推理而误回退。
+        """
+        timeout = self.settings.gpu_task_timeout_seconds
+        cutoff = now - timedelta(seconds=timeout)
+        stuck_tasks = (
+            self.db.query(GpuLipsyncTaskModel)
+            .filter(
+                GpuLipsyncTaskModel.status == "processing",
+                GpuLipsyncTaskModel.last_heartbeat_at < cutoff,
+            )
+            .all()
+        )
+        for t in stuck_tasks:
+            if t.attempt >= MAX_ATTEMPTS:
+                t.status = "failed"
+                t.error_msg = f"worker 心跳超时（{timeout}s），重试次数已耗尽"
+                t.finished_at = now
+            else:
+                t.status = "pending"
+                t.worker_id = ""
+                t.started_at = None
+                t.error_msg = f"worker 心跳超时（{timeout}s），等待重试"
+                logger.warning("GPU 任务 %s 心跳超时，回退 pending（attempt=%d）", t.id, t.attempt)
+            t.updated_at = now
+        if stuck_tasks:
+            self.db.flush()
+
+    # ── 业务侧辅助 ──────────────────────────────────────────────────
+
+    def has_available_worker(self) -> bool:
+        """判断是否有 Worker 在心跳新鲜窗口内可用."""
+        stale_cutoff = datetime.now(UTC) - timedelta(seconds=self.settings.gpu_worker_stale_seconds)
+        return (
+            self.db.query(GpuWorkerModel).filter(GpuWorkerModel.last_heartbeat_at >= stale_cutoff).first() is not None
+        )
+
+    def wait_for_result(
+        self,
+        task_id: str,
+        timeout_seconds: Optional[int] = None,
+        poll_interval: Optional[float] = None,
+    ) -> Optional[GpuLipsyncTaskModel]:
+        """同步轮询等待 GPU 任务完成。
+
+        Args:
+            task_id: 任务 ID（由 create_task 返回）
+            timeout_seconds: 总超时，默认取 settings.gpu_lipsync_wait_timeout
+            poll_interval: 轮询间隔秒，默认取 settings.gpu_lipsync_poll_interval
+
+        Returns:
+            终态 task（status=done/failed）；超时返回 None（此时调用方应回退 MediaKit）。
+            等待期间会自动调用 _recover_timed_out_tasks 做超时回收。
+        """
+        import time
+
+        timeout = timeout_seconds if timeout_seconds is not None else self.settings.gpu_lipsync_wait_timeout
+        interval = poll_interval if poll_interval is not None else self.settings.gpu_lipsync_poll_interval
+        deadline = time.monotonic() + timeout
+
+        while True:
+            now = datetime.now(UTC)
+            # 顺手回收超时任务
+            try:
+                self._recover_timed_out_tasks(now)
+                self.db.commit()
+            except Exception as exc:  # noqa: BLE001 - 回收失败不阻塞主流程
+                logger.warning("wait_for_result 回收超时任务异常: %s", exc)
+                self.db.rollback()
+
+            task = self.db.get(GpuLipsyncTaskModel, task_id)
+            if task is None:
+                return None
+            if task.status == "done":
+                return task
+            if task.status == "failed":
+                return task
+            # pending/processing 继续等
+            if time.monotonic() >= deadline:
+                logger.warning("GPU 任务 %s 等待超时（%ds），回退 MediaKit", task_id, timeout)
+                return None
+            time.sleep(interval)

apps/api/app/services/lipsync_service.py

@@ -36,6 +36,7 @@ from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
 from packages.application.cosyvoice_service import CosyVoiceError
+from packages.config import get_api_settings
 from packages.domain.sentence_timings import (
     compute_sentence_timings,
     probe_audio_duration,
@@ -63,6 +64,7 @@ class LipsyncService:
         self.client = client or get_mediakit_client()
         self._cosyvoice = cosyvoice_service
         self._voice_clone_repo = voice_clone_repo
+        self.settings = get_api_settings()
 
     def _get_cosyvoice(self):
         """延迟获取 CosyVoiceService（与 tts 路由一致，含 OSS 预签名配置）."""
@@ -215,7 +217,52 @@ class LipsyncService:
         if timings:
             job.sentence_timings = timings
 
-        # 4. 签名 URL 并提交 MediaKit
+        # 4. 检查是否走 GPU 路径：开关打开 + 有可用 Worker
+        use_gpu = False
+        if self.settings.use_gpu_lipsync:
+            try:
+                from app.services.gpu_lipsync_service import GpuLipsyncService
+
+                gpu_svc = GpuLipsyncService(self.db)
+                if gpu_svc.has_available_worker():
+                    use_gpu = True
+                    logger.info("[lipsync] 检测到可用 GPU Worker，优先走 MuseTalk 本地推理: job_id=%s", job.id)
+                else:
+                    logger.info("[lipsync] GPU 开关已开但无可用 Worker（心跳过期），回退 MediaKit: job_id=%s", job.id)
+            except Exception as exc:
+                logger.warning("[lipsync] GPU 服务初始化失败，回退 MediaKit: job_id=%s err=%s", job.id, exc)
+
+        if use_gpu:
+            try:
+                gpu_task = self._submit_to_gpu(job=job, gpu_svc=gpu_svc)
+                if gpu_task is not None:
+                    # GPU 任务完成：直接把结果写入 job，标为 completed
+                    job.mediakit_task_id = ""  # GPU 路径不走 MediaKit
+                    job.status = STATUS_COMPLETED
+                    job.output_video_url = gpu_task.result_url
+                    job.output_duration = gpu_task.result_duration or 0.0
+                    job.completed_at = datetime.now(UTC)
+                    job.updated_at = datetime.now(UTC)
+                    self.db.commit()
+                    logger.info(
+                        "[lipsync] GPU MuseTalk 推理完成: job_id=%s gpu_task=%s duration=%.2f",
+                        job.id,
+                        gpu_task.id,
+                        job.output_duration,
+                    )
+                    # 转存到持久 OSS 路径（GPU 结果已在 gpu-lipsync/results/ 下，直接签短链）
+                    return
+                # wait_for_result 返回 None 表示超时/最终失败 → 继续走 MediaKit 兜底
+                logger.warning("[lipsync] GPU 任务等待超时或失败，回退 MediaKit: job_id=%s", job.id)
+                self.db.rollback()  # 回滚可能的中间状态
+            except Exception as exc:
+                logger.exception("[lipsync] GPU 路径异常，回退 MediaKit: job_id=%s err=%s", job.id, exc)
+                try:
+                    self.db.rollback()
+                except Exception:
+                    pass
+
+        # 5. 签名 URL 并提交 MediaKit（兜底路径）
         video_url = self._sign_media_url(job.video_url)
         signed_audio_url = self._sign_media_url(job.audio_url)
         job.audio_url = signed_audio_url
@@ -244,6 +291,51 @@ class LipsyncService:
             self.db.commit()
             raise
 
+    # ── GPU MuseTalk 路径 ────────────────────────────────────────────────
+
+    def _submit_to_gpu(self, *, job, gpu_svc) -> Optional[object]:
+        """创建 GPU 任务并同步等待结果。
+
+        成功返回终态 task 对象（status=done）；超时或 GPU 最终失败返回 None，
+        调用方回退 MediaKit。
+
+        注意：job.video_url / job.audio_url 可能是：
+        - 自家 OSS 存储 key（storage.is_own_url 判断，gpu_svc.create_task 内部
+          get_download_url 会自动签预签名 URL 给 Worker）
+        - 外部公网 URL（CosyVoice 临时链接等）：poll 返回时原样透传给 Worker，
+          Worker 可直接 GET 下载。
+        """
+        # 创建 GPU 任务
+        gpu_task = gpu_svc.create_task(
+            video_url=job.video_url,
+            audio_url=job.audio_url,
+            lipsync_job_id=job.id,
+            user_id=job.user_id,
+            project_id=job.project_id,
+        )
+        logger.info(
+            "[lipsync] 已创建 GPU 任务: job_id=%s gpu_task=%s",
+            job.id,
+            gpu_task.id,
+        )
+        # 同步等待 Worker 处理完成（轮询 DB）
+        final_task = gpu_svc.wait_for_result(gpu_task.id)
+        if final_task is None:
+            logger.warning("[lipsync] GPU 任务等待超时，回退 MediaKit: gpu_task=%s", gpu_task.id)
+            return None
+        if final_task.status != "done":
+            logger.warning(
+                "[lipsync] GPU 任务失败: gpu_task=%s status=%s err=%s",
+                gpu_task.id,
+                final_task.status,
+                final_task.error_msg,
+            )
+            return None
+        # result_url 是 OSS 存储 key；签一个长有效期 URL 写回 job.output_video_url
+        result_signed = self._sign_media_url(final_task.result_url)
+        final_task.result_url = result_signed or final_task.result_url
+        return final_task
+
     # ── 创建任务 ──────────────────────────────────────────────────────────
 
     def create_job(

apps/web/e2e/core-douyin-extract.spec.ts

@@ -0,0 +1,117 @@
+import { expect, test, type APIRequestContext, type Page } from "@playwright/test"
+
+const PASSWORD = "SmokePass123!"
+const apiBase = process.env.E2E_API_BASE || "/api/v1"
+const apiOrigin = apiBase.endsWith("/api/v1") ? apiBase.slice(0, -"/api/v1".length) : ""
+
+async function routeBrowserApiToTestApi(page: Page) {
+  if (!apiOrigin) return
+  await page.route("**/api/v1/**", async (route) => {
+    const sourceUrl = new URL(route.request().url())
+    const response = await route.fetch({
+      url: `${apiOrigin}${sourceUrl.pathname}${sourceUrl.search}`,
+    })
+    await route.fulfill({ response })
+  })
+}
+
+async function loginWithRetry(request: APIRequestContext, email: string, password: string) {
+  for (let i = 0; i <= 2; i++) {
+    const r = await request.post(`${apiBase}/auth/login`, { data: { email, password } })
+    if (r.status() !== 429) {
+      expect(r.ok(), `login: ${await r.text()}`).toBeTruthy()
+      return (await r.json()).access_token as string
+    }
+    console.log(`[douyin] 429 retry ${i + 1}/2`)
+    await new Promise((res) => setTimeout(res, 65000))
+  }
+  throw new Error("Login retries exhausted")
+}
+
+/**
+ * #1972 抖音文案提取冒烟
+ *
+ * 路径：文案库页面 → 点「🎬 从抖音提取」→ 粘贴分享文案 → 点「开始提取」
+ * → mock /api/v1/scripts/extract-from-douyin 返回稳定文案 → 断言「新建文案」弹窗中预填了非空文案
+ */
+test.describe("Douyin Script Extraction (#1972)", () => {
+  test("extract flow: open modal, paste link, text prefilled in create modal", async ({
+    page,
+    request,
+  }) => {
+    test.setTimeout(180_000)
+    await page.setViewportSize({ width: 1440, height: 900 })
+
+    const suffix = Math.random().toString(36).slice(2, 8)
+    const email = `e2e-douyin-${suffix}@example.com`
+    await request.post(`${apiBase}/auth/register`, {
+      data: { email, password: PASSWORD, username: `e2e_dy_${suffix}` },
+    })
+    const token = await loginWithRetry(request, email, PASSWORD)
+    const authHeader = { Authorization: `Bearer ${token}` }
+
+    const proj = await request.post(`${apiBase}/projects`, {
+      headers: authHeader,
+      data: { name: `Smoke Douyin ${suffix}` },
+    })
+    const projectId = (await proj.json()).id ?? (await proj.json()).project_id
+    await request.post(`${apiBase}/asset-libraries`, {
+      headers: authHeader,
+      data: { project_id: projectId, name: "Smoke", kind: "video" },
+    })
+
+    await page.addInitScript((t: string) => {
+      window.localStorage.setItem("access_token", t)
+      window.localStorage.setItem(
+        "auth-storage",
+        JSON.stringify({ state: { token: t, user: null } }),
+      )
+    }, token)
+    await routeBrowserApiToTestApi(page)
+
+    // Mock 抖音提取接口返回稳定文案
+    const extractedText = "大家好，今天给大家推荐一款超好用的产品，性价比非常高，快来看看吧！"
+    await page.route("**/api/v1/scripts/extract-from-douyin", (route) =>
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({ text: extractedText, duration_seconds: 15 }),
+      }),
+    )
+    // 文案列表空态
+    await page.route(
+      (url) => url.pathname.endsWith("/scripts") && !url.pathname.includes("extract-from-douyin"),
+      (route) =>
+        route.fulfill({
+          status: 200,
+          contentType: "application/json",
+          body: JSON.stringify({ items: [], total: 0, page: 1, page_size: 20 }),
+        }),
+    )
+
+    await page.goto("/app/scripts")
+    // 文案库页面加载
+    await expect(page.getByText(/文案库|文案/).first()).toBeVisible({ timeout: 30000 })
+
+    // 点「🎬 从抖音提取」按钮
+    await page.getByRole("button", { name: /从抖音提取/ }).click()
+    await expect(page.getByText("从抖音视频提取文案")).toBeVisible({ timeout: 5000 })
+
+    // 在 TextArea 粘贴"抖音分享文案"
+    const textarea = page.locator(".ant-modal textarea").first()
+    await expect(textarea).toBeVisible()
+    await textarea.fill("8.88 复制打开抖音，看看【推荐视频】https://v.douyin.com/abcDEF/")
+
+    // 点「开始提取」
+    await page.getByRole("button", { name: "开始提取" }).click()
+    await expect(page.getByText(/提取中/)).toBeVisible({ timeout: 3000 })
+
+    // 等待抖音弹窗关闭，「新建文案」弹窗打开并预填提取文案
+    await expect(page.getByText("从抖音视频提取文案")).not.toBeVisible({ timeout: 15000 })
+    await expect(page.getByText("新建文案")).toBeVisible({ timeout: 5000 })
+    const createTextarea = page.locator(".ant-modal textarea").first()
+    await expect(createTextarea).toBeVisible()
+    await expect(createTextarea).toHaveValue(new RegExp(extractedText.slice(0, 10)))
+    console.log("[douyin] Extraction flow completed ✓, text length:", extractedText.length)
+  })
+})

apps/web/e2e/core-generation.spec.ts

@@ -1,4 +1,4 @@
-import { expect, test, type APIRequestContext } from "@playwright/test"
+import { expect, test, type APIRequestContext, type Page } from "@playwright/test"
 import * as fs from "node:fs"
 import * as path from "node:path"
 import { fileURLToPath } from "node:url"
@@ -8,7 +8,8 @@ const PASSWORD = "SmokePass123!"
 const apiBase = process.env.E2E_API_BASE || "/api/v1"
 const apiOrigin = apiBase.endsWith("/api/v1") ? apiBase.slice(0, -"/api/v1".length) : ""
 
-const routeBrowserApiToTestApi = async (page: import("@playwright/test").Page) => {
+/** 将浏览器侧 /api/v1 请求路由到 Playwright request 源（支持跨域） */
+async function routeBrowserApiToTestApi(page: Page) {
   if (!apiOrigin) return
   await page.route("**/api/v1/**", async (route) => {
     const sourceUrl = new URL(route.request().url())
@@ -24,276 +25,358 @@ async function loginWithRetry(
   email: string,
   password: string,
   maxRetries = 2,
-) {
+): Promise<string> {
   for (let i = 0; i <= maxRetries; i++) {
-    const response = await request.post(`${apiBase}/auth/login`, {
-      data: { email, password },
-    })
-    if (response.status() !== 429) return response
-    console.log(`[login] 触发限流，等待 65s 后重试 (${i + 1}/${maxRetries})`)
+    const resp = await request.post(`${apiBase}/auth/login`, { data: { email, password } })
+    if (resp.status() !== 429) {
+      expect(resp.ok(), `Login should succeed: ${await resp.text()}`).toBeTruthy()
+      const data = await resp.json()
+      return data.access_token
+    }
+    console.log(`[login] 429 rate limited, retry ${i + 1}/${maxRetries} after 65s`)
     await new Promise((r) => setTimeout(r, 65000))
   }
-  return request.post(`${apiBase}/auth/login`, {
-    data: { email, password },
+  throw new Error("Login failed after retries")
+}
+
+/**
+ * 注册新用户 + 建项目/视频库/上传 sample.mp4，等素材 ready。返回 { token, projectId, libraryId, assetId }。
+ */
+async function setupFreshUser(
+  request: APIRequestContext,
+  label: string,
+): Promise<{ token: string; libraryId: string; assetId: string; suffix: string }> {
+  const suffix = Math.random().toString(36).slice(2, 8)
+  const email = `e2e-${label}-${suffix}@example.com`
+  await request.post(`${apiBase}/auth/register`, {
+    data: { email, password: PASSWORD, username: `e2e_${label}_${suffix}` },
   })
-}
+  const token = await loginWithRetry(request, email, PASSWORD)
+  const auth = { Authorization: `Bearer ${token}` }
 
-type ProjectResponse = { id: string }
-type LibraryResponse = { id: string }
-type AssetListResponse = {
-  items: Array<{
-    id: string
-    name: string
-    status: string
-  }>
-}
+  const proj = await request.post(`${apiBase}/projects`, {
+    headers: auth,
+    data: { name: `Smoke ${label} ${suffix}` },
+  })
+  expect(proj.ok(), `create project: ${await proj.text()}`).toBeTruthy()
+  const projectId = (await proj.json()).id ?? (await proj.json()).project_id
 
-test.describe("Core generation flow", () => {
-  test.describe.configure({ timeout: 360_000 })
+  const lib = await request.post(`${apiBase}/asset-libraries`, {
+    headers: auth,
+    data: { project_id: projectId, name: "Smoke", kind: "video" },
+  })
+  expect(lib.ok(), `create library: ${await lib.text()}`).toBeTruthy()
+  const libraryId = (await lib.json()).id
 
-  test("walks through wizard with count modal and starts generation", async ({ page, request }) => {
-    test.setTimeout(360_000)
-
-    await routeBrowserApiToTestApi(page)
-    const suffix = Date.now().toString(36)
-    const email = `e2e-gen-${suffix}@example.com`
-    const username = `e2e_gen_${suffix}`
-    const libraryName = `E2E Gen Lib ${suffix}`
-
-    // Register
-    const register = await request.post(`${apiBase}/auth/register`, {
-      data: { email, username, password: PASSWORD, display_name: username },
-    })
-    expect(register.status()).toBe(201)
-    const registerData = (await register.json()) as { user_id: string }
-
-    // Login
-    const login = await loginWithRetry(request, email, PASSWORD)
-    expect(login.status()).toBe(200)
-    const loginData = (await login.json()) as { access_token: string }
-    const headers = { Authorization: `Bearer ${loginData.access_token}` }
-
-    // Create project
-    const project = await request.post(`${apiBase}/projects`, {
-      headers,
-      data: { name: `E2E Gen Proj ${suffix}` },
-    })
-    expect(project.status()).toBe(200)
-    const projectData = (await project.json()) as ProjectResponse
-
-    // Create asset library
-    const library = await request.post(`${apiBase}/asset-libraries`, {
-      headers,
-      data: { project_id: projectData.id, name: libraryName, kind: "video" },
-    })
-    expect(library.status()).toBe(200)
-    const libraryData = (await library.json()) as LibraryResponse
-
-    // Upload source video
-    const sourceFileName = "e2e-gen-source.mp4"
-    const sampleVideoPath = path.join(__dirname, "fixtures", "sample.mp4")
-    const sampleVideoBuffer = fs.readFileSync(sampleVideoPath)
-    const upload = await request.post(`${apiBase}/upload`, {
-      headers,
-      multipart: {
-        project_id: projectData.id,
-        library_id: libraryData.id,
-        file: {
-          name: sourceFileName,
-          mimeType: "video/mp4",
-          buffer: sampleVideoBuffer,
-        },
+  const samplePath = path.join(__dirname, "fixtures", "sample.mp4")
+  const sampleBuf = fs.readFileSync(samplePath)
+  const up = await request.post(`${apiBase}/upload`, {
+    headers: auth,
+    multipart: {
+      project_id: projectId,
+      library_id: libraryId,
+      file: {
+        name: "sample.mp4",
+        mimeType: "video/mp4",
+        buffer: sampleBuf,
       },
-    })
-    expect(upload.status()).toBe(200)
+    },
+  })
+  expect(up.ok(), `upload sample: ${await up.text()}`).toBeTruthy()
+  const assetId = (await up.json()).asset_id
+  await expect
+    .poll(
+      async () => {
+        const r = await request.get(`${apiBase}/assets/${assetId}`, { headers: auth })
+        return r.ok() ? (await r.json()).status : "pending"
+      },
+      { timeout: 90_000, intervals: [3000, 3000, 5000] },
+    )
+    .toBe("ready")
+  return { token, libraryId, assetId, suffix }
+}
 
-    // Wait for asset to be ready
-    await expect
-      .poll(
-        async () => {
-          const assets = await request.get(`${apiBase}/assets`, {
-            headers,
-            params: { library_id: libraryData.id },
-          })
-          if (!assets.ok()) return `http_${assets.status()}`
-          const data = (await assets.json()) as AssetListResponse
-          const asset = data.items.find((a) => a.name === sourceFileName)
-          if (!asset) return "missing"
-          return asset.status
-        },
-        { timeout: 30_000, intervals: [1_000, 2_000, 3_000] },
+/**
+ * #1970 智能剪辑核心冒烟（新 5 步向导）
+ *
+ * 新流程：选择模式 → 选择素材 → 选择标题 → 确认生成 → 选择封面
+ *
+ * 两条路径：
+ *   1) 随机混剪（默认）→ Step1 下一步 → 配音选择弹窗 → Step2 选素材 → 数量弹窗
+ *      → Step3 标题 → Step4 确认生成 → 断言任务创建
+ *   2) 叙事剪辑 → Step1 切模式 → 下一步 → 文案选择弹窗 → TTS 弹窗选音色（mock 合成）
+ *      → Step2 AI 提示卡可见 + 选素材 → 数量弹窗 → Step3 标题 → Step4 确认生成
+ *      → 断言任务创建
+ */
+test.describe("Core Smart-Edit Flow (#1970)", () => {
+  test("random mode: 5-step wizard creates generation task", async ({ page, request }) => {
+    test.setTimeout(600_000)
+    await page.setViewportSize({ width: 1440, height: 1000 })
+    const { token, suffix } = await setupFreshUser(request, "random")
+    const authHeader = { Authorization: `Bearer ${token}` }
+
+    // 确保默认模板存在（智能剪辑页依赖模板）
+    const tmpls = await request.get(`${apiBase}/templates`, { headers: authHeader })
+    const tmplsJson = await tmpls.json()
+    const templates = Array.isArray(tmplsJson)
+      ? tmplsJson
+      : Array.isArray(tmplsJson.items)
+        ? tmplsJson.items
+        : []
+    expect(templates.length).toBeGreaterThan(0)
+
+    // 注入登录态 + 路由 API
+    await page.addInitScript((t: string) => {
+      window.localStorage.setItem("access_token", t)
+      window.localStorage.setItem(
+        "auth-storage",
+        JSON.stringify({ state: { token: t, user: null } }),
       )
-      .toBe("ready")
+    }, token)
+    await routeBrowserApiToTestApi(page)
 
-    // GET /templates auto-creates a default template for new users
-    const templatesResp = await request.get(`${apiBase}/templates`, { headers })
-    expect(templatesResp.status(), await templatesResp.text()).toBe(200)
-    const templatesData = (await templatesResp.json()) as {
-      items: Array<{ id: string }>
-    }
-    expect(Array.isArray(templatesData.items)).toBe(true)
-    expect(templatesData.items.length).toBeGreaterThan(0)
-    const templateId = templatesData.items[0].id
-    expect(templateId).toBeTruthy()
-
-    // Set auth in localStorage
-    await page.addInitScript(
-      ({ token, user }) => {
-        localStorage.setItem("access_token", token)
-        localStorage.setItem(
-          "auth-storage",
-          JSON.stringify({
-            state: { user, isAuthenticated: true },
-            version: 0,
+    // ── 提前 mock 配音列表（VoiceSelectModal 查询 /assets?kind=voice） ──
+    await page.route(
+      (url) => url.pathname.endsWith("/assets") && url.searchParams.get("kind") === "voice",
+      (route) =>
+        route.fulfill({
+          status: 200,
+          contentType: "application/json",
+          body: JSON.stringify({
+            items: [
+              {
+                id: `asset-voice-${suffix}`,
+                name: "测试配音.mp3",
+                file_url: "data:audio/mpeg;base64,",
+                duration: 10,
+                file_size: 1024,
+                kind: "voice",
+                status: "ready",
+              },
+            ],
+            total: 1,
           }),
-        )
-      },
-      {
-        token: loginData.access_token,
-        user: {
-          id: registerData.user_id,
-          user_id: registerData.user_id,
-          email,
-          username,
-          display_name: username,
-          is_email_verified: true,
-          email_verified: true,
-        },
-      },
+        }),
     )
 
-    // Navigate to generate page
     await page.goto("/app/generate")
     await expect(page.getByRole("heading", { name: "智能剪辑" })).toBeVisible({
-      timeout: 20_000,
+      timeout: 30000,
     })
 
-    // 5步向导：素材(1)→配音(2)→标题(3)→确认生成(4)→封面(5)
+    // ── Step 1：默认随机混剪选中，点下一步 ──────────────────────────
+    await expect(page.getByText("选择模式", { exact: true })).toBeVisible()
+    await expect(page.getByText("随机混剪")).toBeVisible()
+    await page.getByRole("button", { name: /下一步/ }).click()
 
-    // ── Step 1: 素材选择 ──
-    await expect(page.getByRole("heading", { name: /选择素材/ })).toBeVisible()
-    const librarySelect = page.locator("select").first()
-    await librarySelect.selectOption({ label: libraryName })
-    const materialCard = page.getByTestId("material-card").filter({ hasText: sourceFileName })
-    await expect(materialCard).toBeVisible({ timeout: 10_000 })
-    await materialCard.click({ position: { x: 15, y: 15 } })
-    await expect(materialCard.getByTestId("material-card-check")).toBeVisible({ timeout: 5_000 })
-    await page.getByRole("button", { name: "下一步" }).click()
+    // ── 配音选择弹窗：选第一个配音 → 确认 ─────────────────────────
+    await expect(page.getByText("🎙️ 选择配音")).toBeVisible({ timeout: 5000 })
+    await page.getByText("测试配音.mp3").first().click()
+    await page.getByRole("button", { name: "确认选择" }).click()
+    await expect(page.getByText("🎙️ 选择配音")).not.toBeVisible()
 
-    // ── 数量弹窗（PreviewCountModal） ──
-    await expect(page.getByRole("heading", { name: "要生成几个视频？" })).toBeVisible({
-      timeout: 5_000,
-    })
+    // ── Step 2：选择素材 ──────────────────────────────────────────
+    await expect(page.getByText("选择素材", { exact: true })).toBeVisible({ timeout: 10000 })
+    await page.getByTestId("material-card").first().click()
+    await page.getByRole("button", { name: /下一步/ }).click()
+
+    // ── 数量弹窗：默认 1 个 → 确认 ───────────────────────────────
+    await expect(page.getByText("要生成几个视频？")).toBeVisible({ timeout: 5000 })
     await page.getByRole("button", { name: "生成 1 个视频" }).click()
 
-    // ── Step 2: 配音（新注册用户无配音素材，跳过） ──
-    await expect(page.getByRole("heading", { name: /选择配音/ })).toBeVisible({ timeout: 15000 })
-    await page.getByRole("button", { name: "下一步" }).click()
-
-    // ── Step 3: 标题设置 ──
-    await expect(page.getByRole("heading", { name: /选择标题/ })).toBeVisible({ timeout: 15000 })
-    await page.waitForTimeout(2000)
-
-    const titleInput = page.locator(".ant-select-auto-complete input")
+    // ── Step 3：填写标题 ──────────────────────────────────────────
+    await expect(page.getByText("选择标题", { exact: true })).toBeVisible({ timeout: 10000 })
+    const titleInput = page.getByPlaceholder("输入或从标题库选择")
     await expect(titleInput).toBeVisible({ timeout: 5000 })
-    await titleInput.fill(`E2E Test ${suffix}`)
+    await titleInput.fill(`测试随机剪辑 ${suffix}`)
+    await page.getByRole("button", { name: /下一步/ }).click()
 
-    // Step 3 底部是「下一步 →」，点击进入 Step 4（确认生成）
-    await page.getByRole("button", { name: "下一步" }).click()
+    // ── Step 4：确认生成 ──────────────────────────────────────────
+    await expect(page.getByText("📋 生成配置")).toBeVisible({ timeout: 10000 })
+    await expect(page.getByText("随机混剪")).toBeVisible()
+    const confirmBtn = page.getByRole("button", { name: /确认生成视频/ })
+    await expect(confirmBtn).toBeEnabled({ timeout: 5000 })
 
-    // ── Step 4: 确认生成 ──
-    // 等待实时预览就绪（占位消失）
-    await page
-      .getByText("准备预览素材")
-      .waitFor({ state: "detached", timeout: 30_000 })
-      .catch(() => {})
+    const createTask = page.waitForResponse(
+      (r) => r.url().includes("/generation/tasks") && r.request().method() === "POST",
+      { timeout: 30000 },
+    )
+    await confirmBtn.click()
+    const taskResp = await createTask
+    expect(taskResp.ok(), `Create task: ${await taskResp.text()}`).toBeTruthy()
+    const taskId = (await taskResp.json()).id ?? (await taskResp.json()).task_id
+    console.log("[random] Generation task created:", taskId)
+    await expect(page.getByText(/正在生成|提交/)).toBeVisible({ timeout: 15000 })
+    console.log("[random] Wizard flow completed ✓")
+  })
 
-    // Step 4 底部是「✨ 确认生成视频」
-    const confirmBtn = page.locator(".xx-step-actions .xx-btn-primary").first()
-    await expect(confirmBtn).toBeVisible({ timeout: 15_000 })
+  test("narrative mode: select script + mock TTS, create generation task", async ({
+    page,
+    request,
+  }) => {
+    test.setTimeout(600_000)
+    await page.setViewportSize({ width: 1440, height: 1000 })
+    const { token, suffix } = await setupFreshUser(request, "narrative")
 
-    // 先挂 API 监听再点击
-    const generatePromise = page.waitForResponse(
-      (response) => {
-        const url = response.url()
-        const path = new URL(url).pathname
-        return response.request().method() === "POST" && path.endsWith("/generation/tasks")
-      },
-      { timeout: 30_000 },
+    await page.addInitScript((t: string) => {
+      window.localStorage.setItem("access_token", t)
+      window.localStorage.setItem(
+        "auth-storage",
+        JSON.stringify({ state: { token: t, user: null } }),
+      )
+    }, token)
+    await routeBrowserApiToTestApi(page)
+
+    // ── Mock 文案列表、音色、TTS 合成（避免真实合成） ──────────────
+    const mockScriptId = `script-mock-${suffix}`
+    const mockVoiceId = `preset-voice-${suffix}`
+    const mockJobId = `tts-job-${suffix}`
+
+    // 文案列表（ScriptSelectModal 查询 /scripts）
+    await page.route("**/api/v1/scripts**", (route) => {
+      const url = new URL(route.request().url())
+      if (url.pathname.includes("/extract-from-douyin")) {
+        route.continue()
+        return
+      }
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({
+          items: [
+            {
+              id: mockScriptId,
+              title: "测试带货文案",
+              content: "这是一段测试用的带货文案内容，用于 E2E 冒烟测试。",
+              tags: ["带货"],
+              title_category: "daihuo",
+              created_at: new Date().toISOString(),
+              updated_at: new Date().toISOString(),
+            },
+          ],
+          total: 1,
+          page: 1,
+          page_size: 200,
+        }),
+      })
+    })
+
+    // 预设音色（TtsVoiceModal 查询 GET /voices/presets）
+    await page.route("**/api/v1/voices/presets**", (route) =>
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({
+          items: [
+            {
+              voice_id: mockVoiceId,
+              name: "晓晓（女声）",
+              description: "温柔女声",
+              gender: "female",
+              language: "zh-CN",
+              preview_url: null,
+              tags: ["温柔"],
+            },
+          ],
+          total: 1,
+        }),
+      }),
     )
 
-    await confirmBtn.click()
+    // 克隆音色：空列表
+    await page.route(
+      (url) => url.pathname.endsWith("/voice-clones"),
+      (route) =>
+        route.fulfill({
+          status: 200,
+          contentType: "application/json",
+          body: JSON.stringify({ items: [] }),
+        }),
+    )
 
-    // 验证生成 API 被调用
-    const genResp = await generatePromise.catch(() => null)
-    if (!genResp) {
-      // staging 预览未就绪导致按钮校验拦截，未触发 API — 向导导航仍通过
-      console.log(
-        "[E2E] Generation API not triggered (preview not ready) — wizard navigation verified",
-      )
-    } else if (genResp.ok()) {
-      const genData = (await genResp.json()) as {
-        items: Array<{ id: string; status: string }>
-        total: number
-      }
-      expect(genData.items.length).toBeGreaterThan(0)
+    // TTS 合成：直接返回 completed 任务
+    await page.route("**/api/v1/tts/synthesize", (route) =>
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({ job_id: mockJobId, status: "queued" }),
+      }),
+    )
+    await page.route(`**/api/v1/tts/jobs/${mockJobId}/status`, (route) =>
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({
+          job_id: mockJobId,
+          status: "completed",
+          progress: 100,
+          audio_url: "data:audio/mpeg;base64,",
+          duration: 5,
+        }),
+      }),
+    )
+    await page.route(`**/api/v1/tts/jobs/${mockJobId}/save-to-library`, (route) =>
+      route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({ id: `tts-asset-${suffix}`, name: "AI合成配音" }),
+      }),
+    )
 
-      // race：渲染完成 vs 生成失败/超时
-      const downloadReady = page
-        .getByText("视频生成完成")
-        .isVisible({ timeout: 180_000 })
-        .then((v) => (v ? "completed" : null))
-      const generationFailed = page
-        .getByText(/生成失败|重新生成/)
-        .isVisible({ timeout: 180_000 })
-        .then((v) => (v ? "failed" : null))
-
-      const outcome = await Promise.any([downloadReady, generationFailed]).catch(() => "timeout")
-
-      if (outcome === "completed") {
-        await page.getByRole("button", { name: /下一步：选择封面/ }).click()
-        await expect(page.getByRole("heading", { name: /选择封面/ })).toBeVisible({
-          timeout: 30_000,
-        })
-      } else {
-        console.log(`[E2E] Video rendering ${outcome} on staging — wizard flow verified`)
-      }
-    } else {
-      console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
-    }
-
-    // 验证成品库页面加载
-    await page.goto("/app/products")
-    await expect(page).toHaveURL(/\/app\/products/)
-    await expect(page.locator(".xx-products-page")).toBeVisible({ timeout: 15_000 })
-
-    await page.unrouteAll({ behavior: "ignoreErrors" })
-  })
-
-  test("generation task API creates and lists tasks", async ({ request }) => {
-    const suffix = Date.now().toString(36)
-    const email = `e2e-gen-api-${suffix}@example.com`
-    const username = `e2e_gen_api_${suffix}`
-
-    const register = await request.post(`${apiBase}/auth/register`, {
-      data: { email, username, password: PASSWORD, display_name: username },
+    await page.goto("/app/generate")
+    await expect(page.getByRole("heading", { name: "智能剪辑" })).toBeVisible({
+      timeout: 30000,
     })
-    expect(register.status()).toBe(201)
 
-    const login = await loginWithRetry(request, email, PASSWORD)
-    expect(login.status()).toBe(200)
-    const loginData = (await login.json()) as { access_token: string }
-    const headers = { Authorization: `Bearer ${loginData.access_token}` }
+    // ── Step 1：切到叙事剪辑 → 下一步 ────────────────────────────
+    await expect(page.getByText("选择模式", { exact: true })).toBeVisible()
+    await page.getByText("叙事剪辑").click()
+    await page.getByRole("button", { name: /下一步/ }).click()
 
-    const project = await request.post(`${apiBase}/projects`, {
-      headers,
-      data: { name: `E2E API Proj ${suffix}` },
-    })
-    expect(project.status()).toBe(200)
+    // ── 文案选择弹窗：选第一条 → 确认 ─────────────────────────────
+    await expect(page.getByText("📝 选择文案")).toBeVisible({ timeout: 5000 })
+    await page.getByText("测试带货文案").first().click()
+    await page.getByRole("button", { name: "确认选择" }).click()
+    await expect(page.getByText("📝 选择文案")).not.toBeVisible()
 
-    const tasks = await request.get(`${apiBase}/tasks`, { headers })
-    expect(tasks.status()).toBe(200)
-    const tasksData = await tasks.json()
-    expect(Array.isArray(tasksData.items)).toBe(true)
+    // ── TTS 音色弹窗：选系统音色 → 合成 ─────────────────────────
+    await expect(page.getByText("🎙️ 合成配音")).toBeVisible({ timeout: 5000 })
+    await page.getByText("晓晓（女声）").first().click()
+    await page.getByRole("button", { name: "🎧 合成配音" }).click()
+    await expect(page.getByText("🎙️ 合成配音")).not.toBeVisible({ timeout: 30000 })
+
+    // ── Step 2：AI 匹配提示卡可见 + 选素材 ────────────────────────
+    await expect(page.getByText("选择素材", { exact: true })).toBeVisible({ timeout: 10000 })
+    await expect(page.getByText(/AI智能匹配/)).toBeVisible()
+    await page.getByTestId("material-card").first().click()
+    await page.getByRole("button", { name: /下一步/ }).click()
+
+    // ── 数量弹窗 ─────────────────────────────────────────────────
+    await expect(page.getByText("要生成几个视频？")).toBeVisible({ timeout: 5000 })
+    await page.getByRole("button", { name: "生成 1 个视频" }).click()
+
+    // ── Step 3：填写标题（handleScriptModalConfirm 已预填 script.title，但我们再覆盖一次） ─
+    await expect(page.getByText("选择标题", { exact: true })).toBeVisible({ timeout: 10000 })
+    const titleInput2 = page.getByPlaceholder("输入或从标题库选择")
+    await expect(titleInput2).toBeVisible({ timeout: 5000 })
+    await titleInput2.fill(`测试叙事剪辑 ${suffix}`)
+    await page.getByRole("button", { name: /下一步/ }).click()
+
+    // ── Step 4：确认生成 ──────────────────────────────────────────
+    await expect(page.getByText("📋 生成配置")).toBeVisible({ timeout: 10000 })
+    await expect(page.getByText("叙事剪辑")).toBeVisible()
+    const confirmBtn2 = page.getByRole("button", { name: /确认生成视频/ })
+    await expect(confirmBtn2).toBeEnabled({ timeout: 5000 })
+
+    const createTask2 = page.waitForResponse(
+      (r) => r.url().includes("/generation/tasks") && r.request().method() === "POST",
+      { timeout: 30000 },
+    )
+    await confirmBtn2.click()
+    const taskResp2 = await createTask2
+    expect(taskResp2.ok(), `Create task: ${await taskResp2.text()}`).toBeTruthy()
+    console.log("[narrative] Generation task created:", (await taskResp2.json()).id)
+    await expect(page.getByText(/正在生成|提交/)).toBeVisible({ timeout: 15000 })
+    console.log("[narrative] Wizard flow completed ✓")
   })
 })

apps/web/e2e/core-navigation.spec.ts

@@ -0,0 +1,105 @@
+import { expect, test, type APIRequestContext, type Page } from "@playwright/test"
+
+const PASSWORD = "SmokePass123!"
+const apiBase = process.env.E2E_API_BASE || "/api/v1"
+const apiOrigin = apiBase.endsWith("/api/v1") ? apiBase.slice(0, -"/api/v1".length) : ""
+
+async function routeBrowserApiToTestApi(page: Page) {
+  if (!apiOrigin) return
+  await page.route("**/api/v1/**", async (route) => {
+    const sourceUrl = new URL(route.request().url())
+    const response = await route.fetch({
+      url: `${apiOrigin}${sourceUrl.pathname}${sourceUrl.search}`,
+    })
+    await route.fulfill({ response })
+  })
+}
+
+async function loginWithRetry(request: APIRequestContext, email: string, password: string) {
+  for (let i = 0; i <= 2; i++) {
+    const r = await request.post(`${apiBase}/auth/login`, { data: { email, password } })
+    if (r.status() !== 429) {
+      expect(r.ok(), `login: ${await r.text()}`).toBeTruthy()
+      return (await r.json()).access_token as string
+    }
+    console.log(`[nav] 429 retry ${i + 1}/2`)
+    await new Promise((res) => setTimeout(res, 65000))
+  }
+  throw new Error("Login retries exhausted")
+}
+
+/**
+ * 核心页面导航冒烟：侧边栏主要入口能访问、文案库/配音库页面能正常加载（不出白屏/无致命 js error）
+ */
+test.describe("Core Navigation", () => {
+  let authToken: string
+
+  test.beforeAll(async ({ request }) => {
+    const suffix = Math.random().toString(36).slice(2, 8)
+    const email = `e2e-nav-${suffix}@example.com`
+    await request.post(`${apiBase}/auth/register`, {
+      data: { email, password: PASSWORD, username: `e2e_nav_${suffix}` },
+    })
+    authToken = await loginWithRetry(request, email, PASSWORD)
+    const authHeader = { Authorization: `Bearer ${authToken}` }
+    const proj = await request.post(`${apiBase}/projects`, {
+      headers: authHeader,
+      data: { name: `Smoke Nav ${suffix}` },
+    })
+    if (proj.ok()) {
+      const projectId = (await proj.json()).id ?? (await proj.json()).project_id
+      await request.post(`${apiBase}/asset-libraries`, {
+        headers: authHeader,
+        data: { project_id: projectId, name: "Nav Lib", kind: "video" },
+      })
+    }
+  })
+
+  test.beforeEach(async ({ page }) => {
+    await page.setViewportSize({ width: 1440, height: 900 })
+    await page.addInitScript((t: string) => {
+      window.localStorage.setItem("access_token", t)
+      window.localStorage.setItem(
+        "auth-storage",
+        JSON.stringify({ state: { token: t, user: null } }),
+      )
+    }, authToken)
+    await routeBrowserApiToTestApi(page)
+  })
+
+  const navCases = [
+    { path: "/app/dashboard", marker: /概览|工作台|最近/i, name: "概览" },
+    { path: "/app/generate", marker: /智能剪辑|剪辑/, name: "智能剪辑" },
+    { path: "/app/assets", marker: /视频库|素材/, name: "视频库" },
+    { path: "/app/scripts", marker: /文案/, name: "文案库" },
+    { path: "/app/voices", marker: /配音|我的音色|配音库/, name: "配音库" },
+    { path: "/app/products", marker: /成品|作品/, name: "成品库" },
+    { path: "/app/history", marker: /历史|任务/, name: "任务历史" },
+    { path: "/app/tasks", marker: /任务中心|任务列表/, name: "任务中心" },
+    { path: "/app/points", marker: /积分|我的积分/, name: "积分中心" },
+  ]
+
+  for (const c of navCases) {
+    test(`visit ${c.name} (${c.path}) loads without fatal pageerror`, async ({ page }) => {
+      const errors: Error[] = []
+      page.on("pageerror", (e) => errors.push(e))
+      await page.goto(c.path)
+      await expect(page.locator("body")).not.toBeEmpty({ timeout: 20000 })
+      // 过滤掉常见第三方/非致命错误
+      const fatal = errors.filter(
+        (e) =>
+          !/ResizeObserver|Loading chunk|network error|Failed to fetch|chunkLoadError/i.test(
+            e.message,
+          ),
+      )
+      expect(fatal, `${c.name} pageerrors: ${fatal.map((e) => e.message).join("; ")}`).toHaveLength(
+        0,
+      )
+      await expect(
+        page.getByText(c.marker).first(),
+        `${c.name} should show relevant text`,
+      ).toBeVisible({ timeout: 15000 })
+      console.log(`[nav] ${c.name} loaded ✓`)
+    })
+  }
+})

apps/worker/video_processing/render_adapter.py

@@ -493,6 +493,41 @@ class RenderAdapter:
             logger.warning("ASR 服务初始化失败，自动字幕将不可用: %s", e)
             return None
 
+    def _resolve_clip_has_text(self, clips: list[Any]) -> list[bool] | None:
+        """#1970：按源视频片段顺序解析 atom_clip.ai_tags.has_text。
+
+        顺序与 UnifiedRenderService 的「非 audio 源片段」口径一致。
+        仅当 atom_clip 存在 ai_tags 字典且 has_text 显式为 False 时标记为
+        无文字（允许 hflip）；atom_clip_id 缺失、ai_tags 未生成、has_text 为
+        true/null/非布尔值时一律按有文字处理（保守不翻转）。
+        查询失败时返回 None，渲染层回退到全保守路径。
+        """
+        video_clips = [c for c in clips if getattr(c, "clip_type", "main") != "audio"]
+        atom_ids: list[str] = []
+        seen: set[str] = set()
+        for c in video_clips:
+            atom_id = getattr(c, "atom_clip_id", "") or ""
+            if atom_id and atom_id not in seen:
+                seen.add(atom_id)
+                atom_ids.append(atom_id)
+        if not atom_ids:
+            return None
+        try:
+            from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+                SQLAlchemyAssetAtomClipRepository,
+            )
+
+            atom_clips = SQLAlchemyAssetAtomClipRepository(self._db).find_by_ids(atom_ids)
+        except Exception as exc:
+            logger.warning("[render-adapter] atom_clip ai_tags 查询失败，hflip 全量保守处理: %s", exc)
+            return None
+        has_text_map: dict[str, bool] = {}
+        for ac in atom_clips:
+            ai_tags = getattr(ac, "ai_tags", None)
+            no_text = isinstance(ai_tags, dict) and ai_tags.get("has_text") is False
+            has_text_map[ac.id] = not no_text
+        return [has_text_map.get((getattr(c, "atom_clip_id", "") or ""), True) for c in video_clips]
+
     def _do_render(
         self,
         plan: Any,
@@ -542,6 +577,7 @@ class RenderAdapter:
         )
 
         # 4. 执行统一渲染
+        clip_has_text = self._resolve_clip_has_text(clips)
         render_svc = UnifiedRenderService(
             plan=plan,
             clips=clips,
@@ -552,6 +588,7 @@ class RenderAdapter:
             bgm_path=bgm_path,
             asr_service=asr_service,
             voiceover_audio_path=voiceover_audio_path,
+            clip_has_text=clip_has_text,
         )
         result = render_svc.render()
 

apps/worker/video_processing/unified_render_service.py

@@ -155,6 +155,7 @@ class UnifiedRenderService:
         asr_service: Any = None,  # ASRService 实例，用于自动生成字幕
         bgm_path: str | None = None,  # BGM 本地文件路径
         voiceover_audio_path: str | None = None,  # 配音素材库音频本地路径
+        clip_has_text: list[bool] | None = None,  # 源视频片段是否有文字（来自 atom_clip.ai_tags.has_text）
     ):
         self.plan = plan
         self.clips = clips
@@ -167,6 +168,8 @@ class UnifiedRenderService:
         self.asr_service = asr_service
         self.bgm_path = bgm_path
         self.voiceover_audio_path = voiceover_audio_path
+        # #1970：片段级文字检测（顺序与非 audio 的源视频片段一致）；None 表示无可靠检测，保守不翻转
+        self._clip_has_text = clip_has_text
         self._transition_engine = TransitionEngine(default_duration=transition_duration)
         self._speed_engine = SpeedEngine()
         self._asr_timeline_cache: Any = None  # ASR 字幕结果缓存，避免重复调用
@@ -186,7 +189,9 @@ class UnifiedRenderService:
 
         种子 hash(generation_task_id + video_index)%10000，同一任务重渲结果一致。
         dedup_enabled=False 时返回 None，调用方不注入任何微变换。
-        P1 字幕检测：无可靠的片段文字轨道信息，hflip 一律关闭（宁可不翻转）。
+        hflip 放开（#1970）：clip_has_text 来自 atom_clip.ai_tags.has_text，
+        仅 AI 明确判定无文字的片段可参与 50% 翻转；未打标签 / has_text 为
+        true/null 或缺位时一律视为有文字，保持保守不翻转。
         """
         if self._micro_plan_loaded:
             return self._micro_plan_cache
@@ -200,11 +205,14 @@ class UnifiedRenderService:
             cfg = self.plan.config or {}
             task_id = str(cfg.get("generation_task_id", "") or "")
             video_index = int(cfg.get("video_index", 0) or 0)
+            # self._clip_has_text 顺序与非 audio 源片段一致；
+            # None（未提供检测，如内存直渲/旧任务）→ 纯函数层按全有文字保守处理；
+            # 列表短于片段数时缺位片段同样按有文字处理
             self._micro_plan_cache = build_micro_transform_plan(
                 task_id,
                 video_index,
                 clip_count,
-                clip_has_text=None,  # P1 保守策略：全部按有文字处理，不翻转
+                clip_has_text=self._clip_has_text,
                 enable_bgm_offset=bool(cfg.get("bgm")),
             )
         except Exception as e:

apps/worker/worker_app/celery_app.py

@@ -28,6 +28,10 @@ celery_app.conf.imports = (
     "worker_app.tasks.health",
     "worker_app.tasks.ingest",
     "worker_app.tasks.atom_clips",
+    # #1970 片段级 AI 标签：必须显式 import 注册，否则 worker 报
+    # "Received unregistered task of type 'worker.tag_atom_clip'"
+    "worker_app.tasks.atom_clip_tagging",
+    "worker_app.tasks.backfill_atom_clip_tags",
     "worker_app.tasks.classification",
     "worker_app.tasks.generation",
     "worker_app.tasks.voice_extraction",

apps/worker/worker_app/tasks/__init__.py

@@ -57,6 +57,14 @@ def __getattr__(name: str):
         from .atom_clips import generate_atom_clips
 
         return generate_atom_clips
+    elif name == "tag_atom_clip_task":
+        from .atom_clip_tagging import tag_atom_clip_task
+
+        return tag_atom_clip_task
+    elif name == "backfill_atom_clip_tags":
+        from .backfill_atom_clip_tags import backfill_atom_clip_tags
+
+        return backfill_atom_clip_tags
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 

apps/worker/worker_app/tasks/atom_clip_tagging.py

@@ -0,0 +1,98 @@
+"""片段级 AI 标签 Celery 任务 — #1970 智能剪辑流程重构 P2.
+
+为单个 atom_clip 调用视觉 AI 生成结构化标签，并更新到 ai_tags 字段。
+失败不阻断流程（降级为仅继承素材标签）。
+
+任务名：worker.tag_atom_clip
+"""
+
+from __future__ import annotations
+
+from celery.utils.log import get_task_logger
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.domain.atom_clip_tagger import tag_atom_clip
+from packages.shared.ai_client import get_doubao_client
+from packages.shared.mediakit_client import get_mediakit_client
+from packages.shared.storage import get_shared_storage_service
+
+logger = get_task_logger(__name__)
+
+
+@celery_app.task(name="worker.tag_atom_clip", bind=True, max_retries=2, default_retry_delay=10)
+def tag_atom_clip_task(self, atom_clip_id: str, force: bool = False) -> dict:
+    """为单个原子片段生成 AI 标签.
+
+    Args:
+        atom_clip_id: 原子片段 ID。
+        force: True 时允许覆盖只有 inherited_tags 的降级记录
+            （视觉 API 曾失败写入的占位标签，#1970）。
+            已有完整标签（含 has_text）始终跳过，保证幂等。
+
+    Returns:
+        任务结果 dict：status / clip_id / ai_tags（部分字段）。
+    """
+    db = SessionLocal()
+    try:
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db)
+        asset_repo = SQLAlchemyAssetRepository(db)
+
+        clip = atom_repo.find_by_id(atom_clip_id)
+        if clip is None:
+            return {"status": "skipped", "reason": "clip not found", "clip_id": atom_clip_id}
+
+        # 已有完整标签则跳过（幂等）；force 仅放行缺失 has_text 的降级记录
+        if clip.ai_tags is not None:
+            has_real_tags = isinstance(clip.ai_tags, dict) and "has_text" in clip.ai_tags
+            if has_real_tags or not force:
+                return {"status": "skipped", "reason": "already tagged", "clip_id": atom_clip_id}
+
+        # 获取素材信息
+        asset = asset_repo.find_by_id(clip.asset_id)
+        if asset is None:
+            return {"status": "skipped", "reason": "asset not found", "clip_id": atom_clip_id}
+
+        # 获取视频可访问 URL
+        storage = get_shared_storage_service()
+        video_url = storage.get_download_url(asset.storage_key, expires_seconds=3600)
+
+        # 初始化客户端
+        doubao_client = get_doubao_client()
+        mediakit_client = get_mediakit_client()
+
+        # 调用 tagger
+        ai_tags = tag_atom_clip(
+            clip=clip,
+            video_url=video_url,
+            doubao_client=doubao_client,
+            mediakit_client=mediakit_client,
+            storage=storage,
+        )
+
+        # 更新数据库
+        atom_repo.update_ai_tags(atom_clip_id, ai_tags)
+
+        logger.info(
+            "[atom_clip_tagging] clip_id=%s ai_tags=%s",
+            atom_clip_id,
+            {k: v for k, v in ai_tags.items() if k != "inherited_tags"},
+        )
+        return {
+            "status": "completed",
+            "clip_id": atom_clip_id,
+            "has_ai_tags": any(v for k, v in ai_tags.items() if k != "inherited_tags" and v),
+        }
+    except Exception as exc:
+        db.rollback()
+        logger.exception("[atom_clip_tagging] clip_id=%s 失败: %s", atom_clip_id, exc)
+        # 可重试异常
+        if self.request.retries < self.max_retries:
+            raise self.retry(exc=exc) from None
+        return {"status": "failed", "clip_id": atom_clip_id, "error": str(exc)}
+    finally:
+        db.close()

apps/worker/worker_app/tasks/atom_clips.py

@@ -3,6 +3,8 @@
 素材入库预处理完成（ingest 置 READY）后异步触发：
 根据素材时长和已缓存的 scdet 切换点计算原子片段并落库。
 失败不阻断素材入库主流程（atom_clips 未就绪时选片有内存兜底）。
+
+P2 增强：切片完成后自动链式触发 AI 标签任务（每个 clip 一个 tag_atom_clip 任务）。
 """
 
 from __future__ import annotations
@@ -72,6 +74,10 @@ def generate_atom_clips(asset_id: str) -> dict:
             asset_id,
             len(clips),
         )
+
+        # P2 增强：链式触发 AI 标签任务（每个 clip 一个异步任务）
+        _dispatch_tagging_tasks(clips)
+
         return {"status": "completed", "asset_id": asset_id, "clips_count": len(clips)}
     except Exception as exc:  # noqa: BLE001 - 后台任务兜底，失败不阻断主流程
         db.rollback()
@@ -79,3 +85,25 @@ def generate_atom_clips(asset_id: str) -> dict:
         return {"status": "failed", "asset_id": asset_id, "error": str(exc)}
     finally:
         db.close()
+
+
+def _dispatch_tagging_tasks(clips: list) -> None:
+    """为每个新建片段发送 AI 标签异步任务.
+
+    失败不阻断（标签任务是锦上添花，不影响核心流程）。
+    """
+    try:
+        for clip in clips:
+            celery_app.send_task(
+                "worker.tag_atom_clip",
+                args=[clip.id],
+            )
+        logger.info(
+            "[atom_clips] 已发送 %d 个 AI 标签任务",
+            len(clips),
+        )
+    except Exception as e:
+        logger.warning(
+            "[atom_clips] 发送 AI 标签任务失败（不影响切片结果）: %s",
+            e,
+        )

apps/worker/worker_app/tasks/backfill_atom_clip_tags.py

@@ -0,0 +1,106 @@
+"""批量回填 AI 标签 Celery 任务 — #1970 智能剪辑流程重构 P2.
+
+查找所有 ai_tags IS NULL 的 atom_clips，分批触发 tag_atom_clip 任务。
+可通过 API 路由触发（管理员权限）。
+
+任务名：worker.backfill_atom_clip_tags
+"""
+
+from __future__ import annotations
+
+import time
+
+from celery.utils.log import get_task_logger
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+
+logger = get_task_logger(__name__)
+
+# 默认批量参数
+DEFAULT_BATCH_SIZE = 10
+DEFAULT_BATCH_INTERVAL = 5  # 秒
+
+
+@celery_app.task(name="worker.backfill_atom_clip_tags")
+def backfill_atom_clip_tags(
+    batch_size: int = DEFAULT_BATCH_SIZE,
+    batch_interval: int = DEFAULT_BATCH_INTERVAL,
+    max_clips: int = 0,
+    force: bool = False,
+) -> dict:
+    """批量回填未打标的 atom_clips.
+
+    Args:
+        batch_size: 每批处理数量，默认 10。
+        batch_interval: 每批间隔秒数，默认 5。
+        max_clips: 最大处理总数，0 表示不限。
+        force: True 时连同只有 inherited_tags 的降级记录一起强制重打
+            （视觉 API 曾失败、DOUBAO_VISION_MODEL 修复后重跑用，#1970）。
+
+    Returns:
+        任务结果 dict：total_submitted / batches。
+    """
+    db = SessionLocal()
+    try:
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db)
+        total_submitted = 0
+        batches = 0
+
+        while True:
+            # 查找未打标的片段
+            remaining = max_clips - total_submitted if max_clips > 0 else batch_size
+            fetch_limit = min(batch_size, remaining) if max_clips > 0 else batch_size
+
+            untagged = atom_repo.find_untagged(limit=fetch_limit, include_downgraded=force)
+            if not untagged:
+                break
+
+            # 逐个发送 tag 任务
+            for clip in untagged:
+                try:
+                    celery_app.send_task(
+                        "worker.tag_atom_clip",
+                        args=[clip.id],
+                        kwargs={"force": force},
+                    )
+                    total_submitted += 1
+                except Exception as e:
+                    logger.warning(
+                        "[backfill] 提交任务失败 clip_id=%s: %s",
+                        clip.id,
+                        e,
+                    )
+
+            batches += 1
+            logger.info(
+                "[backfill] 第 %d 批完成，已提交 %d 个任务",
+                batches,
+                total_submitted,
+            )
+
+            # 检查是否达到上限
+            if max_clips > 0 and total_submitted >= max_clips:
+                break
+
+            # 批间间隔
+            time.sleep(batch_interval)
+
+        logger.info(
+            "[backfill] 回填完成: total_submitted=%d batches=%d",
+            total_submitted,
+            batches,
+        )
+        return {
+            "status": "completed",
+            "total_submitted": total_submitted,
+            "batches": batches,
+        }
+    except Exception as exc:
+        logger.exception("[backfill] 回填失败: %s", exc)
+        return {"status": "failed", "error": str(exc)}
+    finally:
+        db.close()

deploy/configs/.env.production

@@ -234,6 +234,9 @@ DOUBAO_TIMEOUT=60
 # 最大重试次数
 DOUBAO_MAX_RETRIES=2
 
+# 视觉模型 Endpoint ID（支持图片/视频理解的模型）
+DOUBAO_VISION_MODEL=${DOUBAO_VISION_MODEL}
+
 
 # ==================== 微信开放平台 OAuth（网页扫码登录）====================
 # 回调域名：xiaoxiajianji.com（微信开放平台已配置）
@@ -252,3 +255,11 @@ DOUYIN_DEBUG_ERRORS=false
 TIKHUB_API_KEY=${TIKHUB_API_KEY}
 # P2: apizero.cn（国内付费，https://apizero.cn）
 APIZERO_API_KEY=${APIZERO_API_KEY}
+
+# ==================== GPU MuseTalk Worker（反向轮询） ====================
+GPU_WORKER_TOKEN=${GPU_WORKER_TOKEN}
+GPU_TASK_TIMEOUT_SECONDS=900
+USE_GPU_LIPSYNC=false
+GPU_LIPSYNC_POLL_INTERVAL=5
+GPU_LIPSYNC_WAIT_TIMEOUT=1200
+GPU_WORKER_STALE_SECONDS=300

deploy/configs/.env.staging

@@ -251,6 +251,9 @@ DOUBAO_TIMEOUT=60
 # 最大重试次数
 DOUBAO_MAX_RETRIES=2
 
+# 视觉模型 Endpoint ID（支持图片/视频理解的模型）
+DOUBAO_VISION_MODEL=${DOUBAO_VISION_MODEL}
+
 
 # ==================== 微信开放平台 OAuth（网页扫码登录）====================
 # 回调域名：xiaoxiajianji.com（微信开放平台已配置）
@@ -269,3 +272,11 @@ DOUYIN_DEBUG_ERRORS=false
 TIKHUB_API_KEY=${TIKHUB_API_KEY}
 # P2: apizero.cn（国内付费，https://apizero.cn）
 APIZERO_API_KEY=${APIZERO_API_KEY}
+
+# ==================== GPU MuseTalk Worker（反向轮询） ====================
+GPU_WORKER_TOKEN=${GPU_WORKER_TOKEN}
+GPU_TASK_TIMEOUT_SECONDS=900
+USE_GPU_LIPSYNC=true
+GPU_LIPSYNC_POLL_INTERVAL=5
+GPU_LIPSYNC_WAIT_TIMEOUT=1200
+GPU_WORKER_STALE_SECONDS=300

deploy/gpu_worker/.env.example

@@ -0,0 +1,30 @@
+# ============================================================
+# MuseTalk GPU Worker 环境变量
+# 部署到 RTX2060 电脑后，复制为 .env 并修改值
+# ============================================================
+
+# SaaS API 基础 URL（staging / production）
+API_BASE_URL=https://staging-api.xiaoxiajianji.com
+# API_BASE_URL=https://api.xiaoxiajianji.com   # 生产
+
+# 长期 API Token，必须与服务端 GPU_WORKER_TOKEN 一致（找后端拿）
+GPU_WORKER_TOKEN=replace-with-real-token
+
+# 本机 Worker 唯一 ID（默认自动生成 hostname+MAC 后4位，可手动指定）
+# WORKER_ID=rtx2060-0193
+
+# 本地 MuseTalk 地址（默认 http://127.0.0.1:7861）
+MUSE_TALK_URL=http://127.0.0.1:7861
+
+# 轮询/心跳/超时（秒）
+POLL_INTERVAL=5
+HEARTBEAT_INTERVAL=15
+# 下载/推理/上传 HTTP 超时，需与服务端 GPU_TASK_TIMEOUT_SECONDS 对齐（默认 900）
+REQUEST_TIMEOUT=900
+
+# 单个任务本地最大重试次数（仅网络/MuseTalk 瞬时错误才重试，默认 1）
+TASK_MAX_RETRY=1
+# 推理期间任务心跳间隔（秒，独立线程，无需改动）
+TASK_HEARTBEAT_INTERVAL=30
+# 输入视频最短时长（秒），小于则直接上报失败，不调用 MuseTalk
+MIN_VIDEO_DURATION_SECONDS=3

deploy/gpu_worker/README.md

@@ -0,0 +1,100 @@
+# MuseTalk GPU Worker — 部署指南
+
+本目录包含 RTX2060 本地电脑上运行的 GPU Worker 脚本。
+Worker 采用 **反向轮询模式**：主动向 SaaS API 拉取待处理的口型同步任务 → 调用本地 MuseTalk 推理 → 把结果视频回传到 SaaS。不需要内网穿透。
+
+## 目录文件
+
+| 文件 | 作用 |
+|---|---|
+| `gpu_worker.py` | Worker 主程序（单文件，零项目代码依赖，仅依赖 `requests`） |
+| `requirements.txt` | Python 依赖（只有 `requests`） |
+| `xiaoxia-gpu-worker.service` | systemd 服务单元（开机自启、异常自动重启） |
+| `.env.example` | 环境变量样例，复制为 `.env` 后填入真实值 |
+
+## 一、环境准备
+
+1. **Python 3.10+**（Windows 建议从 python.org 安装；Linux 自带）
+2. **本地 MuseTalk 服务** 已启动在 `http://127.0.0.1:7861`，health 接口返回 `{"status":"ok","free_vram_mb":...}`
+3. **ffmpeg**（可选，用于读取输出视频时长；未装则 duration 报 0，不影响功能）
+4. 网络能访问 staging / 生产 API（`curl https://staging-api.xiaoxiajianji.com/health` 应返回 `{"status":"healthy"}`）
+
+## 二、部署步骤（Linux，推荐 systemd）
+
+```bash
+# 1. 创建部署目录
+sudo mkdir -p /opt/xiaoxia-gpu-worker
+sudo chown $USER:$USER /opt/xiaoxia-gpu-worker
+cd /opt/xiaoxia-gpu-worker
+
+# 2. 拷贝脚本和依赖
+cp /path/to/deploy/gpu_worker/{gpu_worker.py,requirements.txt,xiaoxia-gpu-worker.service,.env.example} .
+cp .env.example .env
+#    编辑 .env，填入 API_BASE_URL 和 GPU_WORKER_TOKEN
+
+# 3. 创建虚拟环境并安装依赖
+python3 -m venv venv
+./venv/bin/pip install -r requirements.txt
+
+# 4. 前台先跑一次，确认日志正常
+./venv/bin/python gpu_worker.py
+#    看到 "MuseTalk 健康检查通过" 和 "注册/心跳" 成功即可 Ctrl+C 退出
+
+# 5. 安装 systemd 服务
+sudo cp xiaoxia-gpu-worker.service /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable --now xiaoxia-gpu-worker
+
+# 6. 查看日志
+sudo journalctl -u xiaoxia-gpu-worker -f
+```
+
+## 三、部署步骤（Windows，快速测试）
+
+```bat
+:: 创建虚拟环境
+python -m venv venv
+venv\Scripts\pip install -r requirements.txt
+
+:: 复制并编辑 .env
+copy .env.example .env
+notepad .env
+
+:: 运行
+venv\Scripts\python gpu_worker.py
+```
+
+可在任务计划程序中添加开机启动项：程序选 `venv\Scripts\python.exe`，参数填 `gpu_worker.py`，起始目录填脚本所在目录。
+
+## 四、SaaS 侧配套配置
+
+SaaS 后端部署完成后需配置：
+
+1. 服务端环境变量 `GPU_WORKER_TOKEN` 设为一个随机强 Token（和 Worker `.env` 中一致）
+2. 数据库已跑迁移 `081_add_gpu_lipsync_tasks`（自动随 API 启动的 alembic upgrade head 完成）
+3. OSS bucket 中 `gpu-lipsync/results/` 路径可写（默认 bucket 已配）
+
+## 五、验证联调
+
+1. Worker 启动后日志看到 `注册/心跳` 成功
+2. 后端调用 `GpuLipsyncService.create_task(video_url=..., audio_url=...)` 放入一条测试任务
+3. Worker 在 5 秒内拉到任务，下载 → 推理 → 上传 → 上报
+4. 后端 `GET /api/v1/gpu/lipsync/status/{task_id}` 返回 `status=done`，`result_url` 非空
+
+## 六、故障排查
+
+| 现象 | 可能原因 / 排查 |
+|---|---|
+| 日志 401 `Invalid GPU worker token` | `.env` 的 `GPU_WORKER_TOKEN` 与服务端不一致 |
+| 日志 `MuseTalk 健康检查未通过` | 本地 MuseTalk 没启动，或端口不是 7861；`curl http://127.0.0.1:7861/health` 验证 |
+| 任务长时间不被拉取 | Worker 和服务端连不上；检查 API_BASE_URL 是否可达、Token 是否正确 |
+| 推理后上传 OSS 失败 | 本地出口网络被防火墙拦截 OSS 域名（oss-cn-hangzhou.aliyuncs.com） |
+| 服务端看到任务回退到 pending 重试 | 任务心跳真正超时（默认 900s）：Worker 进程崩溃/断网，或推理彻底卡死；正常长推理期间心跳线程每 30s 续期，不会回退 |
+| 日志 `MuseTalk 推理超时或连接失败` | 视频太长或显存不足；可临时调大 REQUEST_TIMEOUT（服务端 GPU_TASK_TIMEOUT_SECONDS 需同步调大），或限制输入视频时长 |
+| 日志 `视频过短（x.xxs < 3s）` | 输入视频不足 3s，MuseTalk 对短视频会 division by zero，已在本地直接上报失败；可用 MIN_VIDEO_DURATION_SECONDS 调整阈值 |
+
+## 七、安全注意事项
+
+- `.env` 包含长期 Token，文件权限设为 600（`chmod 600 .env`）
+- Token 泄露要立即在服务端更换 `GPU_WORKER_TOKEN` 并重启 Worker
+- Worker 只需要出站访问 SaaS API 和 OSS，不需要开放任何入站端口

deploy/gpu_worker/gpu_worker.py

@@ -0,0 +1,471 @@
+"""MuseTalk GPU Worker — 反向轮询模式.
+
+部署在有 RTX2060 的本地电脑上（192.168.0.193），
+主动轮询 SaaS API 拉取口型任务、调用本地 MuseTalk 推理、上传结果回 SaaS。
+
+环境变量：
+  API_BASE_URL        SaaS API 基础 URL（不含 /api/v1），如 https://staging-api.xiaoxiajianji.com
+  GPU_WORKER_TOKEN    长期 API Token（服务端 GPU_WORKER_TOKEN 需一致）
+  WORKER_ID           本机唯一 ID（默认 hostname+网卡MAC 后4位）
+  MUSE_TALK_URL       本地 MuseTalk 地址，默认 http://127.0.0.1:7861
+  POLL_INTERVAL       轮询间隔秒，默认 5
+  HEARTBEAT_INTERVAL  空闲心跳间隔秒，默认 15
+  REQUEST_TIMEOUT     HTTP 请求超时秒（下载/推理/上传统一使用），默认 900
+                      需与服务端 GPU_TASK_TIMEOUT_SECONDS（默认 900）对齐
+  TASK_MAX_RETRY      单任务本地最大重试次数（仅对瞬时错误重试），默认 1
+  TASK_HEARTBEAT_INTERVAL  推理期间任务心跳间隔秒，默认 30
+  MIN_VIDEO_DURATION_SECONDS  最短输入视频时长秒，小于则直接上报失败，默认 3
+
+用法：
+  python gpu_worker.py
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import platform
+import socket
+import sys
+import tempfile
+import threading
+import time
+import uuid
+from pathlib import Path
+from typing import Optional
+
+import requests
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("musetalk-worker")
+
+# ── 配置 ────────────────────────────────────────────────────────────
+
+
+def _env(name: str, default: str = "") -> str:
+    v = os.environ.get(name, default)
+    return v.strip() if isinstance(v, str) else default
+
+
+class Config:
+    api_base_url: str = _env("API_BASE_URL", "https://staging-api.xiaoxiajianji.com").rstrip("/")
+    gpu_worker_token: str = _env("GPU_WORKER_TOKEN")
+    muse_talk_url: str = _env("MUSE_TALK_URL", "http://127.0.0.1:7861").rstrip("/")
+    poll_interval: float = float(_env("POLL_INTERVAL", "5"))
+    heartbeat_interval: float = float(_env("HEARTBEAT_INTERVAL", "15"))
+    # #1970：RTX2060 6G 处理 720p 长视频可能 >5min；与服务端
+    # GPU_TASK_TIMEOUT_SECONDS 默认值对齐为 900，避免推理被本地/服务端先掐断。
+    request_timeout: float = float(_env("REQUEST_TIMEOUT", "900"))
+    # 本地只在网络/MuseTalk 瞬时错误时重试 1 次；服务端 MAX_ATTEMPTS=3
+    # 负责跨 worker/真正超时后的重派发，总尝试次数不再相乘放大。
+    task_max_retry: int = int(_env("TASK_MAX_RETRY", "1"))
+    # 推理期间任务心跳间隔（独立线程 POST /gpu/register 带 task_id）
+    task_heartbeat_interval: float = float(_env("TASK_HEARTBEAT_INTERVAL", "30"))
+    # 输入视频最短时长（秒）：过短（如 1s）MuseTalk 会 division by zero，
+    # 本地前置拦截，直接上报 failed，不浪费 GPU 时间
+    min_video_duration_seconds: float = float(_env("MIN_VIDEO_DURATION_SECONDS", "3"))
+    worker_id: str = _env("WORKER_ID", "")
+
+    @classmethod
+    def derived_worker_id(cls) -> str:
+        if cls.worker_id:
+            return cls.worker_id
+        # hostname + MAC 后4位 → 稳定唯一 ID
+        try:
+            mac = uuid.getnode()
+            mac_suffix = f"{mac:012x}"[-4:]
+        except Exception:
+            mac_suffix = "0000"
+        host = platform.node() or socket.gethostname() or "rtx2060"
+        return f"{host}-{mac_suffix}"
+
+
+# ── 辅助 ─────────────────────────────────────────────────────────────
+
+
+def _api_headers() -> dict[str, str]:
+    token = Config.gpu_worker_token
+    if not token:
+        logger.warning("GPU_WORKER_TOKEN 未配置，开发模式下会被服务端拒绝（生产环境必须配置）")
+    return {"Authorization": f"Bearer {token}"} if token else {}
+
+
+def _check_musetalk_health() -> tuple[bool, dict]:
+    """检查本地 MuseTalk 健康状态，返回 (ok, info)."""
+    try:
+        r = requests.get(f"{Config.muse_talk_url}/health", timeout=5)
+        if r.status_code == 200:
+            try:
+                return True, r.json()
+            except Exception:
+                return True, {}
+        return False, {"status_code": r.status_code, "body": r.text[:200]}
+    except Exception as exc:
+        return False, {"error": str(exc)}
+
+
+def _register(task_id: Optional[str] = None) -> bool:
+    """向服务端注册 / 心跳，附带 GPU 信息。
+
+    推理期间的心跳线程传 task_id：服务端会同步刷新该 processing 任务的
+    last_heartbeat_at，防止长推理被误判超时回收。
+    """
+    ok, info = _check_musetalk_health()
+    free_vram = int(info.get("free_vram_mb", 0) or 0) if isinstance(info, dict) else 0
+    gpu_name = info.get("gpu_name", "") if isinstance(info, dict) else ""
+    if not gpu_name:
+        # 尝试在 Windows 上读 nvidia-smi
+        gpu_name = _probe_gpu_name()
+    payload = {
+        "worker_id": Config.derived_worker_id(),
+        "hostname": platform.node(),
+        "gpu_name": gpu_name,
+        "free_vram_mb": free_vram,
+        "capabilities": "musetalk",
+    }
+    if task_id:
+        payload["task_id"] = task_id
+    try:
+        r = requests.post(
+            f"{Config.api_base_url}/api/v1/gpu/register",
+            json=payload,
+            headers=_api_headers(),
+            timeout=15,
+        )
+        if r.status_code == 200:
+            return True
+        logger.error("注册/心跳失败: HTTP %d body=%s", r.status_code, r.text[:300])
+        return False
+    except Exception as exc:
+        logger.error("注册/心跳异常: %s", exc)
+        return False
+
+
+def _probe_gpu_name() -> str:
+    """尽力探测 GPU 型号（不强制依赖 pynvml）."""
+    try:
+        import subprocess
+
+        out = subprocess.check_output(
+            ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"],
+            stderr=subprocess.DEVNULL,
+            timeout=5,
+        )
+        return out.decode("utf-8", errors="ignore").strip().splitlines()[0].strip()
+    except Exception:
+        return ""
+
+
+def _poll_task() -> Optional[dict]:
+    """轮询拉取一条待处理任务；无任务返回 None."""
+    try:
+        r = requests.get(
+            f"{Config.api_base_url}/api/v1/gpu/lipsync/poll",
+            params={"worker_id": Config.derived_worker_id()},
+            headers=_api_headers(),
+            timeout=30,
+        )
+        if r.status_code == 204:
+            return None
+        if r.status_code == 200:
+            data = r.json()
+            return data.get("task")
+        logger.error("poll 返回 %d: %s", r.status_code, r.text[:300])
+        return None
+    except Exception as exc:
+        logger.error("poll 异常: %s", exc)
+        return None
+
+
+def _download(url: str, path: Path) -> bool:
+    """下载文件到本地，支持预签名 URL."""
+    try:
+        with requests.get(url, stream=True, timeout=Config.request_timeout) as r:
+            if r.status_code >= 400:
+                logger.error("下载失败 HTTP %d: %s", r.status_code, url[:120])
+                return False
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with open(path, "wb") as f:
+                for chunk in r.iter_content(chunk_size=1024 * 256):
+                    if chunk:
+                        f.write(chunk)
+        return path.stat().st_size > 0
+    except Exception as exc:
+        logger.error("下载异常 %s: %s", url[:120], exc)
+        return False
+
+
+def _call_musetalk(video_path: Path, audio_path: Path, out_path: Path) -> tuple[bool, float, str, bool]:
+    """调用本地 MuseTalk /inference.
+
+    返回 (success, duration_seconds, error_msg, retryable)。
+    duration 用 ffprobe 读结果视频，失败填 0。
+    retryable 仅对瞬时错误（连接失败/超时/5xx）为 True；HTTP 4xx、结果过小
+    等确定性失败不重试，直接上报服务端（服务端 MAX_ATTEMPTS 再决定是否重派发）。
+    """
+    try:
+        with open(video_path, "rb") as vf, open(audio_path, "rb") as af:
+            files = {
+                "video": (video_path.name, vf, "video/mp4"),
+                "audio": (audio_path.name, af, "application/octet-stream"),
+            }
+            r = requests.post(
+                f"{Config.muse_talk_url}/inference",
+                files=files,
+                timeout=Config.request_timeout,
+            )
+        if r.status_code != 200:
+            retryable = r.status_code >= 500
+            return False, 0.0, f"MuseTalk HTTP {r.status_code}: {r.text[:500]}", retryable
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        out_path.write_bytes(r.content)
+        if out_path.stat().st_size < 1024:
+            # 确定性失败（推理产物异常），本地重试大概率还是坏的，不重试
+            return False, 0.0, f"MuseTalk 返回结果过小 ({out_path.stat().st_size} bytes)", False
+        duration = _probe_duration(out_path)
+        return True, duration, "", False
+    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
+        # 瞬时网络/超时错误，允许本地重试 1 次
+        return False, 0.0, f"MuseTalk 推理超时或连接失败（>{Config.request_timeout}s）", True
+    except Exception as exc:
+        return False, 0.0, f"MuseTalk 调用异常: {exc}", False
+
+
+def _probe_duration(path: Path) -> float:
+    """用 ffprobe 读视频时长（若系统装了 ffmpeg）；否则返回 0."""
+    try:
+        import subprocess
+
+        out = subprocess.check_output(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                str(path),
+            ],
+            stderr=subprocess.DEVNULL,
+            timeout=10,
+        )
+        return float(out.decode().strip() or 0)
+    except Exception:
+        return 0.0
+
+
+def _upload_result(upload_url: str, file_path: Path) -> bool:
+    """PUT 上传结果视频到预签名 URL."""
+    try:
+        with open(file_path, "rb") as f:
+            r = requests.put(
+                upload_url,
+                data=f,
+                headers={"Content-Type": "video/mp4"},
+                timeout=Config.request_timeout,
+            )
+        if r.status_code >= 400:
+            logger.error("上传结果失败 HTTP %d: %s", r.status_code, r.text[:500])
+            return False
+        return True
+    except Exception as exc:
+        logger.error("上传结果异常: %s", exc)
+        return False
+
+
+def _report_result(task_id: str, success: bool, duration: float = 0.0, error_msg: str = "") -> bool:
+    """通知服务端结果。失败时也尝试上报错误（不含视频文件）."""
+    try:
+        data = {
+            "task_id": task_id,
+            "worker_id": Config.derived_worker_id(),
+            "success": "true" if success else "false",
+            "duration_seconds": str(duration),
+            "error_msg": error_msg,
+        }
+        r = requests.post(
+            f"{Config.api_base_url}/api/v1/gpu/lipsync/result",
+            data=data,
+            headers=_api_headers(),
+            timeout=30,
+        )
+        if r.status_code != 200:
+            logger.error("上报结果失败 HTTP %d: %s", r.status_code, r.text[:300])
+            return False
+        return True
+    except Exception as exc:
+        logger.error("上报结果异常: %s", exc)
+        return False
+
+
+class TaskHeartbeat(threading.Thread):
+    """推理期间的任务心跳线程。
+
+    主循环的空闲心跳在 ``_handle_task`` 同步阻塞（下载/推理/上传最长 900s）
+    期间无法发送，服务端会因任务 last_heartbeat_at 停滞而误判超时回退 pending。
+    本线程每 task_heartbeat_interval 秒（默认 30s）POST /gpu/register 并
+    携带当前 task_id，让服务端持续续期任务心跳；任务处理结束 stop()。
+    """
+
+    def __init__(self, task_id: str, interval: float):
+        super().__init__(daemon=True, name=f"hb-{task_id[:8]}")
+        self.task_id = task_id
+        self.interval = max(5.0, interval)
+        self._stop_event = threading.Event()
+
+    def run(self) -> None:
+        # 先立即发一次，再按间隔循环（首次心跳失败不影响主流程）
+        while not self._stop_event.is_set():
+            try:
+                if _register(self.task_id):
+                    logger.debug("任务 %s 心跳已发送", self.task_id)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("任务 %s 心跳异常（忽略）: %s", self.task_id, exc)
+            self._stop_event.wait(self.interval)
+
+    def stop(self) -> None:
+        self._stop_event.set()
+
+
+def _handle_task(task: dict) -> None:
+    """处理一条任务（整个串行流程：下载→时长校验→推理→上传→上报）。"""
+    task_id = task["task_id"]
+    logger.info("开始处理任务 %s", task_id)
+    # 领取任务后立即启动任务级心跳线程，覆盖下载/推理/上报全过程
+    hb = TaskHeartbeat(task_id, Config.task_heartbeat_interval)
+    hb.start()
+    try:
+        with tempfile.TemporaryDirectory(prefix="musetalk_") as tmpdir:
+            tmp = Path(tmpdir)
+            video_path = tmp / "input.mp4"
+            audio_path = tmp / "input_audio.bin"
+            out_path = tmp / "output.mp4"
+
+            # 1. 下载
+            if not _download(task["video_url"], video_path):
+                _report_result(task_id, False, 0.0, "下载人物视频失败")
+                return
+            if not _download(task["audio_url"], audio_path):
+                _report_result(task_id, False, 0.0, "下载驱动音频失败")
+                return
+
+            # 2. 输入时长前置校验：短视频 MuseTalk 会 division by zero，
+            #    直接上报 failed，不浪费 GPU 时间。ffprobe 不可用/读失败（0.0）
+            #    时不拦截，交给 MuseTalk 处理，避免误杀。
+            video_duration = _probe_duration(video_path)
+            if video_duration and video_duration < Config.min_video_duration_seconds:
+                msg = (
+                    f"视频过短（{video_duration:.2f}s < {Config.min_video_duration_seconds:.0f}s），"
+                    "MuseTalk 无法处理"
+                )
+                logger.error("任务 %s %s", task_id, msg)
+                _report_result(task_id, False, 0.0, msg)
+                return
+
+            # 3. 推理（本地仅对瞬时错误重试）
+            success = False
+            duration = 0.0
+            err = ""
+            retryable = False
+            for attempt in range(Config.task_max_retry + 1):
+                if attempt > 0:
+                    logger.info("任务 %s 第 %d 次重试（瞬时错误）...", task_id, attempt + 1)
+                    time.sleep(2)
+                success, duration, err, retryable = _call_musetalk(video_path, audio_path, out_path)
+                if success or not retryable:
+                    break
+            if not success:
+                logger.error("任务 %s 推理失败: %s", task_id, err)
+                _report_result(task_id, False, 0.0, err)
+                return
+
+            # 4. 上报结果（multipart 同时上传文件 → API 代为 PUT 到 OSS，逻辑最稳）
+            _report_success_with_file(task_id, duration, out_path)
+    finally:
+        hb.stop()
+
+
+def _report_success_with_file(task_id: str, duration: float, file_path: Path) -> None:
+    """上报成功并 multipart 附带结果视频."""
+    try:
+        data = {
+            "task_id": task_id,
+            "worker_id": Config.derived_worker_id(),
+            "success": "true",
+            "duration_seconds": str(duration),
+            "error_msg": "",
+        }
+        with open(file_path, "rb") as f:
+            files = {"result": (f"{task_id}.mp4", f, "video/mp4")}
+            r = requests.post(
+                f"{Config.api_base_url}/api/v1/gpu/lipsync/result",
+                data=data,
+                files=files,
+                headers=_api_headers(),
+                timeout=Config.request_timeout,
+            )
+        if r.status_code != 200:
+            logger.error("上报成功结果失败 HTTP %d: %s", r.status_code, r.text[:300])
+            return
+        logger.info("任务 %s 完成，duration=%.1fs", task_id, duration)
+    except Exception as exc:
+        logger.error("上报成功结果异常: %s", exc)
+
+
+# ── 主循环 ──────────────────────────────────────────────────────────
+
+
+def main() -> int:
+    logger.info("=" * 60)
+    logger.info("MuseTalk GPU Worker 启动")
+    logger.info("  worker_id   = %s", Config.derived_worker_id())
+    logger.info("  api_base    = %s", Config.api_base_url)
+    logger.info("  muse_talk   = %s", Config.muse_talk_url)
+    logger.info("  poll        = %.1fs / heartbeat = %.1fs", Config.poll_interval, Config.heartbeat_interval)
+    logger.info("=" * 60)
+
+    if not Config.gpu_worker_token:
+        logger.warning("GPU_WORKER_TOKEN 未配置（开发模式），生产环境必须设置")
+
+    # 先检查一次 MuseTalk
+    ok, info = _check_musetalk_health()
+    if ok:
+        logger.info("MuseTalk 健康检查通过: %s", info)
+    else:
+        logger.warning("MuseTalk 健康检查未通过: %s（继续运行，等待服务可用）", info)
+
+    # 启动时立即注册
+    _register()
+    last_heartbeat = time.time()
+
+    while True:
+        try:
+            # 心跳
+            now = time.time()
+            if now - last_heartbeat >= Config.heartbeat_interval:
+                if _register():
+                    last_heartbeat = now
+
+            # 轮询任务
+            task = _poll_task()
+            if task is not None:
+                _handle_task(task)
+                # 处理完立即再 poll（不 sleep），尽可能拉满 GPU
+                continue
+
+            time.sleep(Config.poll_interval)
+        except KeyboardInterrupt:
+            logger.info("收到中断信号，退出")
+            return 0
+        except Exception as exc:
+            logger.exception("主循环异常: %s", exc)
+            time.sleep(Config.poll_interval)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

deploy/gpu_worker/requirements.txt

@@ -0,0 +1 @@
+requests>=2.31.0

deploy/gpu_worker/xiaoxia-gpu-worker.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=MuseTalk GPU Worker (xiaoxia-saas 反向轮询)
+After=network.target musetalk.service
+# 本地 MuseTalk 服务启动后再启动本 Worker；若 MuseTalk 没有 systemd 服务则删除 musetalk.service
+
+[Service]
+Type=simple
+User=%i
+WorkingDirectory=/opt/xiaoxia-gpu-worker
+# 读取环境变量（API 地址、Token、轮询间隔等）
+EnvironmentFile=/opt/xiaoxia-gpu-worker/.env
+ExecStart=/opt/xiaoxia-gpu-worker/venv/bin/python /opt/xiaoxia-gpu-worker/gpu_worker.py
+Restart=always
+RestartSec=10
+# 日志走 journal，用 journalctl -u xiaoxia-gpu-worker -f 查看
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=xiaoxia-gpu-worker
+
+[Install]
+WantedBy=multi-user.target

packages/adapters/sqlalchemy_impl/asset_atom_clip_repository.py

@@ -83,6 +83,31 @@ class SQLAlchemyAssetAtomClipRepository:
         models = query.all()
         return [self._to_domain(m) for m in models]
 
+    def update_ai_tags(self, clip_id: str, ai_tags: dict) -> bool:
+        """更新指定片段的 ai_tags 字段."""
+        count = (
+            self.session.query(AssetAtomClipModel).filter(AssetAtomClipModel.id == clip_id).update({"ai_tags": ai_tags})
+        )
+        self.session.commit()
+        return count > 0
+
+    def find_untagged(self, limit: int = 100, include_downgraded: bool = False) -> list[AssetAtomClip]:
+        """查找未完成 AI 打标的片段，用于回填.
+
+        默认仅匹配 ai_tags IS NULL；include_downgraded=True 时额外包含
+        只有 inherited_tags 的降级记录（视觉 API 失败时写入，无 has_text 字段），
+        供强制回填（#1970 force backfill）使用。
+        """
+        query = self.session.query(AssetAtomClipModel)
+        if include_downgraded:
+            # as_string() → JSON/JSONB ->> 取值；NULL 记录或缺 has_text 键
+            # （降级记录）均为 NULL，has_text 为 true/false 的完整记录被排除
+            query = query.filter(AssetAtomClipModel.ai_tags["has_text"].as_string().is_(None))
+        else:
+            query = query.filter(AssetAtomClipModel.ai_tags.is_(None))
+        models = query.order_by(AssetAtomClipModel.created_at.asc()).limit(limit).all()
+        return [self._to_domain(m) for m in models]
+
     def _to_model(self, clip: AssetAtomClip) -> AssetAtomClipModel:
         return AssetAtomClipModel(
             id=clip.id,
@@ -92,6 +117,7 @@ class SQLAlchemyAssetAtomClipRepository:
             duration=clip.duration,
             clip_index=clip.clip_index,
             tags=clip.tags,
+            ai_tags=clip.ai_tags,
             scene_change_at=clip.scene_change_at,
             is_fallback=clip.is_fallback,
             created_at=clip.created_at or datetime.now(UTC),

packages/adapters/sqlalchemy_impl/models.py

@@ -837,6 +837,7 @@ class AssetAtomClipModel(Base):
     duration = Column(Float, nullable=False)
     clip_index = Column(Integer, nullable=False)
     tags = Column(JSON, nullable=False, default=list)
+    ai_tags = Column(JSON, nullable=True, default=None)
     scene_change_at = Column(Float, nullable=True)
     is_fallback = Column(Boolean, nullable=False, default=False)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
@@ -854,3 +855,61 @@ class DailyUsageRecordModel(Base):
     usage_type = Column(String(50), nullable=False, default="free_clip")
     count = Column(Integer, nullable=False, default=0)
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+
+
+class GpuLipsyncTaskModel(Base):
+    """GPU 口型同步任务 ORM 模型 — MuseTalk 反向轮询模式.
+
+    业务侧（AI 数字人生成/lipsync 流程）提交任务后，GPU Worker 主动 poll 拉取、
+    调用本地 MuseTalk 推理、再通过 result 接口回传结果视频。
+    """
+
+    __tablename__ = "gpu_lipsync_tasks"
+
+    id = Column(String(36), primary_key=True)
+    # 业务关联（原 lipsync_job_id，方便双向查询）
+    lipsync_job_id = Column(String(36), nullable=False, default="", index=True)
+    user_id = Column(String(36), nullable=False, default="", index=True)
+    project_id = Column(String(36), nullable=False, default="", index=True)
+
+    # 输入（预签名下载 URL，由 API 侧生成）
+    video_url = Column(Text, nullable=False)
+    audio_url = Column(Text, nullable=False)
+
+    # 结果
+    result_url = Column(Text, nullable=False, default="")
+    result_duration = Column(Float, nullable=False, default=0.0)
+
+    # 任务状态
+    status = Column(
+        String(20),
+        nullable=False,
+        default="pending",
+        index=True,
+    )  # pending → processing → done / failed / timeout
+    worker_id = Column(String(100), nullable=False, default="", index=True)
+    attempt = Column(Integer, nullable=False, default=0)
+    error_msg = Column(Text, nullable=False, default="")
+
+    # 时间戳
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+    started_at = Column(DateTime, nullable=True)
+    finished_at = Column(DateTime, nullable=True)
+    updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+
+    # 心跳：worker 最近一次 poll/result 的时间，用于判定 worker 失联
+    last_heartbeat_at = Column(DateTime, nullable=True)
+
+
+class GpuWorkerModel(Base):
+    """GPU Worker 注册表 — 反向轮询模式下用于心跳与监控."""
+
+    __tablename__ = "gpu_workers"
+
+    worker_id = Column(String(100), primary_key=True)
+    hostname = Column(String(200), nullable=False, default="")
+    gpu_name = Column(String(200), nullable=False, default="")
+    free_vram_mb = Column(Integer, nullable=False, default=0)
+    capabilities = Column(String(500), nullable=False, default="")  # 逗号分隔，如 "musetalk"
+    last_heartbeat_at = Column(DateTime, nullable=True, index=True)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))

packages/config/base.py

@@ -68,6 +68,7 @@ class SharedSettings(BaseSettings):
     doubao_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
     doubao_timeout: int = 30
     doubao_max_retries: int = 2
+    doubao_vision_model: str = "doubao-1-5-vision-pro-250915"
 
     # ── MediaKit (火山引擎 AI 媒体工具) ──────────────────────────────────
     mediakit_api_key: str = ""
@@ -79,6 +80,29 @@ class SharedSettings(BaseSettings):
     # `if settings.points_enabled:` 包裹，防止未完善的扣点逻辑影响现有用户。
     points_enabled: bool = False
 
+    # ── GPU MuseTalk 反向轮询 Worker ────────────────────────────────────
+    # Worker 用这个长期 Token 鉴权（不是用户 JWT）。多 Worker 共用同一个 Token；
+    # worker_id 用于区分具体机器。生产必须配置；development 留空会跳过校验。
+    gpu_worker_token: str = ""
+    # GPU 任务超时（秒）：processing 状态超过此时长（以任务心跳为准）才回退
+    # pending / failed。#1970：RTX2060 6G 推理 720p 长视频需 5 分钟以上，300→900。
+    # Worker 推理期间每 30s 通过 /gpu/register(task_id=...) 续心跳，
+    # 只有真正超时或 Worker 明确上报 failed 才会回退。
+    gpu_task_timeout_seconds: int = 900
+    # 结果预签名 URL 有效期（秒）
+    gpu_result_url_expires: int = 3600
+    # 输入预签名 URL 有效期（秒，需留出 Worker 下载时间）
+    gpu_input_url_expires: int = 3600
+    # 业务侧是否启用 GPU 口型同步（开关）；关或无可用 Worker 时回退 MediaKit 云端
+    use_gpu_lipsync: bool = False
+    # 业务侧轮询 GPU 任务结果的间隔（秒）
+    gpu_lipsync_poll_interval: float = 5.0
+    # 业务侧等待 GPU 任务结果的总超时（秒）；超时后回退 MediaKit。
+    # 应小于等于 gpu_task_timeout_seconds（默认900s）+ 冗余，留足 Worker 下载/上传时间。
+    gpu_lipsync_wait_timeout: int = 1200
+    # 判断 Worker 可用的心跳新鲜度窗口（秒）—— last_heartbeat_at 在窗口内视为在线
+    gpu_worker_stale_seconds: int = 300
+
     @property
     def effective_database_url(self) -> str:
         """返回实际使用的数据库 URL。

packages/domain/asset_atom_clip.py

@@ -36,6 +36,7 @@ class AssetAtomClip:
     duration: float
     clip_index: int
     tags: list[str] = field(default_factory=list)
+    ai_tags: dict | None = None
     scene_change_at: float | None = None
     is_fallback: bool = False
     created_at: datetime | None = None

packages/domain/atom_clip_tagger.py

@@ -0,0 +1,292 @@
+"""片段级 AI 标签 — #1970 智能剪辑流程重构 P2.
+
+对每个 atom_clip 提取关键帧，调用豆包视觉理解 API 识别内容，
+生成结构化标签（场景、物体、动作、景别、是否有文字）。
+
+纯函数 + IO 分离设计：
+- build_vision_prompt() 返回结构化 prompt
+- parse_vision_response(text) 解析 AI 返回的 JSON 标签
+- tag_atom_clip(...) 主入口，组合帧提取 → 视觉 API → 解析标签
+
+降级策略：任何环节失败都返回 {"inherited_tags": clip.tags}，不阻断流程。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# AI 标签结构的键
+AI_TAG_KEYS = ("scene", "objects", "action", "shot", "has_text")
+
+
+def build_vision_prompt() -> str:
+    """返回结构化标签提取 prompt.
+
+    要求 AI 以 JSON 格式返回片段内容标签，包含：
+    - scene: 场景类型列表（如 "工厂", "办公室", "户外"）
+    - objects: 出现的物体列表（如 "产品", "手机", "电脑"）
+    - action: 动作类型列表（如 "演示", "说话", "操作"）
+    - shot: 景别（"特写" / "中景" / "远景" 之一）
+    - has_text: 画面中是否有显著文字（true/false）
+    """
+    return """请分析这段视频片段的关键帧，识别内容并返回 JSON 格式标签。
+
+要求返回以下 JSON 结构（严格 JSON，不要添加其他文字）：
+{
+  "scene": ["场景1", "场景2"],
+  "objects": ["物体1", "物体2"],
+  "action": ["动作1"],
+  "shot": "特写|中景|远景",
+  "has_text": true/false
+}
+
+规则：
+- scene: 场景类型，如"工厂"、"办公室"、"户外"、"商店"、"家庭"等，1-3个
+- objects: 画面中可见的主要物体，如"产品"、"手机"、"电脑"、"食品"等，1-5个
+- action: 人物或物体正在进行的动作，如"演示"、"说话"、"操作"、"展示"等，1-3个
+- shot: 景别判断，只能是"特写"、"中景"或"远景"之一
+- has_text: 画面中是否有显著可读文字（标题、字幕、标语等）
+
+请只返回 JSON，不要有其他说明文字。"""
+
+
+def parse_vision_response(text: str) -> dict:
+    """解析 AI 返回的 JSON 标签文本.
+
+    Args:
+        text: 视觉 API 返回的文本，期望是 JSON 格式。
+
+    Returns:
+        结构化标签 dict，格式如：
+        {"scene": [...], "objects": [...], "action": [...], "shot": "...", "has_text": bool}
+
+    解析失败时返回空 dict。
+    """
+    if not text or not text.strip():
+        return {}
+
+    # 尝试直接解析
+    cleaned = text.strip()
+
+    # 去除可能的 markdown 代码块包裹
+    if cleaned.startswith("```"):
+        lines = cleaned.split("\n")
+        # 去掉首尾的 ``` 行
+        start = 1
+        end = len(lines)
+        for i in range(len(lines) - 1, 0, -1):
+            if lines[i].strip().startswith("```"):
+                end = i
+                break
+        cleaned = "\n".join(lines[start:end]).strip()
+
+    try:
+        data = json.loads(cleaned)
+    except json.JSONDecodeError:
+        # 尝试从文本中提取 JSON 块
+        try:
+            start_idx = cleaned.index("{")
+            end_idx = cleaned.rindex("}") + 1
+            data = json.loads(cleaned[start_idx:end_idx])
+        except (ValueError, json.JSONDecodeError):
+            logger.warning("无法解析 AI 标签响应: %s", text[:200])
+            return {}
+
+    if not isinstance(data, dict):
+        return {}
+
+    # 验证和清洗各字段
+    result: dict[str, Any] = {}
+    for key in ("scene", "objects", "action"):
+        val = data.get(key)
+        if isinstance(val, list):
+            result[key] = [str(v).strip() for v in val if str(v).strip()]
+        elif isinstance(val, str) and val.strip():
+            result[key] = [val.strip()]
+        else:
+            result[key] = []
+
+    shot_val = data.get("shot", "")
+    if isinstance(shot_val, str) and shot_val.strip() in ("特写", "中景", "远景"):
+        result["shot"] = shot_val.strip()
+    else:
+        result["shot"] = ""
+
+    has_text_val = data.get("has_text")
+    if isinstance(has_text_val, bool):
+        result["has_text"] = has_text_val
+    elif isinstance(has_text_val, str):
+        result["has_text"] = has_text_val.lower() in ("true", "yes", "1")
+    else:
+        result["has_text"] = False
+
+    return result
+
+
+def _extract_frames_via_mediakit(
+    mediakit_client: Any,
+    video_url: str,
+    start_time: float,
+    end_time: float,
+) -> Optional[list[str]]:
+    """通过 MediaKit 提取 3 帧（首、中、尾）.
+
+    Returns:
+        图片 URL 列表（3 个），失败返回 None。
+    """
+    try:
+        frames = mediakit_client.extract_frames(
+            video_url=video_url,
+            strategy="SpecifiedTime",
+            max_frames=3,
+            poll_interval=2.0,
+            max_poll_attempts=30,
+        )
+        # MediaKit SpecifiedTime 策略可能不支持直接传时间点
+        # 如果返回结果不够 3 帧，降级到 ffmpeg
+        if frames and len(frames) >= 1:
+            urls = [f.get("image_url", "") for f in frames if f.get("image_url")]
+            if urls:
+                return urls
+    except Exception as e:
+        logger.warning("MediaKit 抽帧失败，将降级为 ffmpeg: %s", e)
+
+    return None
+
+
+def _extract_frames_via_ffmpeg(
+    video_url: str,
+    start_time: float,
+    end_time: float,
+) -> Optional[list[str]]:
+    """通过 ffmpeg 本地提取 3 帧并转为 base64.
+
+    Returns:
+        base64 data URI 列表（3 个），失败返回 None。
+    """
+    import base64
+
+    mid_time = round((start_time + end_time) / 2, 3)
+    timestamps = [round(start_time, 3), mid_time, round(end_time, 3)]
+
+    try:
+        frames_b64: list[str] = []
+        with tempfile.TemporaryDirectory() as tmpdir:
+            for i, ts in enumerate(timestamps):
+                out_path = Path(tmpdir) / f"frame_{i}.jpg"
+                cmd = [
+                    "ffmpeg",
+                    "-y",
+                    "-ss",
+                    str(ts),
+                    "-i",
+                    video_url,
+                    "-vframes",
+                    "1",
+                    "-q:v",
+                    "2",
+                    str(out_path),
+                ]
+                result = subprocess.run(
+                    cmd,
+                    capture_output=True,
+                    timeout=30,
+                )
+                if result.returncode != 0 or not out_path.exists():
+                    logger.warning("ffmpeg 抽帧失败 ts=%s: %s", ts, result.stderr[:200])
+                    continue
+
+                img_data = out_path.read_bytes()
+                b64 = base64.b64encode(img_data).decode("ascii")
+                frames_b64.append(f"data:image/jpeg;base64,{b64}")
+
+        if frames_b64:
+            return frames_b64
+    except Exception as e:
+        logger.warning("ffmpeg 抽帧异常: %s", e)
+
+    return None
+
+
+def tag_atom_clip(
+    clip: Any,
+    video_url: str,
+    doubao_client: Any,
+    mediakit_client: Any | None = None,
+    storage: Any | None = None,
+) -> dict:
+    """主入口：为单个 atom_clip 生成 AI 标签.
+
+    流程：提取帧 → 调视觉 API → 解析标签 → 返回结构化标签 dict。
+    任何环节失败返回 {"inherited_tags": clip.tags}，不阻断流程。
+
+    Args:
+        clip: AssetAtomClip 领域对象（需有 start_time, end_time, tags）。
+        video_url: 素材视频的公网可访问 URL。
+        doubao_client: DoubaoClient 实例。
+        mediakit_client: MediaKitClient 实例（可选，不可用时降级 ffmpeg）。
+        storage: SharedStorageService 实例（可选，用于获取签名 URL）。
+
+    Returns:
+        结构化标签 dict，格式如：
+        {"scene": [...], "objects": [...], "action": [...], "shot": "...",
+         "has_text": bool, "inherited_tags": [...]}
+    """
+    inherited = list(getattr(clip, "tags", []) or [])
+
+    # 检查 DoubaoClient 是否可用
+    if not getattr(doubao_client, "is_available", False):
+        logger.info("DoubaoClient 不可用，跳过 AI 标签: clip_id=%s", getattr(clip, "id", ""))
+        return {"inherited_tags": inherited}
+
+    # 提取帧图片
+    frame_urls: Optional[list[str]] = None
+    start_time = getattr(clip, "start_time", 0.0)
+    end_time = getattr(clip, "end_time", 0.0)
+
+    # 优先使用 MediaKit
+    if mediakit_client and getattr(mediakit_client, "is_available", False):
+        frame_urls = _extract_frames_via_mediakit(mediakit_client, video_url, start_time, end_time)
+
+    # MediaKit 不可用或失败 → 降级 ffmpeg
+    if not frame_urls:
+        frame_urls = _extract_frames_via_ffmpeg(video_url, start_time, end_time)
+
+    if not frame_urls:
+        logger.warning("帧提取失败，跳过 AI 标签: clip_id=%s", getattr(clip, "id", ""))
+        return {"inherited_tags": inherited}
+
+    # 调用视觉 API
+    prompt = build_vision_prompt()
+    messages = [{"role": "user", "content": prompt}]
+
+    try:
+        response_text = doubao_client.vision_completion(
+            messages=messages,
+            images=frame_urls,
+            timeout=60,
+        )
+    except Exception as e:
+        logger.warning("视觉 API 调用异常: clip_id=%s error=%s", getattr(clip, "id", ""), e)
+        return {"inherited_tags": inherited}
+
+    if not response_text:
+        logger.warning("视觉 API 返回空: clip_id=%s", getattr(clip, "id", ""))
+        return {"inherited_tags": inherited}
+
+    # 解析标签
+    ai_tags = parse_vision_response(response_text)
+    if not ai_tags:
+        logger.warning("标签解析失败: clip_id=%s response=%s", getattr(clip, "id", ""), response_text[:200])
+        return {"inherited_tags": inherited}
+
+    # 合并 inherited_tags
+    ai_tags["inherited_tags"] = inherited
+    return ai_tags

packages/domain/narrative_match.py

@@ -1,4 +1,4 @@
-"""叙事剪辑素材标签匹配 — #1970 PR3.
+"""叙事剪辑素材标签匹配 — #1970 PR3 + P2 AI 标签加权.
 
 叙事模式下，选片在现有评分（smart_match / atom_clip_selector）之前先做一层
 文案标签匹配：
@@ -8,6 +8,12 @@
 - 调用方对优先池跑现有 smart_select_assets，数量不足时用普通池补足
   （无任何匹配 → 完全降级为现有随机逻辑，行为与改造前一致）。
 
+P2 AI 标签加权（#1970 fragment-level AI tagging）：
+- 片段级 AI 标签（scene/objects/action）与文案标签做交集时权重 2.0
+- 素材级标签（tag_ids 映射名）与文案标签交集时权重 1.0
+- 综合得分 = sum(命中权重) / max(可能权重)
+- 有 AI 标签的片段命中时优先于仅素材标签命中的片段
+
 纯函数模块：标签 id→名称映射由调用方查 TagModel 后注入，不直接碰 DB。
 """
 
@@ -18,6 +24,10 @@ from typing import Any, Iterable
 # 标签归一化后仍短于此长度的标签不参与匹配（避免「的」「是」这类噪声短词）
 MIN_TAG_LEN = 2
 
+# 标签匹配权重
+AI_TAG_WEIGHT = 2.0  # AI 标签命中权重
+ASSET_TAG_WEIGHT = 1.0  # 素材标签命中权重
+
 
 def normalize_tag(tag: Any) -> str:
     """标签归一化：去空白、小写。数字/英文统一小写，中文不受影响。"""
@@ -47,19 +57,81 @@ def build_asset_tag_name_index(tag_names_by_id: dict[str, Any]) -> dict[str, set
     return index
 
 
+def _extract_ai_tag_names(ai_tags: dict) -> set[str]:
+    """从 AI 标签 dict 中提取所有标签名（scene + objects + action）.
+
+    Args:
+        ai_tags: 片段级 AI 标签 dict，如 {"scene": [...], "objects": [...], "action": [...], ...}
+
+    Returns:
+        归一化后的标签名集合。
+    """
+    names: set[str] = set()
+    for key in ("scene", "objects", "action"):
+        values = ai_tags.get(key)
+        if isinstance(values, list):
+            names |= _normalize_tags(values)
+    return names
+
+
+def _compute_ai_score(
+    asset_id: str,
+    wanted: set[str],
+    clip_ai_tags_by_asset: dict[str, list[dict]] | None,
+) -> float:
+    """计算单个素材的 AI 标签加权得分.
+
+    对该素材的所有片段 AI 标签，求各片段标签名与文案标签交集的加权总和。
+    每个片段的命中权重 = 命中数 × AI_TAG_WEIGHT。
+    最终取所有片段的最高得分（而非累加，避免片段数多的素材不公平占优）。
+
+    Args:
+        asset_id: 素材 ID。
+        wanted: 归一化后的文案标签集合。
+        clip_ai_tags_by_asset: {asset_id: [ai_tag_dict, ...]} 每个片段一个。
+
+    Returns:
+        AI 标签加权得分（≥0）。
+    """
+    if not clip_ai_tags_by_asset or not wanted:
+        return 0.0
+
+    clips = clip_ai_tags_by_asset.get(asset_id)
+    if not clips:
+        return 0.0
+
+    best_score = 0.0
+    for ai_tags in clips:
+        if not ai_tags or not isinstance(ai_tags, dict):
+            continue
+        ai_names = _extract_ai_tag_names(ai_tags)
+        hits = ai_names & wanted
+        score = len(hits) * AI_TAG_WEIGHT
+        if score > best_score:
+            best_score = score
+
+    return best_score
+
+
 def match_assets_by_script_tags(
     assets: list[Any],
     *,
     script_tags: Iterable[Any],
     tag_names_by_id: dict[str, Any] | None = None,
+    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
 ) -> tuple[list[Any], list[Any]]:
     """按文案标签把素材拆成「命中池 / 未命中池」，保持输入相对顺序。
 
+    P2 加权逻辑：
+    - AI 标签命中（scene/objects/action ∩ 文案标签）权重 2.0
+    - 素材标签命中（tag_ids 映射名 ∩ 文案标签）权重 1.0
+    - 任一权重 > 0 → 命中池，否则 → 未命中池
+
     Args:
         assets: 候选素材（domain Asset，需有 id 与 tag_ids）。
         script_tags: 文案 tags（字符串数组，名称语义）。
-        tag_names_by_id: asset_id → 素材标签名列表；素材只有 tag_ids 时由调用方
-            查 TagModel 名称后传入。为空则视为无素材命中。
+        tag_names_by_id: asset_id → 素材标签名列表。
+        clip_ai_tags_by_asset: #1970 P2 — {asset_id: [ai_tag_dict, ...]}。
 
     Returns:
         (matched, unmatched)：命中任一文案标签的素材 / 其余素材。
@@ -74,23 +146,74 @@ def match_assets_by_script_tags(
     unmatched: list[Any] = []
     for asset in assets:
         asset_id = str(getattr(asset, "id", "") or "")
+
+        # P2: AI 标签加权得分
+        ai_score = _compute_ai_score(asset_id, wanted, clip_ai_tags_by_asset)
+
+        # 素材标签得分
         names = set(name_index.get(asset_id, set()))
-        # 兼容素材自身带字符串 tags（旧链路/测试替身）
         raw_tags = getattr(asset, "tags", None)
         if raw_tags:
             names |= _normalize_tags(raw_tags)
-        if names & wanted:
+        asset_score = len(names & wanted) * ASSET_TAG_WEIGHT
+
+        # 综合得分 > 0 → 命中池
+        if ai_score > 0 or asset_score > 0:
             matched.append(asset)
         else:
             unmatched.append(asset)
     return matched, unmatched
 
 
+def compute_tag_match_score(
+    asset_id: str,
+    *,
+    script_tags: Iterable[Any],
+    tag_names_by_id: dict[str, Any] | None = None,
+    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
+) -> float:
+    """计算单个素材的标签匹配综合得分（0.0 ~ 1.0）.
+
+    综合得分 = sum(命中权重) / max(可能权重)
+    - AI 标签每命中一个 +2.0
+    - 素材标签每命中一个 +1.0
+    - max_possible = len(wanted) * (AI_TAG_WEIGHT + ASSET_TAG_WEIGHT)
+
+    Args:
+        asset_id: 素材 ID。
+        script_tags: 文案标签。
+        tag_names_by_id: 素材标签名索引。
+        clip_ai_tags_by_asset: AI 标签索引。
+
+    Returns:
+        归一化得分 0.0~1.0。
+    """
+    wanted = _normalize_tags(script_tags)
+    if not wanted:
+        return 0.0
+
+    # AI 得分
+    ai_score = _compute_ai_score(asset_id, wanted, clip_ai_tags_by_asset)
+
+    # 素材标签得分
+    name_index = build_asset_tag_name_index(tag_names_by_id or {})
+    names = name_index.get(asset_id, set())
+    asset_score = len(names & wanted) * ASSET_TAG_WEIGHT
+
+    # 归一化：最大可能得分 = 文案标签数 × (AI权重 + 素材权重)
+    max_possible = len(wanted) * (AI_TAG_WEIGHT + ASSET_TAG_WEIGHT)
+    if max_possible <= 0:
+        return 0.0
+
+    return min((ai_score + asset_score) / max_possible, 1.0)
+
+
 def pick_narrative_assets(
     assets: list[Any],
     *,
     script_tags: Iterable[Any],
     tag_names_by_id: dict[str, Any] | None = None,
+    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
     limit: int | None = None,
     rng: Any = None,
 ) -> list[Any]:
@@ -100,9 +223,13 @@ def pick_narrative_assets(
     smart_match.smart_select_assets（质量/时长/新鲜度/未使用 + 随机噪声），
     不重写评分维度。
 
+    P2 增强：有 AI 标签的片段命中时权重更高（2.0 vs 1.0），
+    命中池内部按综合标签得分排序（AI 标签命中多的排前面）。
+
     Args:
         assets: ready 视频素材候选（调用方负责状态/类型过滤）。
         script_tags / tag_names_by_id: 见 match_assets_by_script_tags。
+        clip_ai_tags_by_asset: #1970 P2 — {asset_id: [ai_tag_dict, ...]}。
         limit: 需要的素材数量；None 表示全部（命中池 + 全部未命中池）。
         rng: 注入 smart_select_assets 的随机源（可复现）。
 
@@ -115,6 +242,7 @@ def pick_narrative_assets(
         assets,
         script_tags=script_tags,
         tag_names_by_id=tag_names_by_id,
+        clip_ai_tags_by_asset=clip_ai_tags_by_asset,
     )
 
     need = limit if (limit is not None and limit > 0) else None

packages/shared/ai_client.py

@@ -37,6 +37,7 @@ class DoubaoClient:
         self.base_url: str = settings.doubao_base_url.rstrip("/")
         self.timeout: int = settings.doubao_timeout
         self.max_retries: int = settings.doubao_max_retries
+        self.vision_model: str = settings.doubao_vision_model
 
     @property
     def is_available(self) -> bool:
@@ -103,6 +104,99 @@ class DoubaoClient:
         logger.error("豆包API调用最终失败: %s", last_error)
         return None
 
+    def vision_completion(
+        self,
+        messages: list[dict],
+        images: list[str] | None = None,
+        max_tokens: int = 2048,
+        temperature: float = 0.3,
+        timeout: int | None = None,
+    ) -> Optional[str]:
+        """调用豆包视觉理解 API（OpenAI 兼容多模态格式）.
+
+        将 images 附加到最后一条 user message 的 content 中，
+        使用 vision_model（默认 doubao-1-5-vision-pro-250915）。
+
+        Args:
+            messages: 对话消息列表。最后一条 user message 会被注入图片内容。
+            images: 图片列表，支持 base64 data URI 或 HTTP(S) URL。
+            max_tokens: 最大生成 token 数，默认 2048。
+            temperature: 采样温度，默认 0.3（视觉任务偏低更稳定）。
+            timeout: 单次请求超时秒数，不传则使用默认 self.timeout。
+
+        Returns:
+            模型返回的文本内容，失败返回 None。
+        """
+        if not self.is_available:
+            return None
+
+        # 构造多模态 content：先追加文本，再追加图片
+        vision_messages = []
+        for msg in messages:
+            vision_messages.append(dict(msg))
+
+        # 将图片注入最后一条 user message
+        if images and vision_messages:
+            # 找到最后一条 user message
+            for i in range(len(vision_messages) - 1, -1, -1):
+                if vision_messages[i].get("role") == "user":
+                    text_content = vision_messages[i].get("content", "")
+                    multi_content: list[dict[str, Any]] = []
+                    if text_content:
+                        multi_content.append({"type": "text", "text": text_content})
+                    for img in images:
+                        if img.startswith("data:") or img.startswith("http://") or img.startswith("https://"):
+                            multi_content.append({"type": "image_url", "image_url": {"url": img}})
+                        else:
+                            # 当作 base64 编码
+                            multi_content.append(
+                                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img}"}}
+                            )
+                    vision_messages[i]["content"] = multi_content
+                    break
+
+        url = f"{self.base_url}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+        payload: dict[str, Any] = {
+            "model": self.vision_model,
+            "messages": vision_messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+        }
+
+        req_timeout = timeout or self.timeout
+        last_error: Optional[Exception] = None
+        for attempt in range(self.max_retries + 1):
+            try:
+                response = httpx.post(
+                    url,
+                    headers=headers,
+                    json=payload,
+                    timeout=req_timeout,
+                )
+                response.raise_for_status()
+                data = response.json()
+                content = data["choices"][0]["message"]["content"]
+                return content.strip()
+            except Exception as e:
+                last_error = e
+                if attempt < self.max_retries:
+                    wait = 0.5 * (2**attempt)
+                    logger.warning(
+                        "豆包视觉API调用失败，%.1fs后重试 (第%d/%d次): %s",
+                        wait,
+                        attempt + 1,
+                        self.max_retries + 1,
+                        e,
+                    )
+                    time.sleep(wait)
+
+        logger.error("豆包视觉API调用最终失败: %s", last_error)
+        return None
+
 
 # ── 单例 ─────────────────────────────────────────────────────────────────────
 

packages/shared/storage.py

@@ -339,6 +339,44 @@ class SharedStorageService(StoragePort):
 
     # ── 浏览器直传 POST ────────────────────────────────────────────────
 
+    def get_upload_url(
+        self,
+        storage_key_or_url: str,
+        expires_seconds: int = 3600,
+        content_type: str = "video/mp4",
+    ) -> str:
+        """获取预签名 PUT 上传 URL（供外部 Worker 上传结果文件）。
+
+        bucket未配置时降级为 public_url（本地/开发环境）；
+        本地产物 key 原样返回。
+        """
+        if self.bucket is None:
+            if self._is_local_generated_url(storage_key_or_url):
+                return storage_key_or_url
+            logger.warning(
+                "get_upload_url: OSS bucket not configured, returning raw URL. key=%s",
+                storage_key_or_url[:200],
+            )
+            return self.get_url(self.normalize_storage_key(storage_key_or_url))
+
+        storage_key = self.normalize_storage_key(storage_key_or_url)
+        try:
+            # oss2 sign_url 支持 'PUT'，需指定 headers 才能限定 Content-Type
+            headers = {"Content-Type": content_type} if content_type else None
+            signed = self.bucket.sign_url("PUT", storage_key, expires_seconds, headers=headers)
+            logger.info(
+                "get_upload_url: signed PUT URL generated. key=%s url_prefix=%s",
+                storage_key[:80],
+                signed[:60],
+            )
+            return signed
+        except Exception:
+            logger.exception(
+                "get_upload_url: sign_url failed, falling back to raw URL. key=%s",
+                storage_key[:200],
+            )
+            return self.get_url(storage_key)
+
     def create_direct_upload_post(
         self,
         storage_key: str,

scripts/render_env.sh

@@ -57,7 +57,7 @@ if [ "$TARGET_ENV" = "staging" ]; then
 fi
 
 # 共用 secrets 直接导出（如果存在）
-SHARED_SECRETS="OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET COSYVOICE_API_KEY DASHSCOPE_API_KEY MEDIAKIT_API_KEY DOUBAO_API_KEY DOUBAO_MODEL DOUBAO_BASE_URL WECHAT_APP_ID WECHAT_APP_SECRET TIKHUB_API_KEY APIZERO_API_KEY"
+SHARED_SECRETS="OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET COSYVOICE_API_KEY DASHSCOPE_API_KEY MEDIAKIT_API_KEY DOUBAO_API_KEY DOUBAO_MODEL DOUBAO_BASE_URL DOUBAO_VISION_MODEL WECHAT_APP_ID WECHAT_APP_SECRET TIKHUB_API_KEY APIZERO_API_KEY GPU_WORKER_TOKEN"
 for var in $SHARED_SECRETS; do
   value="${!var:-}"
   # 已经在环境中了，无需额外操作

tests/unit/test_1970_atom_clip_tagger.py

@@ -0,0 +1,292 @@
+"""#1970 P2 片段级 AI 标签模块测试。
+
+测试范围：
+- build_vision_prompt: 返回有效 prompt
+- parse_vision_response: 正常/异常/空值
+- tag_atom_clip: 成功/MediaKit不可用/视觉API失败/超时降级
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+import pytest
+
+from packages.domain.atom_clip_tagger import (
+    build_vision_prompt,
+    parse_vision_response,
+    tag_atom_clip,
+)
+
+# ── Fake 对象 ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeClip:
+    id: str = "clip-001"
+    asset_id: str = "asset-001"
+    start_time: float = 0.0
+    end_time: float = 5.0
+    duration: float = 5.0
+    clip_index: int = 0
+    tags: list[str] = field(default_factory=lambda: ["tag1", "tag2"])
+    ai_tags: dict | None = None
+
+
+class FakeDoubaoClient:
+    """模拟豆包客户端."""
+
+    def __init__(self, available: bool = True, response: str | None = None, raise_error: bool = False):
+        self._available = available
+        self._response = response
+        self._raise_error = raise_error
+        self.vision_calls: list[dict] = []
+
+    @property
+    def is_available(self) -> bool:
+        return self._available
+
+    def vision_completion(self, messages, images=None, timeout=None, **kwargs):
+        self.vision_calls.append({"messages": messages, "images": images, "timeout": timeout})
+        if self._raise_error:
+            raise RuntimeError("API error")
+        return self._response
+
+
+class FakeMediaKitClient:
+    """模拟 MediaKit 客户端."""
+
+    def __init__(self, available: bool = True, frames: list[dict] | None = None):
+        self._available = available
+        self._frames = frames
+
+    @property
+    def is_available(self) -> bool:
+        return self._available
+
+    def extract_frames(self, video_url, strategy=None, max_frames=None, **kwargs):
+        return self._frames
+
+
+# ── build_vision_prompt ────────────────────────────────────────────────────
+
+
+class TestBuildVisionPrompt:
+    def test_returns_non_empty_string(self):
+        prompt = build_vision_prompt()
+        assert isinstance(prompt, str)
+        assert len(prompt) > 100
+
+    def test_contains_required_keys(self):
+        prompt = build_vision_prompt()
+        assert "scene" in prompt
+        assert "objects" in prompt
+        assert "action" in prompt
+        assert "shot" in prompt
+        assert "has_text" in prompt
+
+    def test_requests_json_format(self):
+        prompt = build_vision_prompt()
+        assert "JSON" in prompt or "json" in prompt
+
+
+# ── parse_vision_response ──────────────────────────────────────────────────
+
+
+class TestParseVisionResponse:
+    def test_valid_json(self):
+        response = json.dumps(
+            {
+                "scene": ["工厂", "车间"],
+                "objects": ["产品", "机器"],
+                "action": ["演示"],
+                "shot": "特写",
+                "has_text": True,
+            }
+        )
+        result = parse_vision_response(response)
+        assert result["scene"] == ["工厂", "车间"]
+        assert result["objects"] == ["产品", "机器"]
+        assert result["action"] == ["演示"]
+        assert result["shot"] == "特写"
+        assert result["has_text"] is True
+
+    def test_json_with_markdown_code_block(self):
+        response = '```json\n{"scene": ["办公室"], "objects": ["电脑"], "action": ["说话"], "shot": "中景", "has_text": false}\n```'
+        result = parse_vision_response(response)
+        assert result["scene"] == ["办公室"]
+        assert result["has_text"] is False
+
+    def test_json_embedded_in_text(self):
+        response = '这是一些说明文字\n{"scene": ["户外"], "objects": ["汽车"], "action": ["展示"], "shot": "远景", "has_text": false}\n结束'
+        result = parse_vision_response(response)
+        assert result["scene"] == ["户外"]
+
+    def test_empty_response(self):
+        assert parse_vision_response("") == {}
+        assert parse_vision_response(None) == {}
+        assert parse_vision_response("   ") == {}
+
+    def test_invalid_json(self):
+        assert parse_vision_response("这不是JSON") == {}
+
+    def test_partial_fields(self):
+        response = json.dumps({"scene": ["工厂"]})
+        result = parse_vision_response(response)
+        assert result["scene"] == ["工厂"]
+        assert result["objects"] == []
+        assert result["shot"] == ""
+        assert result["has_text"] is False
+
+    def test_invalid_shot_value(self):
+        response = json.dumps({"scene": [], "objects": [], "action": [], "shot": "全景", "has_text": False})
+        result = parse_vision_response(response)
+        # "全景" 不在有效值 ("特写", "中景", "远景") 中
+        assert result["shot"] == ""
+
+    def test_string_values_converted_to_list(self):
+        response = json.dumps(
+            {"scene": "工厂", "objects": "产品", "action": "演示", "shot": "特写", "has_text": "true"}
+        )
+        result = parse_vision_response(response)
+        assert result["scene"] == ["工厂"]
+        assert result["objects"] == ["产品"]
+        assert result["has_text"] is True
+
+    def test_non_dict_json(self):
+        assert parse_vision_response("[1, 2, 3]") == {}
+        assert parse_vision_response('"hello"') == {}
+
+
+# ── tag_atom_clip ──────────────────────────────────────────────────────────
+
+
+class TestTagAtomClip:
+    def test_success_with_mediakit(self):
+        """MediaKit 可用 + 视觉 API 成功 → 返回完整 AI 标签."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient(
+            response=json.dumps(
+                {
+                    "scene": ["工厂"],
+                    "objects": ["产品"],
+                    "action": ["演示"],
+                    "shot": "特写",
+                    "has_text": False,
+                }
+            )
+        )
+        fake_mediakit = FakeMediaKitClient(
+            frames=[
+                {"image_url": "https://example.com/frame1.jpg", "timestamp": 0.0},
+                {"image_url": "https://example.com/frame2.jpg", "timestamp": 2.5},
+                {"image_url": "https://example.com/frame3.jpg", "timestamp": 5.0},
+            ]
+        )
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+            mediakit_client=fake_mediakit,
+        )
+
+        assert result["scene"] == ["工厂"]
+        assert result["objects"] == ["产品"]
+        assert result["shot"] == "特写"
+        assert result["inherited_tags"] == ["tag1", "tag2"]
+        assert len(fake_doubao.vision_calls) == 1
+
+    def test_doubao_unavailable_returns_inherited(self):
+        """DoubaoClient 不可用 → 返回 inherited_tags."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient(available=False)
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+        )
+
+        assert result == {"inherited_tags": ["tag1", "tag2"]}
+        assert len(fake_doubao.vision_calls) == 0
+
+    def test_mediakit_unavailable_no_ffmpeg(self):
+        """MediaKit 不可用 + 无 ffmpeg → 降级 inherited_tags."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient()
+        fake_mediakit = FakeMediaKitClient(available=False)
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+            mediakit_client=fake_mediakit,
+        )
+
+        # 没有 ffmpeg 的情况下，帧提取失败
+        assert result == {"inherited_tags": ["tag1", "tag2"]}
+
+    def test_vision_api_error_returns_inherited(self):
+        """视觉 API 抛异常 → 降级 inherited_tags."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient(raise_error=True)
+        fake_mediakit = FakeMediaKitClient(frames=[{"image_url": "https://example.com/frame.jpg", "timestamp": 0.0}])
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+            mediakit_client=fake_mediakit,
+        )
+
+        assert result == {"inherited_tags": ["tag1", "tag2"]}
+
+    def test_vision_api_empty_response(self):
+        """视觉 API 返回空 → 降级 inherited_tags."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient(response=None)
+        fake_mediakit = FakeMediaKitClient(frames=[{"image_url": "https://example.com/frame.jpg", "timestamp": 0.0}])
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+            mediakit_client=fake_mediakit,
+        )
+
+        assert result == {"inherited_tags": ["tag1", "tag2"]}
+
+    def test_vision_api_invalid_json_response(self):
+        """视觉 API 返回无效 JSON → 降级 inherited_tags."""
+        clip = FakeClip()
+        fake_doubao = FakeDoubaoClient(response="这不是JSON格式")
+        fake_mediakit = FakeMediaKitClient(frames=[{"image_url": "https://example.com/frame.jpg", "timestamp": 0.0}])
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+            mediakit_client=fake_mediakit,
+        )
+
+        assert result == {"inherited_tags": ["tag1", "tag2"]}
+
+    def test_clip_with_empty_tags(self):
+        """空素材标签 → inherited_tags 为空列表."""
+        clip = FakeClip(tags=[])
+        fake_doubao = FakeDoubaoClient(available=False)
+
+        result = tag_atom_clip(
+            clip=clip,
+            video_url="https://example.com/video.mp4",
+            doubao_client=fake_doubao,
+        )
+
+        assert result == {"inherited_tags": []}
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])

tests/unit/test_1970_celery_ai_tag_registration.py

@@ -0,0 +1,82 @@
+"""#1970 AI 标签 Celery 任务注册回归测试。
+
+背景：staging 上 worker.generate_atom_clips 正常派发 tag_atom_clip，
+但消费端报 "Received unregistered task of type 'worker.tag_atom_clip'"，
+根因是 celery_app.conf.imports 漏列任务模块，worker 进程从未 import 之。
+
+注意：tests/unit 下大量旧测试在 import 期向 sys.modules 注入
+worker_app.celery_app 的 MagicMock 且不还原，全量收集时会污染本测试，
+因此这里用 AST 静态解析 + 隔离子进程验证，不依赖 sys.modules 状态。
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+CELERY_APP_PY = REPO_ROOT / "apps" / "worker" / "worker_app" / "celery_app.py"
+
+REQUIRED_MODULES = (
+    "worker_app.tasks.atom_clip_tagging",
+    "worker_app.tasks.backfill_atom_clip_tags",
+)
+
+
+def _conf_imports_values() -> set[str]:
+    """从 celery_app.py AST 中提取 celery_app.conf.imports 元组的字符串项。"""
+    tree = ast.parse(CELERY_APP_PY.read_text(encoding="utf-8"))
+    values: set[str] = set()
+    for node in ast.walk(tree):
+        if not (isinstance(node, ast.Assign) and len(node.targets) == 1):
+            continue
+        target = node.targets[0]
+        # celery_app.conf.imports = (...) 或 conf.imports = (...)
+        if not (isinstance(target, ast.Attribute) and target.attr == "imports"):
+            continue
+        if isinstance(node.value, (ast.Tuple, ast.List)):
+            for elt in node.value.elts:
+                if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
+                    values.add(elt.value)
+    return values
+
+
+def test_ai_tag_modules_in_celery_imports():
+    imports = _conf_imports_values()
+    for module in REQUIRED_MODULES:
+        assert module in imports, f"{module} 未加入 celery_app.conf.imports"
+
+
+def test_ai_tag_tasks_registered_in_isolated_process():
+    """隔离子进程（无 conftest / 无 sys.modules mock）真实加载 Celery app。"""
+    # 模拟 worker 启动时按 conf.imports import 任务模块的行为；
+    # 只导入 AI 标签两个模块（其他模块依赖 cv2 等本地未安装的重依赖）。
+    code = (
+        "import importlib, sys; "
+        "from worker_app.celery_app import celery_app; "
+        "mods = [m for m in celery_app.conf.imports or () "
+        "if 'atom_clip_tagging' in m or 'backfill_atom_clip_tags' in m]; "
+        "[importlib.import_module(m) for m in mods]; "
+        "missing = [n for n in "
+        "['worker.tag_atom_clip', 'worker.backfill_atom_clip_tags'] "
+        "if n not in celery_app.tasks]; "
+        "sys.exit(1 if missing or len(mods) < 2 else 0)"
+    )
+    env = os.environ.copy()
+    paths = [
+        str(REPO_ROOT),
+        str(REPO_ROOT / "apps" / "worker"),
+        str(REPO_ROOT / "packages"),
+    ]
+    env["PYTHONPATH"] = os.pathsep.join(paths) + os.pathsep + env.get("PYTHONPATH", "")
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=60,
+    )
+    assert result.returncode == 0, "隔离子进程中任务未注册成功:\n" f"stdout={result.stdout}\nstderr={result.stderr}"

tests/unit/test_1970_force_backfill_tags.py

@@ -0,0 +1,347 @@
+"""#1970 force 回填降级 AI 标签记录的回归测试。
+
+背景：DOUBAO_VISION_MODEL 未配置时，tagger 降级写入
+{"inherited_tags": [...]}（非 NULL），默认 backfill 只捞 ai_tags IS NULL，
+这批记录永远不会重打。force=True 时应纳入降级记录，并在打标成功后覆盖。
+
+覆盖：
+- find_untagged(include_downgraded) 的 SQL 过滤（SQLite 验证跨库 JSON 取值）
+- tag_atom_clip_task 的 force 跳过/放行/覆盖逻辑
+- backfill_atom_clip_tags(force=True) 给 tag 任务传 kwargs={"force": True}
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import AssetAtomClipModel
+
+# ── 仓储层：find_untagged 过滤 ─────────────────────────────────────────────
+
+
+@pytest.fixture
+def repo_session():
+    engine = create_engine("sqlite:///:memory:")
+    AssetAtomClipModel.__table__.create(engine)
+    SessionTest = sessionmaker(bind=engine)
+    session = SessionTest()
+    now = datetime.now(UTC)
+    session.add_all(
+        [
+            AssetAtomClipModel(
+                id="c-null",
+                asset_id="a1",
+                start_time=0,
+                end_time=1,
+                duration=1,
+                clip_index=0,
+                tags=[],
+                ai_tags=None,
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-downgraded-empty",
+                asset_id="a1",
+                start_time=1,
+                end_time=2,
+                duration=1,
+                clip_index=1,
+                tags=[],
+                ai_tags={"inherited_tags": []},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-downgraded-tags",
+                asset_id="a1",
+                start_time=2,
+                end_time=3,
+                duration=1,
+                clip_index=2,
+                tags=[],
+                ai_tags={"inherited_tags": ["口播"]},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-tagged-true",
+                asset_id="a1",
+                start_time=3,
+                end_time=4,
+                duration=1,
+                clip_index=3,
+                tags=[],
+                ai_tags={"has_text": True, "scene": ["室内"], "inherited_tags": []},
+                created_at=now,
+            ),
+            AssetAtomClipModel(
+                id="c-tagged-false",
+                asset_id="a1",
+                start_time=4,
+                end_time=5,
+                duration=1,
+                clip_index=4,
+                tags=[],
+                ai_tags={"has_text": False, "inherited_tags": ["风景"]},
+                created_at=now,
+            ),
+        ]
+    )
+    session.commit()
+    # SQLAlchemy JSON 在 SQLite 下把 None 序列化为 'null' 字符串，
+    # 而生产 PostgreSQL 存的是真 SQL NULL；用原生 SQL 对齐生产语义。
+    from sqlalchemy import text
+
+    session.execute(text("UPDATE asset_atom_clips SET ai_tags = NULL WHERE id = 'c-null'"))
+    session.commit()
+    yield session
+    session.close()
+
+
+def test_find_untagged_default_only_null(repo_session):
+    repo = SQLAlchemyAssetAtomClipRepository(repo_session)
+    ids = {c.id for c in repo.find_untagged(limit=100)}
+    assert ids == {"c-null"}
+
+
+def test_find_untagged_include_downgraded(repo_session):
+    repo = SQLAlchemyAssetAtomClipRepository(repo_session)
+    ids = {c.id for c in repo.find_untagged(limit=100, include_downgraded=True)}
+    # NULL + 两条降级记录；含 has_text=true/false 的完整记录都排除
+    assert ids == {"c-null", "c-downgraded-empty", "c-downgraded-tags"}
+
+
+# ── 任务层：tag_atom_clip_task 的 force 语义 ───────────────────────────────
+
+
+def _import_tag_task_module():
+    from worker_app.tasks import atom_clip_tagging as mod
+
+    return mod
+
+
+def _call_tag_task(mod, clip_id, force):
+    """直接调用任务，兼容两种环境。
+
+    全量收集时旧测试向 sys.modules 注入 celery_app MagicMock（其 task
+    装饰器原样返回裸函数），此时是普通函数需显式传 self=None；
+    正常 Celery 环境下属性是 Task 代理对象（非普通 function），
+    已绑定 self，按业务签名直接调用即可。
+    """
+    import inspect
+
+    obj = mod.tag_atom_clip_task
+    if inspect.isfunction(obj):
+        return obj(None, clip_id, force=force)
+    return obj(clip_id, force=force)
+
+
+def test_tag_task_skips_downgraded_without_force(monkeypatch):
+    mod = _import_tag_task_module()
+    monkeypatch.setattr(
+        mod,
+        "SessionLocal",
+        lambda: SimpleNamespace(
+            rollback=lambda: None,
+            close=lambda: None,
+        ),
+    )
+
+    class _Repo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                ai_tags={"inherited_tags": []},
+            )
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _Repo)
+
+    result = _call_tag_task(mod, "clip-downgraded", force=False)
+    assert result["status"] == "skipped"
+    assert result["reason"] == "already tagged"
+
+
+def test_tag_task_force_retags_downgraded_and_overwrites(monkeypatch):
+    mod = _import_tag_task_module()
+    updated: dict[str, dict] = {}
+
+    class _FakeSession:
+        def rollback(self):
+            pass
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(mod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                asset_id="asset-1",
+                start_time=0.0,
+                end_time=2.0,
+                tags=["旧标签"],
+                ai_tags={"inherited_tags": []},
+            )
+
+        def update_ai_tags(self, clip_id, ai_tags):
+            updated[clip_id] = ai_tags
+
+    class _AssetRepo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, asset_id):
+            return SimpleNamespace(id=asset_id, storage_key="k/video.mp4")
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _AtomRepo)
+    monkeypatch.setattr(mod, "SQLAlchemyAssetRepository", _AssetRepo)
+
+    class _Storage:
+        def get_download_url(self, key, expires_seconds=3600):
+            return "https://example.com/signed.mp4"
+
+    monkeypatch.setattr(mod, "get_shared_storage_service", lambda: _Storage())
+    monkeypatch.setattr(mod, "get_doubao_client", lambda: object())
+    monkeypatch.setattr(mod, "get_mediakit_client", lambda: None)
+
+    new_tags = {
+        "scene": ["室内"],
+        "objects": ["人物"],
+        "action": ["说话"],
+        "shot": "中景",
+        "has_text": True,
+        "inherited_tags": ["旧标签"],
+    }
+    monkeypatch.setattr(mod, "tag_atom_clip", lambda **kw: new_tags)
+
+    result = _call_tag_task(mod, "clip-downgraded", force=True)
+    assert result["status"] == "completed"
+    assert result["has_ai_tags"] is True
+    assert updated["clip-downgraded"] == new_tags
+
+
+def test_tag_task_force_still_skips_complete_tags(monkeypatch):
+    mod = _import_tag_task_module()
+    monkeypatch.setattr(
+        mod,
+        "SessionLocal",
+        lambda: SimpleNamespace(rollback=lambda: None, close=lambda: None),
+    )
+
+    class _Repo:
+        def __init__(self, db):
+            pass
+
+        def find_by_id(self, clip_id):
+            return SimpleNamespace(
+                id=clip_id,
+                ai_tags={"has_text": False, "inherited_tags": []},
+            )
+
+    monkeypatch.setattr(mod, "SQLAlchemyAssetAtomClipRepository", _Repo)
+
+    result = _call_tag_task(mod, "clip-complete", force=True)
+    assert result["status"] == "skipped"
+    assert result["reason"] == "already tagged"
+
+
+# ── backfill 任务：force 透传到 send_task ──────────────────────────────────
+
+
+def test_backfill_force_passes_kwarg(monkeypatch):
+    from worker_app.tasks import backfill_atom_clip_tags as bmod
+
+    sent: list[tuple] = []
+
+    class _FakeSession:
+        def close(self):
+            pass
+
+    monkeypatch.setattr(bmod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            self.calls: list[bool] = []
+
+        def find_untagged(self, limit, include_downgraded=False):
+            self.calls.append(include_downgraded)
+            # 第一批返回一条降级记录，第二批返回空结束循环
+            if len(self.calls) == 1:
+                return [SimpleNamespace(id="clip-1")]
+            return []
+
+    repo_holder = {}
+
+    def _repo_factory(db):
+        repo = _AtomRepo(db)
+        repo_holder["repo"] = repo
+        return repo
+
+    monkeypatch.setattr(bmod, "SQLAlchemyAssetAtomClipRepository", _repo_factory)
+
+    def _send_task(name, args=None, kwargs=None):
+        sent.append((name, args, kwargs))
+
+    monkeypatch.setattr(bmod.celery_app, "send_task", _send_task)
+
+    result = bmod.backfill_atom_clip_tags(batch_size=10, batch_interval=0, force=True)
+
+    assert result["status"] == "completed"
+    assert result["total_submitted"] == 1
+    assert repo_holder["repo"].calls == [True, True]
+    assert sent == [
+        ("worker.tag_atom_clip", ["clip-1"], {"force": True}),
+    ]
+
+
+def test_backfill_default_does_not_force(monkeypatch):
+    from worker_app.tasks import backfill_atom_clip_tags as bmod
+
+    sent_kwargs: list[dict | None] = []
+
+    class _FakeSession:
+        def close(self):
+            pass
+
+    monkeypatch.setattr(bmod, "SessionLocal", _FakeSession)
+
+    class _AtomRepo:
+        def __init__(self, db):
+            self.calls: list[bool] = []
+
+        def find_untagged(self, limit, include_downgraded=False):
+            self.calls.append(include_downgraded)
+            return [SimpleNamespace(id="clip-null")] if self.calls == [False] else []
+
+    holder = {}
+
+    def _repo_factory(db):
+        holder["repo"] = _AtomRepo(db)
+        return holder["repo"]
+
+    monkeypatch.setattr(bmod, "SQLAlchemyAssetAtomClipRepository", _repo_factory)
+    monkeypatch.setattr(
+        bmod.celery_app,
+        "send_task",
+        lambda name, args=None, kwargs=None: sent_kwargs.append(kwargs),
+    )
+
+    result = bmod.backfill_atom_clip_tags(batch_size=10, batch_interval=0)
+
+    assert result["total_submitted"] == 1
+    assert holder["repo"].calls == [False, False]
+    assert sent_kwargs == [{"force": False}]

tests/unit/test_1970_gpu_worker_heartbeat.py

@@ -0,0 +1,254 @@
+"""#1970 GPU Worker 修复单测.
+
+覆盖 deploy/gpu_worker/gpu_worker.py（独立部署脚本，不在 apps/packages 包内，
+按文件路径动态加载）：
+1. 默认配置：REQUEST_TIMEOUT=900 / TASK_MAX_RETRY=1 / 心跳 30s / 最短 3s；
+2. 推理期心跳线程 POST /gpu/register 带 task_id，任务结束能停；
+3. <3s 短视频直接上报失败，不调用 MuseTalk；
+4. _call_musetalk 仅对 5xx/网络瞬时错误标记 retryable，4xx 不重试；
+5. _handle_task 只对 retryable 错误本地重试 1 次。
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+import time
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+WORKER_PATH = ROOT / "deploy" / "gpu_worker" / "gpu_worker.py"
+
+
+def _load_worker_module():
+    spec = importlib.util.spec_from_file_location("gpu_worker_standalone_1970", WORKER_PATH)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture
+def worker():
+    return _load_worker_module()
+
+
+# ── 默认配置 ───────────────────────────────────────────────────────
+
+
+def test_config_defaults_900_and_retry_one(monkeypatch):
+    """CI/本机若显式导出过这些 env，说明是运维覆盖，不应拿默认值断言；
+    因此只在四个 env 全部缺失时校验脚本内置默认值（#1970：900/1/30/3）。"""
+    keys = (
+        "REQUEST_TIMEOUT",
+        "TASK_MAX_RETRY",
+        "TASK_HEARTBEAT_INTERVAL",
+        "MIN_VIDEO_DURATION_SECONDS",
+    )
+    if any(k in os.environ for k in keys):
+        pytest.skip("环境显式设置了 worker 超时/重试变量，跳过默认值断言")
+    for key in keys:
+        monkeypatch.delenv(key, raising=False)
+    mod = _load_worker_module()
+    assert mod.Config.request_timeout == 900.0
+    assert mod.Config.task_max_retry == 1
+    assert mod.Config.task_heartbeat_interval == 30.0
+    assert mod.Config.min_video_duration_seconds == 3.0
+
+
+# ── register 携带 task_id ──────────────────────────────────────────
+
+
+def test_register_payload_includes_task_id_only_when_provided(worker, monkeypatch):
+    captured = []
+
+    class _Resp:
+        status_code = 200
+        text = ""
+
+    def _fake_post(url, json=None, headers=None, timeout=None):
+        captured.append(json)
+        return _Resp()
+
+    monkeypatch.setattr(worker.requests, "post", _fake_post)
+    monkeypatch.setattr(worker, "_check_musetalk_health", lambda: (True, {}))
+
+    assert worker._register("task-abc") is True
+    assert captured[-1]["task_id"] == "task-abc"
+    assert captured[-1]["worker_id"]
+
+    worker._register()  # 空闲心跳不带 task_id
+    assert "task_id" not in captured[-1]
+
+
+# ── 推理期心跳线程 ─────────────────────────────────────────────────
+
+
+def test_task_heartbeat_thread_sends_and_stops(worker, monkeypatch):
+    calls = []
+
+    def _fake_register(task_id=None):
+        calls.append(task_id)
+        return True
+
+    monkeypatch.setattr(worker, "_register", _fake_register)
+    hb = worker.TaskHeartbeat("task-hb1", interval=5)
+    hb.start()
+    time.sleep(0.3)  # 启动后立即发一次
+    hb.stop()
+    hb.join(timeout=2)
+    assert not hb.is_alive()
+    assert calls and all(c == "task-hb1" for c in calls)
+
+
+# ── 短视频前置拦截 ─────────────────────────────────────────────────
+
+
+def test_handle_task_short_video_reports_failed_without_inference(worker, monkeypatch, tmp_path):
+    video = tmp_path / "input.mp4"
+    video.write_bytes(b"fake-mp4-bytes")
+    audio = tmp_path / "input_audio.bin"
+    audio.write_bytes(b"fake-audio")
+    reports = []
+
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_download", lambda url, path: True)
+    # ffprobe 读出 1.2s → 低于 3s 阈值
+    monkeypatch.setattr(worker, "_probe_duration", lambda path: 1.2)
+
+    def _boom(*a, **k):
+        raise AssertionError("短视频不应调用 MuseTalk 推理")
+
+    monkeypatch.setattr(worker, "_call_musetalk", _boom)
+    monkeypatch.setattr(
+        worker,
+        "_report_result",
+        lambda task_id, success, duration=0.0, error_msg="": reports.append((task_id, success, error_msg)) or True,
+    )
+
+    task = {
+        "task_id": "task-short",
+        "video_url": "https://example.com/v.mp4",
+        "audio_url": "https://example.com/a.bin",
+    }
+    worker._handle_task(task)
+
+    assert len(reports) == 1
+    tid, ok, err = reports[0]
+    assert tid == "task-short"
+    assert ok is False
+    assert "视频过短" in err
+    assert "3" in err
+
+
+def test_handle_task_probe_failure_does_not_block(worker, monkeypatch):
+    """ffprobe 不可用（duration=0.0）时不能误杀，应继续推理."""
+    reports = []
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_download", lambda url, path: True)
+    monkeypatch.setattr(worker, "_probe_duration", lambda path: 0.0)
+    monkeypatch.setattr(
+        worker,
+        "_call_musetalk",
+        lambda v, a, o: (True, 8.0, "", False),
+    )
+    uploaded = []
+    monkeypatch.setattr(
+        worker,
+        "_report_success_with_file",
+        lambda task_id, duration, path: uploaded.append((task_id, duration)),
+    )
+    monkeypatch.setattr(worker, "_report_result", lambda *a, **k: True)
+
+    worker._handle_task({"task_id": "task-probe0", "video_url": "u", "audio_url": "u"})
+    assert uploaded == [("task-probe0", 8.0)]
+    assert reports == []
+
+
+# ── 重试语义：仅瞬时错误重试 ───────────────────────────────────────
+
+
+def test_call_musetalk_4xx_not_retryable_5xx_retryable(worker, monkeypatch, tmp_path):
+    video = tmp_path / "v.mp4"
+    audio = tmp_path / "a.bin"
+    video.write_bytes(b"v")
+    audio.write_bytes(b"a")
+    out = tmp_path / "o.mp4"
+
+    class _Resp:
+        def __init__(self, code, body=b"x" * 2048):
+            self.status_code = code
+            self.content = body
+            self.text = "err"
+
+    # 4xx：确定性失败，不重试
+    monkeypatch.setattr(worker.requests, "post", lambda *a, **k: _Resp(400))
+    ok, _, _, retryable = worker._call_musetalk(video, audio, out)
+    assert ok is False and retryable is False
+
+    monkeypatch.setattr(worker.requests, "post", lambda *a, **k: _Resp(503))
+    ok, _, _, retryable = worker._call_musetalk(video, audio, out)
+    assert ok is False and retryable is True
+
+    # 连接异常：瞬时错误，可重试
+    import requests as _requests
+
+    def _conn_err(*a, **k):
+        raise _requests.exceptions.ConnectionError("reset")
+
+    monkeypatch.setattr(worker.requests, "post", _conn_err)
+    ok, _, _, retryable = worker._call_musetalk(video, audio, out)
+    assert ok is False and retryable is True
+
+
+def test_handle_task_retries_once_for_transient_then_succeeds(worker, monkeypatch):
+    calls = []
+
+    def _fake_call(v, a, o):
+        calls.append(1)
+        if len(calls) == 1:
+            return False, 0.0, "MuseTalk HTTP 503: busy", True
+        return True, 6.5, "", False
+
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_download", lambda url, path: True)
+    monkeypatch.setattr(worker, "_probe_duration", lambda path: 12.0)
+    monkeypatch.setattr(worker, "_call_musetalk", _fake_call)
+    monkeypatch.setattr(worker, "time", mock.MagicMock())  # 重试 sleep 立即返回
+    uploaded = []
+    monkeypatch.setattr(
+        worker,
+        "_report_success_with_file",
+        lambda task_id, duration, path: uploaded.append((task_id, duration)),
+    )
+
+    worker._handle_task({"task_id": "t-retry", "video_url": "u", "audio_url": "u"})
+    assert len(calls) == 2
+    assert uploaded == [("t-retry", 6.5)]
+
+
+def test_handle_task_no_retry_for_deterministic_failure(worker, monkeypatch):
+    calls = []
+
+    def _fake_call(v, a, o):
+        calls.append(1)
+        return False, 0.0, "MuseTalk HTTP 400: bad input", False
+
+    reports = []
+    monkeypatch.setattr(worker, "_register", lambda *a, **k: True)
+    monkeypatch.setattr(worker, "_download", lambda url, path: True)
+    monkeypatch.setattr(worker, "_probe_duration", lambda path: 12.0)
+    monkeypatch.setattr(worker, "_call_musetalk", _fake_call)
+    monkeypatch.setattr(
+        worker,
+        "_report_result",
+        lambda task_id, success, duration=0.0, error_msg="": reports.append(error_msg) or True,
+    )
+
+    worker._handle_task({"task_id": "t-4xx", "video_url": "u", "audio_url": "u"})
+    assert len(calls) == 1  # 4xx 本地不重试，直接交服务端决定
+    assert reports and "400" in reports[0]

tests/unit/test_1970_hflip_ai_tags.py

@@ -0,0 +1,185 @@
+"""#1970 hflip 放开（has_text 来自 atom_clip.ai_tags）端到端参数链路测试。
+
+覆盖：
+1. UnifiedRenderService 传入 clip_has_text 后微变换计划的翻转门控；
+2. RenderAdapter._resolve_clip_has_text 按 atom_clip.ai_tags.has_text
+   解析布尔列表（显式 False 才可翻转，其余保守），失败回退 None；
+3. 纯函数层在「混合有/无文字」列表下的行为（顺序对齐）。
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.micro_transform_pure import build_micro_transform_plan
+
+
+def _make_service(plan_config: dict | None = None, clip_has_text=None):
+    from video_processing.unified_render_service import UnifiedRenderService
+
+    svc = object.__new__(UnifiedRenderService)
+    svc.plan = MagicMock()
+    svc.plan.config = plan_config or {}
+    svc.plan.id = "plan-1"
+    svc.plan.clips = []
+    svc._micro_plan_cache = None
+    svc._micro_plan_loaded = False
+    svc._clip_has_text = clip_has_text
+    return svc
+
+
+def _clip(clip_id: str, atom_clip_id: str = "", clip_type: str = "main"):
+    return SimpleNamespace(id=clip_id, atom_clip_id=atom_clip_id, clip_type=clip_type)
+
+
+def _atom(clip_id: str, ai_tags):
+    return SimpleNamespace(id=clip_id, ai_tags=ai_tags)
+
+
+class TestServiceClipHasText:
+    def test_none_stays_conservative(self):
+        # 未注入检测列表：所有片段一律不翻转
+        svc = _make_service({"generation_task_id": "t1"}, clip_has_text=None)
+        plan = svc._get_micro_transform_plan(30)
+        assert plan is not None
+        assert all(c.has_text for c in plan.clips)
+        assert all(not c.hflip for c in plan.clips)
+
+    def test_explicit_no_text_allows_hflip(self):
+        # AI 明确判定无文字：允许参与 50% 翻转（40 段应至少出现一些翻转）
+        svc = _make_service({"generation_task_id": "t-allow"}, clip_has_text=[False] * 40)
+        plan = svc._get_micro_transform_plan(40)
+        assert plan is not None
+        assert all(not c.has_text for c in plan.clips)
+        assert any(c.hflip for c in plan.clips)
+        assert all(not c.hflip or not c.has_text for c in plan.clips)
+
+    def test_all_text_never_flips(self):
+        svc = _make_service({"generation_task_id": "t-text"}, clip_has_text=[True] * 40)
+        plan = svc._get_micro_transform_plan(40)
+        assert all(c.has_text for c in plan.clips)
+        assert all(not c.hflip for c in plan.clips)
+
+    def test_mixed_order_alignment(self):
+        # 仅第 0、2 个片段无文字；has_text 标记必须与片段序号严格对齐
+        svc = _make_service({"generation_task_id": "t-mix"}, clip_has_text=[False, True, False, True])
+        plan = svc._get_micro_transform_plan(4)
+        assert [c.has_text for c in plan.clips] == [False, True, False, True]
+        assert all(not plan.clips[i].hflip for i in (1, 3))
+        for i in (0, 2):
+            # 无文字片段的翻转由 50% 种子决定，但允许翻转（不强制一定翻）
+            assert plan.clips[i].has_text is False
+
+    def test_list_shorter_than_clips_missing_are_conservative(self):
+        # 列表短于片段数：缺位片段按有文字处理
+        svc = _make_service({"generation_task_id": "t-short"}, clip_has_text=[False])
+        plan = svc._get_micro_transform_plan(3)
+        assert [c.has_text for c in plan.clips] == [False, True, True]
+        assert not plan.clips[1].hflip and not plan.clips[2].hflip
+
+    def test_plan_reproducible_with_real_list(self):
+        cfg = {"generation_task_id": "task-x", "video_index": 1}
+        flags = [False, True, False, False, True]
+        p1 = _make_service(cfg, clip_has_text=flags)._get_micro_transform_plan(5)
+        p2 = _make_service(dict(cfg), clip_has_text=list(flags))._get_micro_transform_plan(5)
+        assert [c.hflip for c in p1.clips] == [c.hflip for c in p2.clips]
+
+
+class TestPureMixedFlags:
+    def test_pure_function_mixed_flags(self):
+        plan = build_micro_transform_plan("seed-1", 0, 4, clip_has_text=[False, True, False, True])
+        assert [c.has_text for c in plan.clips] == [False, True, False, True]
+        # 有文字片段绝不翻转
+        assert not plan.clips[1].hflip and not plan.clips[3].hflip
+
+
+class TestResolveClipHasText:
+    def _adapter(self):
+        from video_processing.render_adapter import RenderAdapter
+
+        return RenderAdapter(MagicMock())
+
+    def test_no_atom_ids_returns_none(self):
+        adapter = self._adapter()
+        clips = [_clip("c1", ""), _clip("c2", "")]
+        assert adapter._resolve_clip_has_text(clips) is None
+
+    def test_explicit_false_only_maps_to_false(self):
+        adapter = self._adapter()
+        clips = [
+            _clip("c1", "a1"),
+            _clip("c2", "a2"),
+            _clip("c3", "a3"),
+            _clip("c4", "a4"),
+            _clip("c5", "a5"),
+        ]
+        atoms = [
+            _atom("a1", {"has_text": False}),  # 明确无文字 → False
+            _atom("a2", {"has_text": True}),  # 有文字
+            _atom("a3", None),  # 标签未生成
+            _atom("a4", {"scene": ["工厂"]}),  # has_text 缺失（null）
+            _atom("a5", {"has_text": "false"}),  # 非布尔 → 保守
+        ]
+        with patch(
+            "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository."
+            "SQLAlchemyAssetAtomClipRepository.find_by_ids",
+            return_value=atoms,
+        ):
+            result = adapter._resolve_clip_has_text(clips)
+        assert result == [False, True, True, True, True]
+
+    def test_audio_clips_excluded_and_order_kept(self):
+        adapter = self._adapter()
+        clips = [
+            _clip("c1", "a1", clip_type="main"),
+            _clip("bgm", "", clip_type="audio"),
+            _clip("c2", "a2", clip_type="pip"),
+        ]
+        atoms = [
+            _atom("a1", {"has_text": False}),
+            _atom("a2", {"has_text": False}),
+        ]
+        with patch(
+            "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository."
+            "SQLAlchemyAssetAtomClipRepository.find_by_ids",
+            return_value=atoms,
+        ) as mock_find:
+            result = adapter._resolve_clip_has_text(clips)
+        # 只查非 audio 片段的 atom id，且顺序为 main → pip
+        assert mock_find.call_args.args[0] == ["a1", "a2"]
+        assert result == [False, False]
+
+    def test_missing_atom_record_defaults_true(self):
+        adapter = self._adapter()
+        clips = [_clip("c1", "a1"), _clip("c2", "a2")]
+        with patch(
+            "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository."
+            "SQLAlchemyAssetAtomClipRepository.find_by_ids",
+            return_value=[_atom("a1", {"has_text": False})],  # a2 查不到
+        ):
+            result = adapter._resolve_clip_has_text(clips)
+        assert result == [False, True]
+
+    def test_query_failure_returns_none(self):
+        adapter = self._adapter()
+        clips = [_clip("c1", "a1")]
+        with patch(
+            "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository."
+            "SQLAlchemyAssetAtomClipRepository.find_by_ids",
+            side_effect=RuntimeError("db down"),
+        ):
+            assert adapter._resolve_clip_has_text(clips) is None
+
+    def test_duplicate_atom_ids_queried_once(self):
+        adapter = self._adapter()
+        clips = [_clip("c1", "a1"), _clip("c2", "a1")]
+        with patch(
+            "packages.adapters.sqlalchemy_impl.asset_atom_clip_repository."
+            "SQLAlchemyAssetAtomClipRepository.find_by_ids",
+            return_value=[_atom("a1", {"has_text": False})],
+        ) as mock_find:
+            result = adapter._resolve_clip_has_text(clips)
+        assert mock_find.call_args.args[0] == ["a1"]
+        assert result == [False, False]

tests/unit/test_1970_micro_transform_render.py

@@ -22,6 +22,7 @@ def _make_service(plan_config: dict | None = None, clips=None):
     svc.plan.clips = clips or []
     svc._micro_plan_cache = None
     svc._micro_plan_loaded = False
+    svc._clip_has_text = None
     return svc
 
 
@@ -159,6 +160,7 @@ class TestStreamCopyGate:
         svc.clips = [source]
         svc._micro_plan_cache = None
         svc._micro_plan_loaded = False
+        svc._clip_has_text = None
         resolved = ResolvedClip(
             clip_id="c1",
             asset_id="a1",

tests/unit/test_1970_narrative_clip_weighted.py

@@ -0,0 +1,306 @@
+"""#1970 P2 叙事匹配 AI 标签加权测试。
+
+测试范围：
+- AI 标签命中时权重 2.0
+- 无 AI 标签时降级到素材标签权重 1.0
+- 混合场景（部分素材有 AI 标签，部分只有素材标签）
+- compute_tag_match_score 归一化得分
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import random
+from dataclasses import dataclass, field
+
+import pytest
+
+from packages.domain.narrative_match import (
+    AI_TAG_WEIGHT,
+    ASSET_TAG_WEIGHT,
+    _compute_ai_score,
+    _extract_ai_tag_names,
+    compute_tag_match_score,
+    match_assets_by_script_tags,
+    pick_narrative_assets,
+)
+
+
+@dataclass
+class FakeAsset:
+    id: str
+    tag_ids: list[str] = field(default_factory=list)
+    tags: list[str] = field(default_factory=list)
+    status: str = "ready"
+    file_type: str = "video"
+    duration: float = 10.0
+    quality_score: float | None = None
+    created_at: object = None
+    metadata: dict = field(default_factory=dict)
+
+
+def _make_old_dt():
+    return dt.datetime(2020, 1, 1, tzinfo=dt.UTC)
+
+
+# ── _extract_ai_tag_names ─────────────────────────────────────────────────
+
+
+class TestExtractAiTagNames:
+    def test_extracts_all_keys(self):
+        ai_tags = {
+            "scene": ["工厂", "车间"],
+            "objects": ["产品"],
+            "action": ["演示"],
+            "shot": "特写",  # shot 不参与标签匹配
+            "has_text": False,
+        }
+        names = _extract_ai_tag_names(ai_tags)
+        assert names == {"工厂", "车间", "产品", "演示"}
+
+    def test_empty_dict(self):
+        assert _extract_ai_tag_names({}) == set()
+
+    def test_none_values(self):
+        ai_tags = {"scene": None, "objects": None, "action": None}
+        assert _extract_ai_tag_names(ai_tags) == set()
+
+    def test_case_insensitive(self):
+        ai_tags = {"scene": ["Factory"], "objects": [], "action": []}
+        names = _extract_ai_tag_names(ai_tags)
+        assert "factory" in names
+
+
+# ── _compute_ai_score ─────────────────────────────────────────────────────
+
+
+class TestComputeAiScore:
+    def test_single_clip_hit(self):
+        wanted = {"工厂", "演示"}
+        clips = [{"scene": ["工厂"], "objects": [], "action": ["演示"]}]
+        score = _compute_ai_score("a1", wanted, {"a1": clips})
+        # 命中 2 个 × 2.0 = 4.0
+        assert score == 2 * AI_TAG_WEIGHT
+
+    def test_multiple_clips_takes_best(self):
+        wanted = {"工厂", "演示"}
+        clips = [
+            {"scene": ["工厂"], "objects": [], "action": []},  # 1 hit = 2.0
+            {"scene": ["工厂"], "objects": [], "action": ["演示"]},  # 2 hits = 4.0
+        ]
+        score = _compute_ai_score("a1", wanted, {"a1": clips})
+        assert score == 2 * AI_TAG_WEIGHT  # best = 2 hits
+
+    def test_no_match(self):
+        wanted = {"美食"}
+        clips = [{"scene": ["工厂"], "objects": [], "action": ["演示"]}]
+        score = _compute_ai_score("a1", wanted, {"a1": clips})
+        assert score == 0.0
+
+    def test_no_clips_for_asset(self):
+        wanted = {"工厂"}
+        assert _compute_ai_score("a1", wanted, {}) == 0.0
+        assert _compute_ai_score("a1", wanted, None) == 0.0
+
+    def test_empty_wanted(self):
+        clips = [{"scene": ["工厂"], "objects": [], "action": []}]
+        assert _compute_ai_score("a1", set(), {"a1": clips}) == 0.0
+
+
+# ── match_assets_by_script_tags with AI tags ──────────────────────────────
+
+
+class TestMatchWithAiTags:
+    def test_ai_tag_hit_puts_in_matched(self):
+        """有 AI 标签命中 → 进入命中池."""
+        assets = [FakeAsset("a1", created_at=_make_old_dt())]
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": [], "action": []}]}
+
+        matched, unmatched = match_assets_by_script_tags(
+            assets,
+            script_tags=["工厂"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+
+        assert [a.id for a in matched] == ["a1"]
+        assert unmatched == []
+
+    def test_ai_tag_no_match_puts_in_unmatched(self):
+        """AI 标签未命中 → 进入未命中池."""
+        assets = [FakeAsset("a1", created_at=_make_old_dt())]
+        clip_ai_tags = {"a1": [{"scene": ["办公室"], "objects": [], "action": []}]}
+
+        matched, unmatched = match_assets_by_script_tags(
+            assets,
+            script_tags=["工厂"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+
+        assert matched == []
+        assert [a.id for a in unmatched] == ["a1"]
+
+    def test_asset_tag_still_works_without_ai_tags(self):
+        """无 AI 标签时，素材标签仍按权重 1.0 匹配."""
+        assets = [FakeAsset("a1", tags=["工厂"], created_at=_make_old_dt())]
+
+        matched, unmatched = match_assets_by_script_tags(
+            assets,
+            script_tags=["工厂"],
+        )
+
+        assert [a.id for a in matched] == ["a1"]
+
+    def test_mixed_ai_and_asset_tags(self):
+        """混合场景：一个素材有 AI 标签，另一个只有素材标签."""
+        assets = [
+            FakeAsset("a1", created_at=_make_old_dt()),  # AI 标签命中
+            FakeAsset("a2", tags=["工厂"], created_at=_make_old_dt()),  # 素材标签命中
+            FakeAsset("a3", tags=["美食"], created_at=_make_old_dt()),  # 无命中
+        ]
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": [], "action": []}]}
+
+        matched, unmatched = match_assets_by_script_tags(
+            assets,
+            script_tags=["工厂"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+
+        assert {a.id for a in matched} == {"a1", "a2"}
+        assert [a.id for a in unmatched] == ["a3"]
+
+    def test_ai_tag_and_asset_tag_both_hit(self):
+        """同一素材 AI 标签和素材标签都命中 → 仍在命中池."""
+        assets = [FakeAsset("a1", tags=["工厂"], created_at=_make_old_dt())]
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": [], "action": []}]}
+
+        matched, unmatched = match_assets_by_script_tags(
+            assets,
+            script_tags=["工厂"],
+            tag_names_by_id={"a1": ["工厂"]},
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+
+        assert [a.id for a in matched] == ["a1"]
+
+
+# ── compute_tag_match_score ───────────────────────────────────────────────
+
+
+class TestComputeTagMatchScore:
+    def test_ai_only_score(self):
+        """仅 AI 标签命中."""
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": [], "action": ["演示"]}]}
+        score = compute_tag_match_score(
+            "a1",
+            script_tags=["工厂", "演示"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+        # AI: 2 hits × 2.0 = 4.0; asset: 0; max = 2 × 3.0 = 6.0
+        assert abs(score - 4.0 / 6.0) < 0.01
+
+    def test_asset_only_score(self):
+        """仅素材标签命中."""
+        score = compute_tag_match_score(
+            "a1",
+            script_tags=["工厂", "演示"],
+            tag_names_by_id={"a1": ["工厂"]},
+        )
+        # AI: 0; asset: 1 hit × 1.0 = 1.0; max = 2 × 3.0 = 6.0
+        assert abs(score - 1.0 / 6.0) < 0.01
+
+    def test_both_ai_and_asset_score(self):
+        """AI 标签 + 素材标签同时命中."""
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": [], "action": []}]}
+        score = compute_tag_match_score(
+            "a1",
+            script_tags=["工厂", "演示"],
+            tag_names_by_id={"a1": ["工厂"]},
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+        # AI: 1 hit × 2.0 = 2.0; asset: 1 hit × 1.0 = 1.0; max = 2 × 3.0 = 6.0
+        assert abs(score - 3.0 / 6.0) < 0.01
+
+    def test_no_match_score_zero(self):
+        """无命中 → 得分 0."""
+        score = compute_tag_match_score(
+            "a1",
+            script_tags=["工厂"],
+            tag_names_by_id={"a1": ["美食"]},
+        )
+        assert score == 0.0
+
+    def test_full_match_score_one(self):
+        """全命中 → 得分接近 1.0."""
+        clip_ai_tags = {"a1": [{"scene": ["工厂"], "objects": ["产品"], "action": ["演示"]}]}
+        score = compute_tag_match_score(
+            "a1",
+            script_tags=["工厂", "产品", "演示"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+        )
+        # AI: 3 hits × 2.0 = 6.0; max = 3 × 3.0 = 9.0 → 6/9 = 0.667
+        # 注意：仅 AI 标签命中不可能达到 1.0（因为 max 包含素材权重）
+        assert score > 0.5
+
+    def test_empty_script_tags(self):
+        """空文案标签 → 得分 0."""
+        assert compute_tag_match_score("a1", script_tags=[]) == 0.0
+
+
+# ── pick_narrative_assets with AI tags ────────────────────────────────────
+
+
+class TestPickNarrativeWithAiTags:
+    def _assets(self):
+        old = _make_old_dt()
+        return [
+            FakeAsset("ai_match", created_at=old),  # AI 标签命中
+            FakeAsset("asset_match", tags=["工厂"], created_at=old),  # 素材标签命中
+            FakeAsset("no_match", tags=["美食"], created_at=old),  # 无命中
+        ]
+
+    def test_ai_match_prioritized(self):
+        """AI 标签命中的素材进入命中池."""
+        clip_ai_tags = {"ai_match": [{"scene": ["工厂"], "objects": [], "action": []}]}
+
+        picked = pick_narrative_assets(
+            self._assets(),
+            script_tags=["工厂"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+            limit=2,
+            rng=random.Random(0),
+        )
+
+        ids = {a.id for a in picked}
+        assert "ai_match" in ids
+        assert "asset_match" in ids
+
+    def test_fallback_when_no_ai_match(self):
+        """AI 标签和素材标签都未命中 → 降级."""
+        clip_ai_tags = {"ai_match": [{"scene": ["办公室"], "objects": [], "action": []}]}
+
+        picked = pick_narrative_assets(
+            self._assets(),
+            script_tags=["不存在"],
+            clip_ai_tags_by_asset=clip_ai_tags,
+            limit=2,
+            rng=random.Random(0),
+        )
+
+        assert len(picked) == 2  # 从全量中选取
+
+    def test_backward_compat_without_ai_tags(self):
+        """不传 clip_ai_tags_by_asset 时行为与之前完全一致."""
+        picked = pick_narrative_assets(
+            self._assets(),
+            script_tags=["工厂"],
+            limit=2,
+            rng=random.Random(0),
+        )
+
+        # 仅素材标签匹配
+        ids = {a.id for a in picked}
+        assert "asset_match" in ids
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])

tests/unit/test_gpu_lipsync_service.py

@@ -0,0 +1,258 @@
+"""GpuLipsyncService 单元测试 — 覆盖任务创建、轮询认领、结果上报、超时回退等核心逻辑.
+
+使用 SQLite 内存数据库，mock 掉存储层（不真实调用 OSS）。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from datetime import UTC, datetime, timedelta
+from unittest import mock
+
+import pytest
+
+# 确保 packages / apps/api 可导入
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+for p in (ROOT, os.path.join(ROOT, "apps", "api"), os.path.join(ROOT, "packages")):
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+# 强制使用内存 SQLite（避免依赖 PG）
+os.environ["APP_ENV"] = "development"
+os.environ["JWT_SECRET_KEY"] = "dev-secret-key-for-testing-00000000"
+os.environ["DATABASE_URL"] = "sqlite:///:memory:"
+os.environ["USE_IN_MEMORY_DB"] = "1"
+os.environ["GPU_WORKER_TOKEN"] = ""  # development 空 token 放行
+
+
+def _build_session():
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    # 使用 packages 的 Base
+    from packages.adapters.sqlalchemy_impl import models as _  # noqa: F401  # 触发 ORM 注册
+    from packages.adapters.sqlalchemy_impl.models import Base
+
+    engine = create_engine("sqlite:///:memory:", future=True)
+    Base.metadata.create_all(engine)
+    Session = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
+    return Session()
+
+
+@pytest.fixture
+def svc():
+    from app.services.gpu_lipsync_service import GpuLipsyncService
+
+    db = _build_session()
+    service = GpuLipsyncService(db)
+    # mock 存储签名（SQLite 测试无 OSS）
+    service.storage = mock.MagicMock()
+    service.storage.get_download_url.side_effect = (
+        lambda k, expires_seconds=3600: f"https://signed.example.com/download/{k}?e={expires_seconds}"
+    )
+    service.storage.get_upload_url.side_effect = (
+        lambda k, expires_seconds=3600, content_type="video/mp4": f"https://signed.example.com/upload/{k}?e={expires_seconds}"
+    )
+    return service
+
+
+# ── 创建任务 ──────────────────────────────────────────────────────
+
+
+def test_create_task(svc):
+    task = svc.create_task(
+        video_url="uploads/v.mp4",
+        audio_url="uploads/a.mp3",
+        lipsync_job_id="lip-1",
+        user_id="u-1",
+        project_id="p-1",
+    )
+    assert task.id
+    assert task.status == "pending"
+    assert task.lipsync_job_id == "lip-1"
+    assert task.attempt == 0
+    assert task.video_url == "uploads/v.mp4"
+
+
+# ── 轮询认领 ──────────────────────────────────────────────────────
+
+
+def test_poll_returns_none_when_empty(svc):
+    assert svc.poll_task("w-1") is None
+
+
+def test_poll_claims_pending_task(svc):
+    svc.create_task(video_url="uploads/v.mp4", audio_url="uploads/a.mp3")
+    claimed = svc.poll_task("w-1")
+    assert claimed is not None
+    assert claimed.status == "processing"
+    assert claimed.worker_id == "w-1"
+    assert claimed.attempt == 1
+    # 带签名 URL
+    assert claimed._signed_video_url.startswith("https://signed.example.com/download/")
+    assert claimed._signed_upload_url.startswith("https://signed.example.com/upload/")
+    # 再 poll 无任务
+    assert svc.poll_task("w-1") is None
+
+
+def test_poll_concurrent_claim_only_one_wins(svc):
+    """并发场景：两个 worker 同时 poll 只有一个能拿到任务（借助 update where status=pending）。"""
+    svc.create_task(video_url="v", audio_url="a")
+    t1 = svc.poll_task("w-1")
+    t2 = svc.poll_task("w-2")
+    assert t1 is not None
+    assert t2 is None
+
+
+# ── 结果上报 ──────────────────────────────────────────────────────
+
+
+def test_report_result_success(svc):
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")  # claim
+    done = svc.report_result(t.id, "w-1", success=True, duration_seconds=12.5)
+    assert done.status == "done"
+    assert done.result_duration == 12.5
+    assert done.result_url.startswith("gpu-lipsync/results/")
+    assert done.finished_at is not None
+
+
+def test_report_result_failure_requeues(svc):
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    failed = svc.report_result(t.id, "w-1", success=False, error_msg="MuseTalk crash")
+    assert failed.status == "pending"  # 仍在重试次数内 → 回队
+    assert failed.worker_id == ""
+    assert failed.started_at is None
+    assert "MuseTalk crash" in failed.error_msg
+
+
+def test_report_failure_exhausted_goes_failed(svc):
+    """失败达到 MAX_ATTEMPTS 后标记 failed，不再回队.
+
+    poll 成功会将 attempt 从 0 开始自增；
+    第 1/2 次失败回队，第 3 次失败（attempt==MAX_ATTEMPTS）置 failed。
+    """
+    from app.services import gpu_lipsync_service as mod
+
+    t = svc.create_task(video_url="v", audio_url="a")
+    # 模拟失败到上限：poll + fail 重复 MAX_ATTEMPTS 次
+    for i in range(mod.MAX_ATTEMPTS):
+        claimed = svc.poll_task(f"w-{i}")
+        assert claimed is not None, f"第 {i} 次 poll 应能拿到任务"
+        svc.report_result(t.id, claimed.worker_id, success=False, error_msg=f"fail {i}")
+        svc.db.refresh(t)
+        if i == mod.MAX_ATTEMPTS - 1:
+            assert t.status == "failed"
+        else:
+            assert t.status == "pending"
+
+
+# ── 心跳/超时回退 ─────────────────────────────────────────────────
+
+
+def test_timed_out_task_is_redispatched(svc):
+    """processing 超过 gpu_task_timeout_seconds 无心跳 → 回退 pending."""
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    svc.db.refresh(t)
+    assert t.status == "processing"
+    # 手动把 last_heartbeat_at 设到很久以前
+    t.last_heartbeat_at = datetime.now(UTC) - timedelta(seconds=svc.settings.gpu_task_timeout_seconds + 10)
+    svc.db.commit()
+    # 再次 poll 会触发 _recover_timed_out_tasks 把它回队
+    claimed = svc.poll_task("w-2")
+    assert claimed is not None
+    assert claimed.id == t.id
+    assert claimed.worker_id == "w-2"
+    assert claimed.attempt == 2  # 又认领了一次
+
+
+# ── Worker 注册 ───────────────────────────────────────────────────
+
+
+def test_register_worker_creates_then_updates(svc):
+    w = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=3500)
+    assert w.worker_id == "w-1"
+    assert w.gpu_name == "RTX2060"
+    w2 = svc.register_worker("w-1", hostname="pc1", gpu_name="RTX2060", free_vram_mb=2000)
+    assert w2.free_vram_mb == 2000  # 更新
+    assert w2.created_at == w.created_at  # 没新建
+
+
+def test_register_with_task_id_refreshes_task_heartbeat(svc):
+    """#1970 推理期心跳：register(task_id=...) 只刷新本 worker 的 processing 任务."""
+    from packages.adapters.sqlalchemy_impl.models import GpuWorkerModel
+
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    svc.db.refresh(t)
+    old_hb = t.last_heartbeat_at
+    assert t.status == "processing"
+    # 模拟时间流逝后心跳到达
+    svc.db.query(GpuWorkerModel).filter_by(worker_id="w-1").update(
+        {"last_heartbeat_at": old_hb - timedelta(seconds=300)}
+    )
+    svc.db.commit()
+    svc.register_worker("w-1", task_id=t.id)
+    svc.db.refresh(t)
+    assert t.last_heartbeat_at > old_hb
+    assert t.status == "processing"  # 心跳不改变状态
+    # worker 表心跳也被刷新
+    w = svc.db.query(GpuWorkerModel).filter_by(worker_id="w-1").one()
+    assert w.last_heartbeat_at > old_hb
+
+
+def test_register_task_heartbeat_ignores_finished_or_foreign_task(svc):
+    """任务已 done，或已被超时回收重新派发给别的 worker 时，旧心跳必须忽略."""
+    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel, GpuWorkerModel
+
+    # 场景 1：任务已完成 → register 带 task_id 不得改写任务心跳
+    t = svc.create_task(video_url="v", audio_url="a")
+    svc.poll_task("w-1")
+    done = svc.report_result(t.id, "w-1", success=True, duration_seconds=10.0)
+    hb_when_done = done.last_heartbeat_at
+    svc.register_worker("w-1", task_id=t.id)
+    svc.db.refresh(t)
+    assert t.status == "done"
+    assert t.last_heartbeat_at == hb_when_done  # 没被改写
+
+    # 场景 2：任务超时回收后被 w-2 重新认领，旧 worker w-1 的迟到心跳无效
+    t2 = svc.create_task(video_url="v2", audio_url="a2")
+    svc.poll_task("w-1")
+    svc.db.refresh(t2)
+    t2.last_heartbeat_at = datetime.now(UTC) - timedelta(days=1)
+    svc.db.commit()
+    claimed = svc.poll_task("w-2")  # 触发回收并由 w-2 重新认领
+    assert claimed is not None and claimed.id == t2.id
+    owner_hb = claimed.last_heartbeat_at
+    # 把 w-2 的 worker 心跳拨早，确认旧心跳不会影响任务归属
+    svc.db.query(GpuWorkerModel).filter_by(worker_id="w-2").update(
+        {"last_heartbeat_at": owner_hb - timedelta(seconds=600)}
+    )
+    svc.db.commit()
+    svc.register_worker("w-1", task_id=t2.id)  # 旧 worker 迟到心跳
+    svc.db.refresh(t2)
+    assert t2.worker_id == "w-2"
+    assert t2.status == "processing"
+    assert t2.last_heartbeat_at == owner_hb
+
+    # 场景 3：不存在的 task_id 不报错
+    svc.register_worker("w-1", task_id="nonexistent-id")
+    assert svc.db.get(GpuLipsyncTaskModel, "nonexistent-id") is None
+
+
+def test_default_gpu_task_timeout_is_900(svc):
+    """#1970 默认超时 300→900，覆盖 RTX2060 长视频推理."""
+    assert svc.settings.gpu_task_timeout_seconds == 900
+
+
+# ── get_by_lipsync_job ─────────────────────────────────────────────
+
+
+def test_get_by_lipsync_job_returns_latest(svc):
+    svc.create_task(video_url="v", audio_url="a", lipsync_job_id="lip-1")
+    svc.create_task(video_url="v", audio_url="a", lipsync_job_id="lip-1")
+    latest = svc.get_by_lipsync_job("lip-1")
+    assert latest is not None

tests/unit/test_lipsync_gpu_integration.py

@@ -0,0 +1,155 @@
+"""LipsyncService GPU 路径集成测试."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture()
+def fake_db():
+    db = MagicMock()
+    return db
+
+
+@pytest.fixture()
+def fake_mediakit():
+    client = MagicMock()
+    client.submit_lipsync.return_value = {"task_id": "mk-task-1"}
+    return client
+
+
+def _make_job(video_url="oss://video.mp4", audio_url="oss://audio.wav"):
+    job = MagicMock()
+    job.id = "job-1"
+    job.user_id = "u1"
+    job.project_id = "p1"
+    job.video_url = video_url
+    job.audio_url = audio_url
+    job.enable_video_loop = True
+    job.script_text = ""
+    job.sentence_timings = None
+    return job
+
+
+def _make_svc(db, mediakit, use_gpu=False):
+    from app.services.lipsync_service import LipsyncService
+
+    svc = LipsyncService(db=db, client=mediakit)
+    svc.settings.use_gpu_lipsync = use_gpu
+    svc._sign_media_url = lambda u: (u or "") + "?signed"
+    return svc
+
+
+class TestGpuFallback:
+    def test_switch_off_uses_mediakit(self, fake_db, fake_mediakit):
+        """开关关闭时直接走 MediaKit，不调用 _submit_to_gpu."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=False)
+        job = _make_job()
+        with patch.object(svc, "_submit_to_gpu") as m_sub:
+            svc._submit_audio_direct(job=job)
+            m_sub.assert_not_called()
+        fake_mediakit.submit_lipsync.assert_called_once()
+        assert job.status == "submitted"
+
+    def test_switch_on_no_worker_falls_back(self, fake_db, fake_mediakit):
+        """开关打开但 has_available_worker=False → 回退 MediaKit."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
+        fake_gpu_svc = MagicMock()
+        fake_gpu_svc.has_available_worker.return_value = False
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+            job = _make_job()
+            svc._submit_audio_direct(job=job)
+        fake_gpu_svc.create_task.assert_not_called()
+        fake_mediakit.submit_lipsync.assert_called_once()
+        assert job.status == "submitted"
+
+    def test_gpu_success_marks_completed(self, fake_db, fake_mediakit):
+        """GPU 路径成功：job 直接 completed，不调 MediaKit."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
+        gpu_done = MagicMock(
+            id="gpu-task-1",
+            status="done",
+            result_url="oss://gpu-results/r.mp4",
+            result_duration=12.5,
+        )
+        fake_gpu_svc = MagicMock()
+        fake_gpu_svc.has_available_worker.return_value = True
+        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1")
+        fake_gpu_svc.wait_for_result.return_value = gpu_done
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+            job = _make_job()
+            svc._submit_audio_direct(job=job)
+        fake_gpu_svc.create_task.assert_called_once()
+        fake_mediakit.submit_lipsync.assert_not_called()
+        assert job.status == "completed"
+        assert job.output_duration == 12.5
+        assert "?signed" in job.output_video_url
+        fake_db.commit.assert_called()
+
+    def test_gpu_timeout_falls_back(self, fake_db, fake_mediakit):
+        """wait_for_result 返回 None（超时）→ 回退 MediaKit."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
+        fake_gpu_svc = MagicMock()
+        fake_gpu_svc.has_available_worker.return_value = True
+        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t")
+        fake_gpu_svc.wait_for_result.return_value = None
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+            job = _make_job()
+            svc._submit_audio_direct(job=job)
+        fake_mediakit.submit_lipsync.assert_called_once()
+        assert job.status == "submitted"
+
+    def test_gpu_failed_status_falls_back(self, fake_db, fake_mediakit):
+        """GPU 终态 failed → 回退 MediaKit."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
+        fake_gpu_svc = MagicMock()
+        fake_gpu_svc.has_available_worker.return_value = True
+        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t")
+        fake_gpu_svc.wait_for_result.return_value = MagicMock(status="failed", error_msg="musetalk crash")
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+            job = _make_job()
+            svc._submit_audio_direct(job=job)
+        fake_mediakit.submit_lipsync.assert_called_once()
+        assert job.status == "submitted"
+
+    def test_gpu_exception_falls_back(self, fake_db, fake_mediakit):
+        """GPU 路径抛异常 → 回退 MediaKit."""
+        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
+        fake_gpu_svc = MagicMock()
+        fake_gpu_svc.has_available_worker.return_value = True
+        fake_gpu_svc.create_task.side_effect = RuntimeError("DB down")
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+            job = _make_job()
+            svc._submit_audio_direct(job=job)
+        fake_mediakit.submit_lipsync.assert_called_once()
+        assert job.status == "submitted"
+
+
+class TestGpuServiceHelpers:
+    """GpuLipsyncService.has_available_worker 测试."""
+
+    def test_no_workers(self, fake_db):
+        from app.services.gpu_lipsync_service import GpuLipsyncService
+
+        svc = GpuLipsyncService(db=fake_db)
+        fake_db.query.return_value.filter.return_value.first.return_value = None
+        assert svc.has_available_worker() is False
+
+    def test_fresh_worker_available(self, fake_db):
+        from app.services.gpu_lipsync_service import GpuLipsyncService
+
+        svc = GpuLipsyncService(db=fake_db)
+        svc.settings.gpu_worker_stale_seconds = 300
+        # 模拟SQL filter条件成立 → first() 返回非None
+        fake_db.query.return_value.filter.return_value.first.return_value = MagicMock()
+        assert svc.has_available_worker() is True
+
+    def test_stale_worker_unavailable(self, fake_db):
+        from app.services.gpu_lipsync_service import GpuLipsyncService
+
+        svc = GpuLipsyncService(db=fake_db)
+        # filter条件不成立（stale）→ first() 返回None
+        fake_db.query.return_value.filter.return_value.first.return_value = None
+        assert svc.has_available_worker() is False