apps/api/app/api/routes/voice_clones.py

@@ -27,7 +27,10 @@ from packages.application.voice_clone.use_cases import (
     VoiceCloneNotFoundError,
     VoiceCloneNotRetryableError,
 )
-from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
+from packages.application.voice_clone.workflow import (
+    VoiceCloneWorkflowError,
+    VoiceCloneWorkflowService,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -104,6 +107,15 @@ def create_voice_clone(
             logger.info(f"Celery task dispatched for voice clone {profile.id}")
         except Exception as e:
             logger.error(f"Failed to dispatch Celery task: {e}")
+            # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing
+            try:
+                workflow.process_clone_failure(
+                    profile.id, f"Celery 任务调度失败: {e}"
+                )
+            except Exception as inner_e:
+                logger.error(
+                    f"Failed to mark profile as failed after dispatch error: {inner_e}"
+                )
 
     return _to_response(profile)
 
@@ -232,5 +244,14 @@ def retry_voice_clone(
             logger.info(f"Celery task dispatched for voice clone retry {profile.id}")
         except Exception as e:
             logger.error(f"Failed to dispatch Celery task: {e}")
+            # P2-3: Celery 调度失败时标记 profile 为 failed，避免永久卡在 processing
+            try:
+                workflow.process_clone_failure(
+                    profile.id, f"Celery 任务调度失败: {e}"
+                )
+            except Exception as inner_e:
+                logger.error(
+                    f"Failed to mark profile as failed after dispatch error: {inner_e}"
+                )
 
     return _to_response(profile)

apps/worker/worker_app/tasks/voice_clone.py

@@ -1,7 +1,6 @@
 """Voice clone tasks - process voice clone requests via CosyVoice API."""
 
 import logging
-from typing import Optional
 
 from celery import Task
 from celery.exceptions import Retry
@@ -16,6 +15,7 @@ from packages.application.cosyvoice_service import (
     CosyVoiceService,
     CosyVoiceTimeoutError,
 )
+from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
 
 logger = logging.getLogger(__name__)
 
@@ -24,7 +24,8 @@ logger = logging.getLogger(__name__)
 def process_voice_clone(self: Task, profile_id: str) -> dict:
     """处理音色克隆任务。
 
-    轮询 CosyVoice 克隆任务状态，更新 VoiceCloneProfile。
+    通过 VoiceCloneWorkflowService.poll_and_process_clone() 轮询 CosyVoice
+    克隆任务状态，更新 VoiceCloneProfile。
     失败时自动重试（最多 2 次）。
 
     Args:
@@ -36,31 +37,22 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
     session = SessionLocal()
     try:
         repo = SQLAlchemyVoiceCloneProfileRepository(session)
-        profile = repo.get(profile_id)
-        if profile is None:
-            raise ValueError(f"VoiceCloneProfile {profile_id} not found")
+        workflow = VoiceCloneWorkflowService(
+            repository=repo, cosyvoice_service=CosyVoiceService(),
+        )
 
-        # 获取 CosyVoice task_id
-        task_id = (profile.metadata or {}).get("cosyvoice_task_id", "")
-        if not task_id:
-            raise ValueError(
-                f"VoiceCloneProfile {profile_id} has no cosyvoice_task_id in metadata"
-            )
-
-        # 轮询 CosyVoice 任务状态
-        service = CosyVoiceService()
-        result = service._poll_clone_task(task_id, timeout=300)
-        voice_id = result["voice_id"]
-
-        # 更新 profile 状态为 ready
-        profile.mark_ready(voice_id)
-        repo.update(profile)
+        updated_profile = workflow.poll_and_process_clone(profile_id, timeout=300)
         session.commit()
 
         logger.info(
-            f"Voice clone completed: profile_id={profile_id}, voice_id={voice_id}"
+            f"Voice clone completed: profile_id={profile_id}, "
+            f"voice_id={updated_profile.voice_id}"
         )
-        return {"ok": True, "profile_id": profile_id, "voice_id": voice_id}
+        return {
+            "ok": True,
+            "profile_id": profile_id,
+            "voice_id": updated_profile.voice_id,
+        }
 
     except Retry:
         # Celery Retry 异常必须向上传播，不能被后续 except 捕获
@@ -75,9 +67,8 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
     except CosyVoiceError as e:
         logger.error(f"Voice clone failed for {profile_id}: {e}")
         session.rollback()
-        # API 错误，标记为 failed
+        # 标记 profile 为 failed
         try:
-            repo = SQLAlchemyVoiceCloneProfileRepository(session)
             profile = repo.get(profile_id)
             if profile is not None:
                 profile.mark_failed(str(e))
@@ -91,9 +82,8 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
     except Exception as e:
         logger.error(f"Voice clone unexpected error for {profile_id}: {e}")
         session.rollback()
-        # 未知错误，标记为 failed
+        # 标记 profile 为 failed
         try:
-            repo = SQLAlchemyVoiceCloneProfileRepository(session)
             profile = repo.get(profile_id)
             if profile is not None:
                 profile.mark_failed(str(e))

packages/application/cosyvoice_service.py

@@ -240,6 +240,24 @@ class CosyVoiceService:
             "message": message,
         }
 
+    def poll_clone_task(self, task_id: str, timeout: float = 300.0) -> dict:
+        """轮询音色克隆任务状态（公开方法）。
+
+        供 Celery 后台任务调用，轮询直到完成或超时。
+
+        Args:
+            task_id: CosyVoice 任务 ID
+            timeout: 超时时间（秒），默认 300
+
+        Returns:
+            dict: {"voice_id": str}
+
+        Raises:
+            CosyVoiceError: 任务失败
+            CosyVoiceTimeoutError: 超时
+        """
+        return self._poll_clone_task(task_id, timeout=timeout)
+
     def clone_voice(
         self,
         audio_url: str,

packages/application/voice_clone/workflow.py

@@ -66,10 +66,10 @@ class VoiceCloneWorkflowService:
         """启动音色克隆流程。
 
         1. 创建 VoiceCloneProfile (pending)
-        2. 标记为 processing
-        3. 提交 CosyVoice 克隆任务
-        4. 保存 task_id 到 metadata
-        5. 返回 profile（Celery task 由调用方触发）
+        2. 若有音频 URL：标记 processing → 提交 CosyVoice 克隆任务
+           若无音频 URL：保持 pending，等待用户上传
+        3. 保存 task_id 到 metadata
+        4. 返回 profile（Celery task 由调用方触发）
 
         Args:
             user_id: 用户 ID
@@ -83,7 +83,7 @@ class VoiceCloneWorkflowService:
             metadata: 扩展元数据
 
         Returns:
-            VoiceCloneProfile: 已创建的 profile（状态为 processing）
+            VoiceCloneProfile: 已创建的 profile（有 URL 时为 processing，无 URL 时为 pending）
 
         Raises:
             VoiceCloneWorkflowError: CosyVoice 提交失败
@@ -102,12 +102,12 @@ class VoiceCloneWorkflowService:
             metadata=metadata,
         )
 
-        # 2. 标记为 processing
-        profile.mark_processing()
-        profile = self.repository.update(profile)
-
-        # 3. 提交 CosyVoice 克隆任务
+        # 2. 提交 CosyVoice 克隆任务（仅有音频 URL 时才标记 processing）
         if source_audio_url:
+            # 标记为 processing
+            profile.mark_processing()
+            profile = self.repository.update(profile)
+
             try:
                 submit_result = self.cosyvoice_service.submit_clone_task(
                     audio_url=source_audio_url,
@@ -152,13 +152,34 @@ class VoiceCloneWorkflowService:
                 logger.error(f"音色克隆参数错误: profile_id={profile.id}, error={e}")
                 return profile
         else:
-            # 没有音频 URL，保持 processing 状态等待用户上传
+            # 没有音频 URL，保持 pending 状态等待用户上传
             logger.info(
-                f"音色克隆已创建但无音频URL: profile_id={profile.id}"
+                f"音色克隆已创建但无音频URL，保持pending: profile_id={profile.id}"
             )
 
         return profile
 
+    def poll_and_process_clone(
+        self, profile_id: str, timeout: float = 300.0
+    ) -> VoiceCloneProfile:
+        """轮询 CosyVoice 克隆任务并处理结果。
+
+        从 profile.metadata 获取 task_id，调用 CosyVoiceService.poll_clone_task()
+        轮询状态，然后通过 process_clone_result / process_clone_failure 更新 profile。
+
+        供 Celery 后台任务调用，避免直接访问私有方法。
+        """
+        profile = self.repository.get(profile_id)
+        if profile is None:
+            raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found")
+        task_id = (profile.metadata or {}).get("cosyvoice_task_id", "")
+        if not task_id:
+            raise ValueError(
+                f"VoiceCloneProfile {profile_id} has no cosyvoice_task_id in metadata"
+            )
+        result = self.cosyvoice_service.poll_clone_task(task_id, timeout=timeout)
+        return self.process_clone_result(profile_id, result["voice_id"])
+
     def process_clone_result(self, profile_id: str, voice_id: str) -> VoiceCloneProfile:
         """处理克隆成功结果。
 

tests/unit/test_voice_clone_task.py

@@ -77,7 +77,7 @@ class TestProcessVoiceCloneSuccess:
         mock_repo.update.side_effect = lambda p: p
         mock_repo_cls.return_value = mock_repo
 
-        mock_service._poll_clone_task.return_value = {"voice_id": "voice-xyz"}
+        mock_service.poll_clone_task.return_value = {"voice_id": "voice-xyz"}
         mock_service_cls.return_value = mock_service
 
         _mock_db_module.SessionLocal.return_value = mock_session
@@ -89,7 +89,7 @@ class TestProcessVoiceCloneSuccess:
 
         assert result["ok"] is True
         assert result["voice_id"] == "voice-xyz"
-        mock_service._poll_clone_task.assert_called_once_with("task-abc", timeout=300)
+        mock_service.poll_clone_task.assert_called_once_with("task-abc", timeout=300)
         mock_session.commit.assert_called_once()
         mock_session.close.assert_called_once()
 
@@ -135,7 +135,7 @@ class TestProcessVoiceCloneTimeout:
         mock_repo.get.return_value = profile
         mock_repo_cls.return_value = mock_repo
 
-        mock_service._poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时")
+        mock_service.poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时")
         mock_service_cls.return_value = mock_service
 
         _mock_db_module.SessionLocal.return_value = mock_session
@@ -172,7 +172,7 @@ class TestProcessVoiceCloneFailure:
         mock_repo.update.side_effect = lambda p: p
         mock_repo_cls.return_value = mock_repo
 
-        mock_service._poll_clone_task.side_effect = CosyVoiceError("克隆失败")
+        mock_service.poll_clone_task.side_effect = CosyVoiceError("克隆失败")
         mock_service_cls.return_value = mock_service
 
         _mock_db_module.SessionLocal.return_value = mock_session
@@ -200,7 +200,7 @@ class TestProcessVoiceCloneFailure:
         mock_repo.update.side_effect = lambda p: p
         mock_repo_cls.return_value = mock_repo
 
-        mock_service._poll_clone_task.side_effect = RuntimeError("未知错误")
+        mock_service.poll_clone_task.side_effect = RuntimeError("未知错误")
         mock_service_cls.return_value = mock_service
 
         _mock_db_module.SessionLocal.return_value = mock_session

tests/unit/test_voice_clone_workflow.py

@@ -153,7 +153,7 @@ class TestStartClone:
         assert "认证失败" in profile.error_message
 
     def test_start_clone_without_audio_url(self) -> None:
-        """没有音频 URL 时，profile 保持 processing 状态。"""
+        """没有音频 URL 时，profile 保持 pending 状态（P2-1 修复后）。"""
         mock_repo = MagicMock()
         mock_cosyvoice = MagicMock(spec=CosyVoiceService)
 
@@ -167,8 +167,8 @@ class TestStartClone:
             source_audio_url="",
         )
 
-        # 没有音频 URL，保持 processing 状态
-        assert profile.status == VoiceCloneStatus.PROCESSING
+        # P2-1: 没有音频 URL 时不标记 processing，保持 pending
+        assert profile.status == VoiceCloneStatus.PENDING
         mock_cosyvoice.submit_clone_task.assert_not_called()
 
 
@@ -320,3 +320,71 @@ class TestRetryClone:
 
         assert result.status == VoiceCloneStatus.FAILED
         assert "重试失败" in result.error_message
+
+
+# ── poll_and_process_clone ───────────────────────────────
+
+
+class TestPollAndProcessClone:
+    """测试 poll_and_process_clone 方法（P2-2 修复）。"""
+
+    def test_poll_and_process_clone_success(self) -> None:
+        """轮询成功：调用 poll_clone_task → process_clone_result。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.PROCESSING,
+            metadata={"cosyvoice_task_id": "task-abc"},
+        )
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        mock_cosyvoice.poll_clone_task.return_value = {"voice_id": "voice-poll-xyz"}
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        result = service.poll_and_process_clone(profile.id)
+
+        assert result.status == VoiceCloneStatus.READY
+        assert result.voice_id == "voice-poll-xyz"
+        mock_cosyvoice.poll_clone_task.assert_called_once_with("task-abc", timeout=300)
+
+    def test_poll_and_process_clone_no_task_id(self) -> None:
+        """metadata 中没有 task_id 时抛出 ValueError。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(status=VoiceCloneStatus.PROCESSING, metadata={})
+        mock_repo.get.return_value = profile
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        with pytest.raises(ValueError, match="cosyvoice_task_id"):
+            service.poll_and_process_clone(profile.id)
+
+    def test_poll_and_process_clone_not_found(self) -> None:
+        """profile 不存在时抛出 VoiceCloneNotFoundError。"""
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+
+        service = _make_service(repo=mock_repo)
+        with pytest.raises(VoiceCloneNotFoundError):
+            service.poll_and_process_clone("nonexistent")
+
+    def test_poll_and_process_clone_timeout(self) -> None:
+        """超时时透传 CosyVoiceTimeoutError（由 Celery task 捕获重试）。"""
+        from packages.application.cosyvoice_service import CosyVoiceTimeoutError
+
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.PROCESSING,
+            metadata={"cosyvoice_task_id": "task-abc"},
+        )
+        mock_repo.get.return_value = profile
+
+        mock_cosyvoice.poll_clone_task.side_effect = CosyVoiceTimeoutError("超时")
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        with pytest.raises(CosyVoiceTimeoutError):
+            service.poll_and_process_clone(profile.id)