style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

根因：TTS+MediaKit都成功了，但状态回写完全依赖前端轮询触发的FastAPI background task，
background task中db.commit()异常时无try/except且无日志，静默失败导致completed永远写不回DB。

修复三层防御：
1. refresh_job_status: db.commit()包try/except+rollback，失败打完整堆栈日志
2. GET /lipsync/jobs/{id}: 距上次更新超30秒时同步刷新（不挂后台），确保stale任务被及时拉取
3. 新增Celery兜底轮询任务lipsync_tts.poll_mediakit_status：TTS提交MediaKit成功后
   链式触发，worker主动轮询到终态（最多10分钟），不依赖前端

apps/api/app/api/routes/lipsync.py

@@ -144,7 +144,22 @@ def get_lipsync_job(
         raise HTTPException(status_code=404, detail="任务不存在")
 
     if job.status not in ("completed", "failed"):
-        background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
+        # 如果距上次更新超过 30 秒，同步刷新一次（避免 background task 静默失败导致永久卡 running）；
+        # 否则挂后台异步刷新（避免阻塞前端轮询）。
+        from datetime import datetime, timezone
+        now = datetime.now(timezone.utc)
+        stale = (
+            job.updated_at is None
+            or (now - job.updated_at).total_seconds() > 30
+        )
+        if stale:
+            try:
+                job = svc.refresh_job_status(job_id, current_user.user.id) or job
+            except Exception as exc:  # noqa: BLE001
+                logger.error("同步刷新对口型状态失败 job_id=%s err=%s", job_id, exc, exc_info=True)
+                background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
+        else:
+            background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
 
     return job
 

apps/api/app/services/lipsync_service.py

@@ -310,26 +310,40 @@ class LipsyncService:
         mk_status = status_data.get("status", STATUS_RUNNING)
         logger.info("MediaKit 对口型状态 [%s]: %s", job_id, mk_status)
 
-        if mk_status == STATUS_COMPLETED:
-            result = status_data.get("result", {})
-            job.status = STATUS_COMPLETED
-            output_url = result.get("video_url", "")
-            # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL）
-            job.output_video_url = self._persist_output_video(output_url, job_id, user_id)
-            job.output_duration = result.get("duration", 0.0)
-            job.completed_at = datetime.now(timezone.utc)
-        elif mk_status == STATUS_FAILED:
-            error = status_data.get("error", {})
-            job.status = "failed"
-            job.error_message = error.get("message", "任务执行失败")
-            job.error_code = error.get("code", "TaskFailed")
-            job.completed_at = datetime.now(timezone.utc)
-        else:
-            # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted
-            if isinstance(mk_status, str) and mk_status:
-                job.status = mk_status
-        job.updated_at = datetime.now(timezone.utc)
-        self.db.commit()
+        try:
+            if mk_status == STATUS_COMPLETED:
+                result = status_data.get("result", {})
+                job.status = STATUS_COMPLETED
+                output_url = result.get("video_url", "")
+                # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL）
+                job.output_video_url = self._persist_output_video(output_url, job_id, user_id)
+                job.output_duration = result.get("duration", 0.0)
+                job.completed_at = datetime.now(timezone.utc)
+            elif mk_status == STATUS_FAILED:
+                error = status_data.get("error", {})
+                job.status = "failed"
+                job.error_message = error.get("message", "任务执行失败")
+                job.error_code = error.get("code", "TaskFailed")
+                job.completed_at = datetime.now(timezone.utc)
+            else:
+                # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted
+                if isinstance(mk_status, str) and mk_status:
+                    job.status = mk_status
+            job.updated_at = datetime.now(timezone.utc)
+            self.db.commit()
+        except Exception as exc:  # noqa: BLE001 - DB 提交失败必须记录日志并重试，否则后台任务静默失败
+            logger.error(
+                "refresh_job_status 提交 DB 失败 job_id=%s mk_status=%s err=%s",
+                job_id,
+                mk_status,
+                exc,
+                exc_info=True,
+            )
+            try:
+                self.db.rollback()
+            except Exception:
+                pass
+            # DB commit 失败不 raise，返回当前 job 对象让下次轮询再试
         self.db.refresh(job)
         return job
 

apps/api/app/tasks/lipsync_tts.py

@@ -207,6 +207,14 @@ def tts_synthesize_and_submit(
 
         db.commit()
 
+        # 4. 链式触发 Celery 兜底轮询：MediaKit 提交成功后由 worker 主动拉取状态到终态，
+        # 不依赖前端轮询触发的 FastAPI background task（后台任务可能静默失败导致永久卡 running）
+        if job.status == "submitted" and job.mediakit_task_id:
+            poll_mediakit_status.apply_async(
+                kwargs={"job_id": job_id, "user_id": user_id},
+                countdown=10,  # 10 秒后开始轮询（给 MediaKit 一点处理时间）
+            )
+
     except Exception:
         logger.exception("[lipsync_tts] 未预期的异常: job_id=%s", job_id)
         try:
@@ -221,3 +229,88 @@ def tts_synthesize_and_submit(
             logger.exception("[lipsync_tts] 回写失败状态时异常: job_id=%s", job_id)
     finally:
         db.close()
+
+
+@shared_task(
+    bind=True,
+    name="lipsync_tts.poll_mediakit_status",
+    max_retries=60,  # 最多轮询 60 次
+    default_retry_delay=10,  # 每次间隔 10 秒（总兜底时长 10 分钟）
+)
+def poll_mediakit_status(self, job_id: str, user_id: str):
+    """Celery 兜底轮询：TTS 提交 MediaKit 后，由 worker 主动拉取状态直到终态。
+
+    不依赖前端轮询，避免 background task 静默失败导致任务永久卡 running/submitted。
+    """
+    from sqlalchemy.orm import Session as DBSession
+
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+
+    try:
+        from worker_app.db import SessionLocal  # type: ignore
+    except Exception:  # noqa: BLE001
+        from app.db import SessionLocal  # type: ignore
+
+    db: DBSession = SessionLocal()
+    try:
+        job = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id, LipsyncJobModel.user_id == user_id).first()
+        if job is None:
+            logger.warning("[lipsync_poll] Job not found: job_id=%s", job_id)
+            return
+
+        # 已终态，不需要再轮询
+        if job.status in ("completed", "failed", "cancelled"):
+            return
+
+        if not job.mediakit_task_id:
+            logger.warning("[lipsync_poll] Job has no mediakit_task_id: job_id=%s status=%s", job_id, job.status)
+            return
+
+        from app.services.mediakit_client import MediaKitError, get_mediakit_client
+
+        client = get_mediakit_client()
+        try:
+            status_data = client.get_task_status(job.mediakit_task_id)
+        except MediaKitError as exc:
+            logger.warning("[lipsync_poll] 拉取 MediaKit 状态失败，将重试: job_id=%s err=%s", job_id, exc)
+            raise self.retry(exc=exc)
+
+        mk_status = status_data.get("status", "running")
+
+        if mk_status == "succeeded":
+            # 复用 LipsyncService 的持久化逻辑
+            from app.services.lipsync_service import LipsyncService
+
+            svc = LipsyncService(db)
+            result = status_data.get("result", {})
+            job.status = "completed"
+            output_url = result.get("video_url", "")
+            job.output_video_url = svc._persist_output_video(output_url, job_id, user_id)
+            job.output_duration = result.get("duration", 0.0)
+            job.completed_at = datetime.now(timezone.utc)
+            job.updated_at = datetime.now(timezone.utc)
+            db.commit()
+            logger.info("[lipsync_poll] 任务完成: job_id=%s", job_id)
+        elif mk_status in ("failed", "error"):
+            error = status_data.get("error", {})
+            job.status = "failed"
+            job.error_message = error.get("message", "任务执行失败")
+            job.error_code = error.get("code", "TaskFailed")
+            job.completed_at = datetime.now(timezone.utc)
+            job.updated_at = datetime.now(timezone.utc)
+            db.commit()
+            logger.info("[lipsync_poll] 任务失败: job_id=%s err=%s", job_id, job.error_message)
+        else:
+            # 中间状态，更新时间戳，继续重试
+            job.updated_at = datetime.now(timezone.utc)
+            if isinstance(mk_status, str) and mk_status:
+                job.status = mk_status
+            db.commit()
+            logger.debug("[lipsync_poll] 任务仍在 %s，继续轮询: job_id=%s", mk_status, job_id)
+            raise self.retry()
+    except Exception as exc:
+        logger.exception("[lipsync_poll] 未预期异常: job_id=%s", job_id)
+        db.rollback()
+        raise self.retry(exc=exc)
+    finally:
+        db.close()