fix(backend): #1867 对口型卡running三层防御修复（rebase到最新develop） (#1916)

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

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

@@ -193,7 +193,23 @@ 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 as _dt
+        from datetime import timezone as _tz
+
+        _now = _dt.now(_tz.utc)
+        _stale = job.updated_at is None or (_now - job.updated_at).total_seconds() > 30
+        if _stale:
+            try:
+                refreshed = svc.refresh_job_status(job_id, current_user.user.id)
+                if refreshed is not None:
+                    job = refreshed
+            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

@@ -489,39 +489,52 @@ 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
-            temp_url = result.get("video_url", "")
-            job.output_video_url = temp_url
-            job.output_duration = result.get("duration", 0.0)
-            job.completed_at = datetime.now(timezone.utc)
+        try:
+            if mk_status == STATUS_COMPLETED:
+                result = status_data.get("result", {})
+                job.status = STATUS_COMPLETED
+                temp_url = result.get("video_url", "")
+                job.output_video_url = temp_url
+                job.output_duration = result.get("duration", 0.0)
+                job.completed_at = datetime.now(timezone.utc)
+                job.updated_at = datetime.now(timezone.utc)
+                self.db.commit()
+                # 异步转存自家 OSS
+                try:
+                    from app.tasks.lipsync_tts import persist_output_video_task
+
+                    persist_output_video_task.apply_async(args=(job_id, user_id, temp_url))
+                except Exception as exc:
+                    logger.warning(
+                        "提交输出视频异步转存任务失败，保留临时 URL: job_id=%s err=%s",
+                        job_id,
+                        exc,
+                    )
+            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()
-            # 异步转存自家 OSS
+        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:
-                from app.tasks.lipsync_tts import persist_output_video_task
-
-                persist_output_video_task.apply_async(args=(job_id, user_id, temp_url))
-            except Exception as exc:
-                logger.warning(
-                    "提交输出视频异步转存任务失败，保留临时 URL: job_id=%s err=%s",
-                    job_id,
-                    exc,
-                )
-            self.db.refresh(job)
-            return job
-        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:
-            if isinstance(mk_status, str) and mk_status:
-                job.status = mk_status
-        job.updated_at = datetime.now(timezone.utc)
-        self.db.commit()
+                self.db.rollback()
+            except Exception:
+                pass
+            # DB commit 失败不 raise，返回当前 job 对象让下次轮询再试
         self.db.refresh(job)
         return job
 

apps/api/app/tasks/lipsync_tts.py

@@ -282,6 +282,18 @@ def tts_synthesize_and_submit(
             job.error_code = exc.code
             logger.error("[lipsync_tts] 提交 MediaKit 失败: job_id=%s err=%s", job_id, exc)
 
+        # 三层防御 ③：链式触发 Celery 兜底轮询——MediaKit 提交成功后由 worker
+        # 主动拉取状态到终态，不依赖前端轮询触发的 FastAPI background task
+        # （background task 可能静默失败导致永久卡 running）。
+        if job.status == "submitted" and job.mediakit_task_id:
+            try:
+                poll_mediakit_status.apply_async(
+                    kwargs={"job_id": job_id, "user_id": user_id},
+                    countdown=10,  # 10 秒后开始轮询，给 MediaKit 一点处理时间
+                )
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("[lipsync_tts] 提交兜底轮询任务失败（不影响主流程）: job_id=%s err=%s", job_id, exc)
+
         db.commit()
 
     except Exception:
@@ -300,6 +312,99 @@ def tts_synthesize_and_submit(
         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.lipsync_service import STATUS_COMPLETED as _SC
+        from app.services.lipsync_service import STATUS_FAILED as _SF
+        from app.services.lipsync_service import LipsyncService
+        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) from exc
+
+        mk_status = status_data.get("status", "running")
+
+        if mk_status in ("succeeded", _SC):
+
+            svc = LipsyncService(db)
+            result = status_data.get("result", {})
+            job.status = "completed"
+            output_url = result.get("video_url", "")
+            try:
+                job.output_video_url = svc._persist_output_video(output_url, job_id, user_id)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("[lipsync_poll] 转存 OSS 失败，保留临时 URL: job_id=%s err=%s", job_id, exc)
+                job.output_video_url = output_url
+            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", _SF):
+            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)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        raise self.retry(exc=exc) from exc
+    finally:
+        db.close()
+
+
 @shared_task(
     name="lipsync_tts.persist_output_video",
     max_retries=2,

tests/unit/test_lipsync_routes.py

@@ -227,9 +227,7 @@ class TestLipsyncServiceUnit:
         mock_repo = MagicMock()
         mock_repo.get.return_value = None  # 预置音色，原样返回 voice_id
 
-        with patch(
-            "app.services.lipsync_service.tts_synthesize_and_submit"
-        ) as mock_task:
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
             mock_task.apply_async.return_value = MagicMock(id="celery-task-123")
 
             svc = LipsyncService(
@@ -275,9 +273,7 @@ class TestLipsyncServiceUnit:
             voice_clone_repo=mock_repo,
         )
 
-        with patch(
-            "app.services.lipsync_service.tts_synthesize_and_submit"
-        ) as mock_task:
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
             mock_task.apply_async.return_value = MagicMock(id="celery-task-456")
 
             job = svc.create_job(
@@ -444,9 +440,7 @@ class TestLipsyncServiceUnit:
         mock_repo = MagicMock()
         mock_repo.get.return_value = None
 
-        with patch(
-            "app.services.lipsync_service.tts_synthesize_and_submit"
-        ) as mock_task:
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
             mock_task.apply_async.return_value = MagicMock(id="celery-task-789")
 
             svc = LipsyncService(
@@ -568,9 +562,7 @@ class TestErrorHandling:
             voice_clone_repo=mock_repo,
         )
 
-        with patch(
-            "app.services.lipsync_service.tts_synthesize_and_submit"
-        ) as mock_task:
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
             mock_task.apply_async.return_value = MagicMock(id="celery-task-789")
 
             # TTS 模式下 create_job 不再同步失败
@@ -718,3 +710,119 @@ class TestSignMediaUrl403Fix:
         kw = mock_mediakit.submit_lipsync.call_args.kwargs
         assert kw["video_url"] == signed_video
         assert kw["audio_url"] == signed_audio
+
+
+class TestLipsyncRouteStaleRefresh:
+    """#1867 三层防御 ①：路由层 stale>30s 同步刷新 / 新鲜走后台 / 异常 fallback 单测.
+
+    直接调用路由函数 get_lipsync_job（不经过 TestClient/HTTP 栈），mock 掉
+    current_user、svc、background 三个依赖，覆盖 stale/sync/exception/terminal/404 五个分支。
+    """
+
+    def _call(self, job, svc, background=None):
+        from app.api.routes.lipsync import get_lipsync_job
+
+        user = MagicMock()
+        user.user.id = "user-1"
+        bg = background if background is not None else MagicMock()
+        return get_lipsync_job(job_id="job-1", background=bg, current_user=user, svc=svc)
+
+    def test_stale_job_triggers_sync_refresh(self, mock_mediakit, mock_cosyvoice):
+        """updated_at 超过 30 秒 → 同步调用 svc.refresh_job_status，不走 background."""
+        from datetime import datetime, timedelta, timezone
+
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = _make_mock_job(status="submitted")
+        mock_job.updated_at = datetime.now(timezone.utc) - timedelta(seconds=45)
+        refreshed_job = _make_mock_job(status="completed", output_video_url="https://cdn/out.mp4", output_duration=12.0)
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_filter = MagicMock()
+        mock_filter.first.return_value = mock_job
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+
+        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc.get_job = MagicMock(return_value=mock_job)
+        svc.refresh_job_status = MagicMock(return_value=refreshed_job)
+        bg = MagicMock()
+
+        result = self._call(mock_job, svc, bg)
+        svc.refresh_job_status.assert_called_once_with("job-1", "user-1")
+        bg.add_task.assert_not_called()
+        assert result is refreshed_job
+
+    def test_fresh_job_uses_background_refresh(self, mock_mediakit, mock_cosyvoice):
+        """updated_at 在 30 秒内 → 直接挂后台刷新，不阻塞."""
+        from datetime import datetime, timezone
+
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = _make_mock_job(status="submitted")
+        mock_job.updated_at = datetime.now(timezone.utc)
+        mock_db = MagicMock()
+        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc.get_job = MagicMock(return_value=mock_job)
+        svc.refresh_job_status = MagicMock()
+        bg = MagicMock()
+
+        result = self._call(mock_job, svc, bg)
+        svc.refresh_job_status.assert_not_called()
+        bg.add_task.assert_called_once()
+        args, _ = bg.add_task.call_args
+        assert args[0] == svc.refresh_job_status
+        assert args[1:] == ("job-1", "user-1")
+        assert result is mock_job
+
+    def test_completed_job_skips_refresh(self, mock_mediakit, mock_cosyvoice):
+        """终态 completed → 不刷新，直接返回."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = _make_mock_job(status="completed", output_video_url="https://cdn/out.mp4", output_duration=12.0)
+        mock_db = MagicMock()
+        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc.get_job = MagicMock(return_value=mock_job)
+        svc.refresh_job_status = MagicMock()
+        bg = MagicMock()
+
+        result = self._call(mock_job, svc, bg)
+        svc.refresh_job_status.assert_not_called()
+        bg.add_task.assert_not_called()
+        assert result is mock_job
+
+    def test_sync_refresh_exception_falls_back_to_background(self, mock_mediakit, mock_cosyvoice):
+        """同步刷新抛异常 → 捕获后 fallback 到 background.add_task，接口仍正常返回 job."""
+        from datetime import datetime, timedelta, timezone
+
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = _make_mock_job(status="submitted")
+        mock_job.updated_at = datetime.now(timezone.utc) - timedelta(seconds=60)
+        mock_db = MagicMock()
+        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc.get_job = MagicMock(return_value=mock_job)
+        svc.refresh_job_status = MagicMock(side_effect=ConnectionError("db down"))
+        bg = MagicMock()
+
+        result = self._call(mock_job, svc, bg)
+        svc.refresh_job_status.assert_called_once_with("job-1", "user-1")
+        bg.add_task.assert_called_once()
+        args, _ = bg.add_task.call_args
+        assert args[0] == svc.refresh_job_status
+        assert args[1:] == ("job-1", "user-1")
+        assert result is mock_job
+
+    def test_job_not_found_raises_404(self, mock_mediakit, mock_cosyvoice):
+        """svc.get_job 返回 None → 抛 HTTPException(404)."""
+        import pytest
+        from app.services.lipsync_service import LipsyncService
+        from fastapi import HTTPException
+
+        mock_db = MagicMock()
+        svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice)
+        svc.get_job = MagicMock(return_value=None)
+        bg = MagicMock()
+        with pytest.raises(HTTPException) as exc_info:
+            self._call(None, svc, bg)
+        assert exc_info.value.status_code == 404

tests/unit/test_lipsync_tts.py

@@ -626,3 +626,351 @@ class TestLipsyncServiceRefreshCompletedAsyncPersist:
         assert result.status == "completed"
         assert result.output_video_url == "https://temp.mk/out2.mp4"
         fake_persist_task.apply_async.assert_called_once()
+
+
+class TestRefreshJobStatusDbRollback:
+    """#1867 三层防御 ②：refresh_job_status DB commit 失败时 rollback 不 raise."""
+
+    def test_refresh_db_commit_failure_rollback_and_return(self):
+        """commit 抛异常 → rollback，不 raise，返回当前 job 对象让下次轮询重试."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = MagicMock()
+        mock_job.id = "job-rollback"
+        mock_job.user_id = "u"
+        mock_job.mediakit_task_id = "mk"
+        mock_job.status = "submitted"
+        mock_job.output_video_url = ""
+        mock_job.output_duration = 0.0
+
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_filter = MagicMock()
+        mock_filter.first.return_value = mock_job
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+        mock_db.commit.side_effect = Exception("DB connection lost")
+
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "completed",
+            "result": {"video_url": "https://temp.mk/out.mp4", "duration": 5.0},
+        }
+
+        # refresh_job_status 内部懒 import persist_output_video_task
+        # 用 patch 替换模块属性，create=True 因为原模块里它是 shared_task，patch 后是 MagicMock
+        with patch("app.tasks.lipsync_tts.persist_output_video_task", MagicMock(), create=True):
+            svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock())
+            result_job = svc.refresh_job_status("job-rollback", "u")
+
+        assert result_job is mock_job
+        mock_db.rollback.assert_called()
+
+    def test_refresh_intermediate_status_commits(self):
+        """中间状态（running/processing 等）→ 更新 status 和 updated_at，正常 commit."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = MagicMock()
+        mock_job.id = "job-mid"
+        mock_job.user_id = "u"
+        mock_job.mediakit_task_id = "mk"
+        mock_job.status = "submitted"
+        mock_job.output_video_url = ""
+        mock_job.output_duration = 0.0
+
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_filter = MagicMock()
+        mock_filter.first.return_value = mock_job
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {"status": "processing"}
+
+        svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock())
+        result_job = svc.refresh_job_status("job-mid", "u")
+        assert result_job is mock_job
+        assert mock_job.status == "processing"
+        mock_db.commit.assert_called()
+        mock_db.rollback.assert_not_called()
+
+    def test_refresh_failed_commits_error_info(self):
+        """failed 状态 → 写入 error_message/error_code/completed_at，commit."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_job = MagicMock()
+        mock_job.id = "job-fail"
+        mock_job.user_id = "u"
+        mock_job.mediakit_task_id = "mk"
+        mock_job.status = "submitted"
+        mock_job.output_video_url = ""
+        mock_job.output_duration = 0.0
+
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_filter = MagicMock()
+        mock_filter.first.return_value = mock_job
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "failed",
+            "error": {"code": "BadAudio", "message": "音频损坏"},
+        }
+
+        svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock())
+        result_job = svc.refresh_job_status("job-fail", "u")
+        assert result_job.status == "failed"
+        assert mock_job.error_message == "音频损坏"
+        assert mock_job.error_code == "BadAudio"
+        mock_db.commit.assert_called()
+
+
+class TestPollMediakitStatus:
+    """#1867 三层防御 ③：Celery 兜底轮询任务单元测试."""
+
+    def _make_session(self, job):
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_filter = MagicMock()
+        mock_filter.first.return_value = job
+        mock_query.filter.return_value = mock_filter
+        mock_db.query.return_value = mock_query
+        return mock_db
+
+    def _patch_sessions(self, mock_db):
+        """用 patch.dict 注入 fake SessionLocal 到 worker_app.db 和 app.db."""
+        from types import ModuleType
+
+        fake_app_db = ModuleType("app.db")
+        fake_worker_db = ModuleType("worker_app.db")
+        fake_sl = MagicMock(return_value=mock_db)
+        fake_app_db.SessionLocal = fake_sl
+        fake_worker_db.SessionLocal = fake_sl
+        fake_worker_pkg = ModuleType("worker_app")
+        return patch.dict(
+            sys.modules,
+            {
+                "worker_app": fake_worker_pkg,
+                "worker_app.db": fake_worker_db,
+                "app.db": fake_app_db,
+            },
+        )
+
+    def _run(self, mock_db, job_id="j", user_id="u1", fake_self=None):
+        from app.tasks.lipsync_tts import poll_mediakit_status
+
+        self_ref = fake_self if fake_self is not None else MagicMock()
+        real = poll_mediakit_status._get_current_object()
+        return real.run.__func__(self_ref, job_id=job_id, user_id=user_id), self_ref
+
+    def test_poll_job_not_found_returns(self):
+        mock_db = self._make_session(None)
+        with self._patch_sessions(mock_db):
+            self._run(mock_db, job_id="missing")
+        mock_db.close.assert_called_once()
+
+    def test_poll_terminal_completed_skips_mediakit(self):
+        mock_job = MagicMock(status="completed", mediakit_task_id="mk-1")
+        mock_db = self._make_session(mock_job)
+        with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg:
+            self._run(mock_db)
+            fg.assert_not_called()
+        mock_db.close.assert_called_once()
+
+    def test_poll_terminal_failed_skips(self):
+        mock_job = MagicMock(status="failed", mediakit_task_id="mk-1")
+        mock_db = self._make_session(mock_job)
+        with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg:
+            self._run(mock_db, job_id="j1b")
+            fg.assert_not_called()
+
+    def test_poll_terminal_cancelled_skips(self):
+        mock_job = MagicMock(status="cancelled", mediakit_task_id="mk-1")
+        mock_db = self._make_session(mock_job)
+        with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg:
+            self._run(mock_db, job_id="j1c")
+            fg.assert_not_called()
+
+    def test_poll_no_mk_task_id_returns(self):
+        mock_job = MagicMock(status="submitted", mediakit_task_id=None)
+        mock_db = self._make_session(mock_job)
+        with self._patch_sessions(mock_db):
+            self._run(mock_db, job_id="j6")
+        mock_db.close.assert_called_once()
+
+    def test_poll_succeeded_marks_completed(self):
+        mock_job = MagicMock(status="submitted", mediakit_task_id="mk-2", output_video_url="", output_duration=0.0)
+        mock_db = self._make_session(mock_job)
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "succeeded",
+            "result": {"video_url": "https://temp/x.mp4", "duration": 12.0},
+        }
+        with (
+            self._patch_sessions(mock_db),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client),
+            patch("app.services.lipsync_service.LipsyncService") as MS,
+        ):
+            svc = MS.return_value
+            svc._persist_output_video.return_value = "https://oss/x.mp4"
+            self._run(mock_db, job_id="j2")
+        assert mock_job.status == "completed"
+        assert mock_job.output_video_url == "https://oss/x.mp4"
+        assert mock_job.output_duration == 12.0
+        svc._persist_output_video.assert_called_once_with("https://temp/x.mp4", "j2", "u1")
+        mock_db.commit.assert_called()
+
+    def test_poll_succeeded_persist_fail_fallback(self):
+        mock_job = MagicMock(status="submitted", mediakit_task_id="mk-3", output_video_url="", output_duration=0.0)
+        mock_db = self._make_session(mock_job)
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "succeeded",
+            "result": {"video_url": "https://temp/y.mp4", "duration": 8.0},
+        }
+        with (
+            self._patch_sessions(mock_db),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client),
+            patch("app.services.lipsync_service.LipsyncService") as MS,
+        ):
+            svc = MS.return_value
+            svc._persist_output_video.side_effect = Exception("OSS down")
+            self._run(mock_db, job_id="j3")
+        assert mock_job.status == "completed"
+        assert mock_job.output_video_url == "https://temp/y.mp4"
+
+    def test_poll_failed_marks_failed(self):
+        mock_job = MagicMock(status="running", mediakit_task_id="mk-4", error_message="", error_code="")
+        mock_db = self._make_session(mock_job)
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "failed",
+            "error": {"code": "TaskFailed", "message": "音频错误"},
+        }
+        with (
+            self._patch_sessions(mock_db),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client),
+        ):
+            self._run(mock_db, job_id="j4")
+        assert mock_job.status == "failed"
+        assert mock_job.error_message == "音频错误"
+        assert mock_job.error_code == "TaskFailed"
+
+    def test_poll_error_status_marks_failed(self):
+        mock_job = MagicMock(status="submitted", mediakit_task_id="mk-4b", error_message="", error_code="")
+        mock_db = self._make_session(mock_job)
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {
+            "status": "error",
+            "error": {"code": "Internal", "message": "内部错误"},
+        }
+        with (
+            self._patch_sessions(mock_db),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client),
+        ):
+            self._run(mock_db, job_id="j4b")
+        assert mock_job.status == "failed"
+        assert mock_job.error_message == "内部错误"
+
+    def test_poll_running_triggers_retry(self):
+        mock_job = MagicMock(status="submitted", mediakit_task_id="mk-5")
+        mock_db = self._make_session(mock_job)
+        mock_client = MagicMock()
+        mock_client.get_task_status.return_value = {"status": "running"}
+
+        class _Retry(Exception):
+            pass
+
+        fake_self = MagicMock()
+        fake_self.retry.side_effect = _Retry()
+        with (
+            self._patch_sessions(mock_db),
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client),
+            patch("app.services.lipsync_service.LipsyncService"),
+        ):
+            try:
+                self._run(mock_db, job_id="j5", fake_self=fake_self)
+                assert False, "expected retry"
+            except _Retry:
+                pass
+        assert mock_job.updated_at is not None
+        assert mock_job.status == "running"
+        assert fake_self.retry.call_count >= 1
+
+
+class TestTtsSubmitDispatchesPoll:
+    """#1867 三层防御 ③：tts_synthesize_and_submit 成功后链式触发 poll."""
+
+    def _run_tts(self, job, mk_client, fake_poll):
+        """运行 tts_synthesize_and_submit.run，直接在用户函数 __globals__ 替换 poll_mediakit_status.
+
+        pytest --import-mode=importlib 下，patch("app.tasks.lipsync_tts.xxx") 拿到的模块对象
+        可能与 celery autoretry 包装下用户函数的 __globals__ 不一致（之前的 Poll 测试通过
+        patch.dict(sys.modules, ...) 注入 fake 模块，进一步加剧了模块命名空间不一致）。
+        直接 patch _orig_run.__func__.__globals__ 是最稳的方式。
+        """
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        session, patches = _apply_all_patches(job=job, mk_client=mk_client)
+        for p in patches:
+            p.__enter__()
+        # 定位 autoretry 包装下的真实用户函数
+        real_tts = tts_synthesize_and_submit._get_current_object()
+        fn = real_tts._orig_run
+        while hasattr(fn, "__func__"):
+            fn = fn.__func__
+        g = fn.__globals__
+        orig_poll = g.get("poll_mediakit_status")
+        g["poll_mediakit_status"] = fake_poll
+        try:
+            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
+        finally:
+            if orig_poll is not None:
+                g["poll_mediakit_status"] = orig_poll
+            else:
+                g.pop("poll_mediakit_status", None)
+            for p in reversed(patches):
+                p.__exit__(None, None, None)
+        return session
+
+    def test_submit_success_dispatches_poll(self):
+        """提交 MediaKit 成功后链式触发 poll_mediakit_status.apply_async (countdown=10)."""
+        job = _make_fake_job()
+        job.emotion = ""
+        mk_client = MagicMock()
+        mk_client.submit_lipsync.return_value = {"success": True, "task_id": "mk-poll-1"}
+        fake_poll = MagicMock()
+        self._run_tts(job, mk_client, fake_poll)
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-poll-1"
+        fake_poll.apply_async.assert_called_once()
+        kwargs = fake_poll.apply_async.call_args.kwargs
+        assert kwargs["kwargs"] == {"job_id": "job-1", "user_id": "user-1"}
+        assert kwargs["countdown"] == 10
+
+    def test_submit_failure_no_poll(self):
+        from app.services.mediakit_client import MediaKitError
+
+        job = _make_fake_job()
+        job.emotion = ""
+        mk_client = MagicMock()
+        mk_client.submit_lipsync.side_effect = MediaKitError("bad", code="X")
+        fake_poll = MagicMock()
+        self._run_tts(job, mk_client, fake_poll)
+        assert job.status == "failed"
+        fake_poll.apply_async.assert_not_called()
+
+    def test_poll_dispatch_fail_does_not_break(self):
+        job = _make_fake_job()
+        job.emotion = ""
+        mk_client = MagicMock()
+        mk_client.submit_lipsync.return_value = {"success": True, "task_id": "mk-poll-2"}
+        fake_poll = MagicMock()
+        fake_poll.apply_async.side_effect = ConnectionError("celery down")
+        session = self._run_tts(job, mk_client, fake_poll)
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-poll-2"
+        session.commit.assert_called()