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

根因:
- 声音克隆走 API提交CosyVoice→Celery worker轮询5分钟→mark ready/failed
- 当worker容器重启、进程OOM或Celeryprefetch消息丢失时,已进入processing状态的voice_clone_profile没有兜底恢复机制
- 对比ingest链路有cleanup_stale_ingest_jobs beat任务+worker_ready启动恢复,voice_clone链路缺失
- 结果: 用户看到"克隆处理中，请稍候..."永久转圈,刷新也不变

修复:
1. SQLAlchemyVoiceCloneProfileRepository新增cleanup_stale_processing方法:
   扫描updated_at超过10分钟的processing记录(正常克隆<5分钟),标记failed并提示用户重试
2. _startup.py新增recover_stale_voice_clones_on_startup+worker_ready信号:
   worker启动时一次性扫描恢复,用户重启/部署后立即解锁卡死任务
3. cleanup.py新增scheduled_cleanup_stale_voice_clones beat任务:
   每5分钟巡检兜底,防止运行期OOM/消息丢失导致的新卡死
4. celery_app.py beat_schedule注册新任务
5. voice_clone.py任务入口加received日志,便于日志排查

apps/worker/worker_app/celery_app.py

@@ -76,4 +76,10 @@ celery_app.conf.beat_schedule = {
         "schedule": 600.0,  # 每 10 分钟（秒）
         "options": {"expires": 540},
     },
+    # 音色克隆卡死巡检：worker 重启/消息丢失后 processing 卡 10 分钟标 failed，用户可点重试
+    "cleanup-stale-voice-clones": {
+        "task": "worker.cleanup_stale_voice_clones",
+        "schedule": 300.0,  # 每 5 分钟
+        "options": {"expires": 240},
+    },
 }

apps/worker/worker_app/tasks/_startup.py

@@ -287,3 +287,50 @@ def _recover_stuck_ingest_jobs_on_ready(sender, **kwargs):  # pragma: no cover
         logger.info("Worker 启动 ingest 恢复完成，共重新派单 %d 个卡死任务", recovered)
     except Exception as e:  # noqa: BLE001 — 启动恢复失败不能阻断 worker 起服
         logger.error("启动 ingest 恢复扫描失败（beat 巡检仍会兜底标 failed）: %s", e, exc_info=True)
+
+
+def recover_stale_voice_clones_on_startup(timeout_minutes: int = 10) -> int:
+    """Worker 启动时恢复卡死在 processing 的音色克隆任务。
+
+    容器重启/进程 OOM 时 worker 中正在轮询的克隆任务会丢失，
+    voice_clone_profiles 永久卡在 processing 无兜底。启动时扫描
+    updated_at 超过 timeout_minutes 的 processing 记录，直接标记
+    为 failed（错误信息指引用户重试）。选择标 failed 而非重新派单，
+    因为 CosyVoice 侧的 voice_id 无法在无上下文下恢复轮询，重试需
+    用户确认后显式触发。
+
+    Args:
+        timeout_minutes: 判定卡死的阈值，默认 10 分钟
+
+    Returns:
+        恢复的记录数
+    """
+    from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+        SQLAlchemyVoiceCloneProfileRepository,
+    )
+
+    try:
+        session = SessionLocal()
+        try:
+            repo = SQLAlchemyVoiceCloneProfileRepository(session)
+            count = repo.cleanup_stale_processing(timeout_minutes)
+        finally:
+            session.close()
+        if count > 0:
+            logger.warning("启动时恢复了 %d 个卡死在 processing 的音色克隆（超时 %d 分钟）", count, timeout_minutes)
+        else:
+            logger.info("无卡死 processing 音色克隆需要恢复")
+        return count
+    except Exception as e:
+        logger.error("启动时音色克隆恢复扫描失败（beat 巡检仍会兜底）: %s", e, exc_info=True)
+        return 0
+
+
+@worker_ready.connect
+def _recover_stuck_voice_clones_on_ready(sender, **kwargs):
+    """Worker 启动完成后恢复卡死的音色克隆任务。"""
+    try:
+        recovered = recover_stale_voice_clones_on_startup()
+        logger.info("Worker 启动音色克隆恢复完成，共标记 %d 个卡死任务为 failed", recovered)
+    except Exception as e:
+        logger.error("启动音色克隆恢复失败（beat 巡检仍会兜底标 failed）: %s", e, exc_info=True)

apps/worker/worker_app/tasks/cleanup.py

@@ -22,6 +22,10 @@ from packages.application.ingest_orphan_cleanup import (
     INGEST_PROCESSING_TIMEOUT_MINUTES,
 )
 
+# 音色克隆 processing 超时：正常克隆轮询最多 5 分钟，10 分钟无更新视为卡死
+VOICE_CLONE_PROCESSING_TIMEOUT_MINUTES = 10
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -125,3 +129,40 @@ def scheduled_cleanup_stale_ingest_jobs(
             purged,
         )
     return {"stale_jobs": total_jobs, "assets_to_error": total_assets, "purged_messages": purged}
+
+
+@shared_task(name="worker.cleanup_stale_voice_clones")
+def scheduled_cleanup_stale_voice_clones(
+    processing_timeout_minutes: int = VOICE_CLONE_PROCESSING_TIMEOUT_MINUTES,
+) -> dict:
+    """Celery Beat: 清理卡死在 processing 的音色克隆档案。
+
+    每 5 分钟执行一次。worker 重启/Celery 消息丢失/进程 OOM 时，
+    已 prefetch 的克隆任务消息丢失，voice_clone_profile 永久卡在 processing。
+    超过 processing_timeout_minutes 未更新的记录标记为 failed，
+    错误信息指引用户点击重试。
+    """
+    from worker_app.db import SessionLocal
+
+    from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+        SQLAlchemyVoiceCloneProfileRepository,
+    )
+
+    session = None
+    try:
+        session = SessionLocal()
+        repo = SQLAlchemyVoiceCloneProfileRepository(session)
+        count = repo.cleanup_stale_processing(processing_timeout_minutes)
+        if count > 0:
+            logger.warning(
+                "[Beat] 清理了 %d 个卡死 processing 的音色克隆（超时 %d 分钟）",
+                count,
+                processing_timeout_minutes,
+            )
+        return {"cleaned": count}
+    except Exception as e:
+        logger.error("[Beat] 清理卡死音色克隆失败: %s", e, exc_info=True)
+        return {"cleaned": 0, "error": str(e)}
+    finally:
+        if session is not None:
+            session.close()

apps/worker/worker_app/tasks/voice_clone.py

@@ -44,6 +44,7 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
     # P2-2 修复：session 初始化为 None，避免 SessionLocal() 抛异常时
     # finally 块中 session.close() 触发 UnboundLocalError
     session = None
+    logger.info(f"Voice clone task started: profile_id={profile_id}")
     try:
         session = SessionLocal()
         repo = SQLAlchemyVoiceCloneProfileRepository(session)

packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py

@@ -136,6 +136,39 @@ class SQLAlchemyVoiceCloneProfileRepository:
         )
         return {voice_id: profile_id for voice_id, profile_id in rows}
 
+    def cleanup_stale_processing(self, timeout_minutes: int = 10) -> int:
+        """清理超时卡在 processing 的克隆档案。
+
+        worker 重启、Celery 任务丢失或 OOM 被杀时，processing 档案会永久卡住。
+        updated_at < NOW() - timeout_minutes 的 processing 记录，标记为 failed
+        并附带明确错误信息，用户可在前端点击「重试」。
+
+        Args:
+            timeout_minutes: 超时分钟数，默认 10 分钟（正常克隆 < 5 分钟）
+
+        Returns:
+            清理的记录数
+        """
+        from datetime import UTC, datetime, timedelta
+
+        cutoff = datetime.now(UTC) - timedelta(minutes=timeout_minutes)
+        models = (
+            self.session.query(VoiceCloneProfileModel)
+            .filter(
+                VoiceCloneProfileModel.status == "processing",
+                VoiceCloneProfileModel.updated_at < cutoff,
+            )
+            .all()
+        )
+        count = 0
+        for model in models:
+            model.status = "failed"
+            model.error_message = f"克隆任务执行超时（超过 {timeout_minutes} 分钟未更新，可能因服务重启中断），请重试"
+            count += 1
+        if count > 0:
+            self.session.commit()
+        return count
+
     @staticmethod
     def _model_to_entity(model: VoiceCloneProfileModel) -> VoiceCloneProfile:
         return VoiceCloneProfile(