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(

tests/unit/test_voice_clone_cleanup.py

@@ -0,0 +1,156 @@
+"""SQLAlchemyVoiceCloneProfileRepository.cleanup_stale_processing 单元测试。
+
+通过 monkeypatch sys.modules['packages.adapters.sqlalchemy_impl.models']，
+注入一个具备 SQLAlchemy 列比较语义（== / < 返回可链式 .all() 的 mock）的假模型类，
+不依赖真实 DB，也不会触发 SQLAlchemy 映射。
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class _Col:
+    """模拟 SQLAlchemy Column：比较运算返回 MagicMock，可被 filter 链式调用。"""
+
+    def __init__(self, name: str):
+        self._name = name
+
+    def __eq__(self, other):  # type: ignore[override]
+        return MagicMock(name=f"{self._name}=={other!r}")
+
+    def __ne__(self, other):  # type: ignore[override]
+        return MagicMock(name=f"{self._name}!={other!r}")
+
+    def __lt__(self, other):
+        return MagicMock(name=f"{self._name}<{other!r}")
+
+    def __gt__(self, other):
+        return MagicMock(name=f"{self._name}>{other!r}")
+
+    def __le__(self, other):
+        return MagicMock(name=f"{self._name}<={other!r}")
+
+    def __ge__(self, other):
+        return MagicMock(name=f"{self._name}>={other!r}")
+
+    def __hash__(self):
+        return id(self)
+
+
+class _FakeVoiceCloneProfileModel:
+    """假模型：类属性是 _Col；实例上可读写 status/error_message/updated_at。"""
+
+    status = _Col("status")
+    updated_at = _Col("updated_at")
+    id = _Col("id")
+    error_message = _Col("error_message")
+
+    def __init__(self, **kwargs):
+        self.__dict__.update(kwargs)
+
+
+# ── 预注入 mock 模型模块，避免真实 import 拉起 DB / SQLAlchemy 映射 ──
+_fake_models = SimpleNamespace(VoiceCloneProfileModel=_FakeVoiceCloneProfileModel)
+sys.modules.setdefault("packages.adapters.sqlalchemy_impl.models", _fake_models)
+if "packages.adapters.sqlalchemy_impl.voice_clone_profile_repository" in sys.modules:
+    mod = sys.modules["packages.adapters.sqlalchemy_impl.voice_clone_profile_repository"]
+    mod.VoiceCloneProfileModel = _FakeVoiceCloneProfileModel  # type: ignore[attr-defined]
+
+from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+    SQLAlchemyVoiceCloneProfileRepository,
+)
+
+
+def _make_fake_row(
+    *,
+    status: str = "processing",
+    updated_at: datetime | None = None,
+    error_message: str = "",
+) -> _FakeVoiceCloneProfileModel:
+    return _FakeVoiceCloneProfileModel(
+        status=status,
+        error_message=error_message,
+        updated_at=updated_at or datetime.now(UTC),
+    )
+
+
+def _make_repo(fake_rows: list[_FakeVoiceCloneProfileModel]):
+    """构造 repo + mock session。
+
+    生产代码使用 .query(Model).filter(A, B).all()（一次 filter，两个表达式参数）。
+    """
+    session = MagicMock()
+    filtered = MagicMock()
+    filtered.all.return_value = list(fake_rows)
+    session.query.return_value.filter.return_value = filtered
+
+    repo = SQLAlchemyVoiceCloneProfileRepository.__new__(SQLAlchemyVoiceCloneProfileRepository)
+    repo.session = session
+    return repo, session
+
+
+class TestCleanupStaleProcessing:
+    """cleanup_stale_processing 行为测试。"""
+
+    def test_no_stale_records_returns_zero_and_no_commit(self):
+        """无卡死记录时返回 0，不调用 commit。"""
+        repo, session = _make_repo([])
+        assert repo.cleanup_stale_processing() == 0
+        session.commit.assert_not_called()
+
+    def test_stale_record_marked_failed_with_timeout_message(self):
+        """超时 processing 记录被标记为 failed，错误信息包含超时分钟数。"""
+        old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=15))
+        repo, session = _make_repo([old])
+
+        count = repo.cleanup_stale_processing(timeout_minutes=10)
+
+        assert count == 1
+        assert old.status == "failed"
+        assert "超时" in old.error_message
+        assert "10" in old.error_message
+        session.commit.assert_called_once()
+
+    def test_error_message_reflects_custom_timeout(self):
+        """自定义 timeout_minutes 会反映在错误信息里。"""
+        old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(hours=1))
+        repo, _session = _make_repo([old])
+
+        repo.cleanup_stale_processing(timeout_minutes=5)
+
+        assert old.status == "failed"
+        assert "5" in old.error_message
+
+    def test_multiple_stale_records_all_cleaned_in_single_commit(self):
+        """多条卡死记录都被清理，返回正确计数并只 commit 一次。"""
+        m1 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=20))
+        m2 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=11))
+        repo, session = _make_repo([m1, m2])
+
+        assert repo.cleanup_stale_processing(timeout_minutes=10) == 2
+        assert m1.status == "failed"
+        assert m2.status == "failed"
+        session.commit.assert_called_once()
+
+    def test_queries_model_with_status_and_updated_at_filters(self):
+        """query 被调用，filter 同时传入 status=='processing' 与 updated_at<cutoff 两个条件。
+
+        注：不同测试加载顺序下 sys.modules['packages...models'] 可能是真模型类
+        （因为其他测试文件已先 import），所以这里不断言模型类身份，
+        只断言 query/filter 被正确调用。
+        """
+        repo, session = _make_repo([])
+        repo.cleanup_stale_processing()
+
+        assert session.query.called, "session.query 应被调用"
+        # filter 被调用一次，且传入两个过滤表达式
+        q = session.query.return_value
+        assert q.filter.called, "query.filter 应被调用"
+        args_f, _kwargs = q.filter.call_args
+        assert len(args_f) == 2, f"filter 应接收 2 个位置参数（status + updated_at），实际 {len(args_f)}"

tests/unit/test_voice_clone_task.py

@@ -8,13 +8,14 @@
 Celery bind=True 任务的底层函数签名为 (self, profile_id)，
 CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow，必须 mock 防止真实初始化。
 
-跨环境兼容：
-  Python 3.13 + Celery 5.4.0 → import 返回 Celery Proxy
-    → _get_current_object() 返回 Task 实例 → .run 是 bound method（self 已绑定）
-    → 调用方式：task.run(profile_id)，retry mock 在 task.run.retry
-  Python 3.10 + Celery 5.4.0 → import 返回原始函数（装饰器未生效）
-    → 签名 (self, profile_id)，需手动传 mock_self
-    → 调用方式：func(mock_self, profile_id)，retry mock 在 mock_self.retry
+跨环境兼容（_resolve_task）：
+  不同 Celery 版本 / Python 版本 / 是否有 active Celery app，task 对象形态不同：
+    1) Celery Proxy（LocalProxy/LazyProxy）：import 结果是代理对象，调用
+       _get_current_object() 可能抛 RuntimeError（无 active context），必须 try 保护。
+       成功取到真实 Task 实例后，使用 bound method .run。
+    2) Celery Task 实例（bind=True 时 @task 返回的典型形态）：直接有 .run/.retry。
+    3) 原始函数（某些环境装饰器未生效或 patch 时序问题）：需手动传 mock_self。
+  统一返回 (callable, mock_self, real_task)，调用方不需要重复解析。
 """
 
 from __future__ import annotations
@@ -58,24 +59,33 @@ def _make_mock_profile(
 
 
 def _resolve_task(task_obj):
-    """解析 Celery 任务对象，返回 (callable, mock_self_or_none)。
+    """解析 Celery 任务对象，兼容 Proxy / Task 实例 / 原始函数三种形态。
 
-    跨环境兼容 Celery Proxy / Task 实例 / 原始函数三种情况。
+    所有分支均做异常保护，避免因 Celery Proxy 在无 app context 时抛错导致测试挂掉。
 
     Returns:
-        tuple: (callable, mock_self)
-            - Proxy/Task: callable 是 bound method task.run，mock_self=None
-            - 原始函数: callable 是原始函数，mock_self 需由调用方提供
+        tuple: (callable, mock_self, real_task)
+            - callable: 最终执行用的可调用对象
+            - mock_self: 仅原始函数分支需要手动传入 mock self；其他分支为 None
+            - real_task: 真实 Task 实例（Proxy 分支为 _get_current_object() 结果；
+              Task 分支为 task_obj 本身；原始函数分支为 None）。用于 patch .retry。
     """
-    # Case 1: Celery Proxy → 提取 Task 实例的 .run（bound method）
+    # Case 1: Celery Proxy → 安全尝试 _get_current_object()
     if hasattr(task_obj, "_get_current_object"):
-        real_task = task_obj._get_current_object()
-        return real_task.run, None
+        try:
+            real_task = task_obj._get_current_object()
+            if real_task is not None and hasattr(real_task, "run"):
+                return real_task.run, None, real_task
+        except Exception:
+            # 无 active app context 或 Proxy 未绑定，退化为其他分支处理
+            pass
+
     # Case 2: Celery Task 实例（非 Proxy）
     if hasattr(task_obj, "run") and hasattr(task_obj, "retry"):
-        return task_obj.run, None
-    # Case 3: 原始函数（CI 环境中装饰器未生效）
-    return task_obj, MagicMock()
+        return task_obj.run, None, task_obj
+
+    # Case 3: 原始函数（装饰器未生效）
+    return task_obj, MagicMock(), None
 
 
 # ── 成功场景 ──────────────────────────────────────────────
@@ -110,8 +120,8 @@ class TestProcessVoiceCloneSuccess:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
-        args = (mock_self, "profile-123") if mock_self else ("profile-123",)
+        func, mock_self, _ = _resolve_task(process_voice_clone)
+        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
         result = func(*args)
 
         assert result["ok"] is True
@@ -148,8 +158,8 @@ class TestProcessVoiceCloneSuccess:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
-        args = (mock_self, "nonexistent") if mock_self else ("nonexistent",)
+        func, mock_self, _ = _resolve_task(process_voice_clone)
+        args = (mock_self, "nonexistent") if mock_self is not None else ("nonexistent",)
         result = func(*args)
 
         assert result["ok"] is False
@@ -189,24 +199,24 @@ class TestProcessVoiceCloneTimeout:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
+        func, mock_self, real_task = _resolve_task(process_voice_clone)
 
-        # 设置 retry mock：根据环境不同，retry 在不同对象上
-        if mock_self is None:
-            # Proxy/Task 环境：retry 在 Task 实例上（func 是 bound method task.run）
-            real_task = process_voice_clone._get_current_object()
-            mock_retry = MagicMock()
-            mock_retry.side_effect = Retry("retrying")
-            with patch.object(real_task, "retry", mock_retry):
-                with pytest.raises(Retry):
-                    func("profile-123")
-                mock_retry.assert_called_once()
-        else:
+        if mock_self is not None:
             # 原始函数环境：retry 在 mock_self 上
             mock_self.retry.side_effect = Retry("retrying")
             with pytest.raises(Retry):
                 func(mock_self, "profile-123")
             mock_self.retry.assert_called_once()
+        else:
+            # Proxy/Task 环境：retry 在 Task 实例上。用 _resolve_task 返回的 real_task，
+            # 避免再次 _get_current_object() 在无 context 时抛 AttributeError。
+            retry_target = real_task if real_task is not None else process_voice_clone
+            mock_retry = MagicMock()
+            mock_retry.side_effect = Retry("retrying")
+            with patch.object(retry_target, "retry", mock_retry):
+                with pytest.raises(Retry):
+                    func("profile-123")
+                mock_retry.assert_called_once()
 
         mock_session.rollback.assert_called_once()
         mock_session.close.assert_called_once()
@@ -243,8 +253,8 @@ class TestProcessVoiceCloneFailure:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
-        args = (mock_self, "profile-123") if mock_self else ("profile-123",)
+        func, mock_self, _ = _resolve_task(process_voice_clone)
+        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
         result = func(*args)
 
         assert result["ok"] is False
@@ -277,8 +287,8 @@ class TestProcessVoiceCloneFailure:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
-        args = (mock_self, "profile-123") if mock_self else ("profile-123",)
+        func, mock_self, _ = _resolve_task(process_voice_clone)
+        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
         result = func(*args)
 
         assert result["ok"] is False
@@ -311,8 +321,8 @@ class TestProcessVoiceCloneFailure:
 
         from worker_app.tasks.voice_clone import process_voice_clone
 
-        func, mock_self = _resolve_task(process_voice_clone)
-        args = (mock_self, "profile-123") if mock_self else ("profile-123",)
+        func, mock_self, _ = _resolve_task(process_voice_clone)
+        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
         result = func(*args)
 
         assert result["ok"] is False