feat(task-308): 音色克隆完整流程实现

- CosyVoiceService: 新增 submit_clone_task() 和 check_task_status() 非阻塞方法
- VoiceCloneWorkflowService: 编排层，处理 start_clone/process_result/process_failure/retry
- Celery 任务 process_voice_clone: 异步轮询 CosyVoice 克隆结果
- API 路由: 创建后触发 Celery 异步任务，支持重试
- 57 个单元测试全部通过（workflow 14 + task 6 + cosyvoice 37）

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

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_db_session
+from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
 from app.schemas.voice_clone import (
     CreateVoiceCloneRequest,
     ListVoiceCloneResponse,
@@ -13,29 +14,26 @@ from app.schemas.voice_clone import (
     VoiceCloneStatusResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
-from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
     SQLAlchemyVoiceCloneProfileRepository,
 )
+from packages.application.cosyvoice_service import CosyVoiceService
 from packages.application.voice_clone.use_cases import (
-    CreateVoiceCloneUseCase,
     DeleteVoiceCloneUseCase,
     GetVoiceCloneStatusUseCase,
     GetVoiceCloneUseCase,
     ListVoiceClonesUseCase,
-    RetryVoiceCloneUseCase,
     VoiceCloneNotFoundError,
     VoiceCloneNotRetryableError,
 )
+from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
 
-def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyVoiceCloneProfileRepository:
-    return SQLAlchemyVoiceCloneProfileRepository(session)
-
-
 def _to_response(profile) -> VoiceCloneProfileResponse:
     return VoiceCloneProfileResponse(
         id=profile.id,
@@ -57,19 +55,34 @@ def _to_response(profile) -> VoiceCloneProfileResponse:
     )
 
 
-@router.post("", response_model=VoiceCloneProfileResponse, status_code=status.HTTP_201_CREATED)
+def _get_workflow_service(
+    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(
+        get_voice_clone_profile_repository
+    ),
+    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
+) -> VoiceCloneWorkflowService:
+    return VoiceCloneWorkflowService(
+        repository=repository, cosyvoice_service=cosyvoice_service
+    )
+
+
+@router.post(
+    "",
+    response_model=VoiceCloneProfileResponse,
+    status_code=status.HTTP_201_CREATED,
+)
 def create_voice_clone(
     request: CreateVoiceCloneRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
 ) -> VoiceCloneProfileResponse:
     """创建音色克隆任务。
 
-    新建的克隆任务状态为 pending，等待后续 CosyVoice API 调用。
+    创建 VoiceCloneProfile → 提交 CosyVoice 克隆任务 → 触发 Celery 异步轮询。
+    如果有 source_audio_url，状态会变为 processing；否则保持 pending。
     """
     user_id = authenticated_user.user.id
-    use_case = CreateVoiceCloneUseCase(repository)
-    profile = use_case.execute(
+    profile = workflow.start_clone(
         user_id=user_id,
         name=request.name,
         description=request.description,
@@ -80,6 +93,18 @@ def create_voice_clone(
         max_retries=request.max_retries,
         metadata=request.metadata_,
     )
+
+    # 如果 profile 处于 processing 且有 task_id，触发 Celery 异步轮询
+    task_id = (profile.metadata or {}).get("cosyvoice_task_id", "")
+    if profile.status == "processing" and task_id:
+        try:
+            from worker_app.tasks import process_voice_clone
+
+            process_voice_clone.delay(profile.id)
+            logger.info(f"Celery task dispatched for voice clone {profile.id}")
+        except Exception as e:
+            logger.error(f"Failed to dispatch Celery task: {e}")
+
     return _to_response(profile)
 
 
@@ -89,7 +114,9 @@ def list_voice_clones(
     skip: int = Query(0, ge=0),
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(
+        get_voice_clone_profile_repository
+    ),
 ) -> ListVoiceCloneResponse:
     """获取用户的音色克隆列表。"""
     user_id = authenticated_user.user.id
@@ -107,7 +134,9 @@ def list_voice_clones(
 def get_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(
+        get_voice_clone_profile_repository
+    ),
 ) -> VoiceCloneProfileResponse:
     """获取音色克隆详情。"""
     user_id = authenticated_user.user.id
@@ -115,7 +144,9 @@ def get_voice_clone(
     try:
         profile = use_case.execute(clone_id, user_id)
     except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found"
+        )
     return _to_response(profile)
 
 
@@ -123,7 +154,9 @@ def get_voice_clone(
 def get_voice_clone_status(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(
+        get_voice_clone_profile_repository
+    ),
 ) -> VoiceCloneStatusResponse:
     """查询音色克隆状态（用于前端轮询）。"""
     user_id = authenticated_user.user.id
@@ -131,7 +164,9 @@ def get_voice_clone_status(
     try:
         profile = use_case.execute(clone_id, user_id)
     except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found"
+        )
     return VoiceCloneStatusResponse(
         id=profile.id,
         status=profile.status,
@@ -141,18 +176,26 @@ def get_voice_clone_status(
     )
 
 
-@router.delete("/{clone_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response)
+@router.delete(
+    "/{clone_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    response_class=Response,
+)
 def delete_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(
+        get_voice_clone_profile_repository
+    ),
 ) -> Response:
     """删除音色克隆档案。"""
     user_id = authenticated_user.user.id
     use_case = DeleteVoiceCloneUseCase(repository)
     deleted = use_case.execute(clone_id, user_id)
     if not deleted:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found"
+        )
     return Response(status_code=204)
 
 
@@ -160,21 +203,34 @@ def delete_voice_clone(
 def retry_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
-    repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_repository),
+    workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
 ) -> VoiceCloneProfileResponse:
     """重试失败的音色克隆。
 
-    仅当状态为 failed 时可重试，重试后状态变为 pending。
+    仅当状态为 failed 时可重试，重试后重新提交 CosyVoice 克隆任务。
     """
     user_id = authenticated_user.user.id
-    use_case = RetryVoiceCloneUseCase(repository)
     try:
-        profile = use_case.execute(clone_id, user_id)
+        profile = workflow.retry_clone(clone_id, user_id)
     except VoiceCloneNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found"
+        )
     except VoiceCloneNotRetryableError:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="Voice clone is not retryable (only failed clones can be retried)",
         )
+
+    # 如果 profile 处于 processing 且有 task_id，触发 Celery 异步轮询
+    task_id = (profile.metadata or {}).get("cosyvoice_task_id", "")
+    if profile.status == "processing" and task_id:
+        try:
+            from worker_app.tasks import process_voice_clone
+
+            process_voice_clone.delay(profile.id)
+            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}")
+
     return _to_response(profile)

apps/api/app/dependencies.py

@@ -43,6 +43,9 @@ from packages.adapters.sqlalchemy_impl.title_library_repository import (
     SQLAlchemyTitleLibraryRepository,
 )
 from packages.adapters.sqlalchemy_impl.user_repository import SQLAlchemyUserRepository
+from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+    SQLAlchemyVoiceCloneProfileRepository,
+)
 from packages.adapters.sqlalchemy_impl.voice_library_repository import (
     SQLAlchemyVoiceLibraryRepository,
 )
@@ -57,6 +60,7 @@ from packages.ports.job_repository import JobRepository
 from packages.ports.project_repository import ProjectRepository
 from packages.ports.title_library_repository import TitleLibraryRepository
 from packages.ports.user_repository import UserRepository
+from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
 from packages.ports.voice_library_repository import VoiceLibraryRepository
 
 _engine, _SessionLocal = build_session_factory(settings.DATABASE_URL)
@@ -178,3 +182,17 @@ def get_voice_library_repository(
 ) -> SQLAlchemyVoiceLibraryRepository:
     """Provide the SQLAlchemy voice library repository implementation."""
     return SQLAlchemyVoiceLibraryRepository(session)
+
+
+def get_voice_clone_profile_repository(
+    session: Session = Depends(get_db_session),
+) -> SQLAlchemyVoiceCloneProfileRepository:
+    """Provide the SQLAlchemy voice clone profile repository implementation."""
+    return SQLAlchemyVoiceCloneProfileRepository(session)
+
+
+def get_cosyvoice_service() -> "CosyVoiceService":
+    """Provide the CosyVoice service instance."""
+    from packages.application.cosyvoice_service import CosyVoiceService
+
+    return CosyVoiceService()

apps/worker/worker_app/celery_app.py

@@ -12,6 +12,7 @@ celery_app.conf.imports = (
     "worker_app.tasks.classification",
     "worker_app.tasks.generation",
     "worker_app.tasks.voice_extraction",
+    "worker_app.tasks.voice_clone",
     "worker_app.tasks.edit_plan_generation",
     "worker_app.tasks.compose_video",
     "apps.worker.video_processing.dedup",

apps/worker/worker_app/tasks/__init__.py

@@ -33,6 +33,10 @@ def __getattr__(name: str):
         from .voice_extraction import extract_background_task
 
         return extract_background_task
+    elif name == "process_voice_clone":
+        from .voice_clone import process_voice_clone
+
+        return process_voice_clone
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
@@ -44,4 +48,5 @@ __all__ = [
     "ingest_asset",
     "extract_voice_task",
     "extract_background_task",
+    "process_voice_clone",
 ]

apps/worker/worker_app/tasks/voice_clone.py

@@ -0,0 +1,108 @@
+"""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
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+    SQLAlchemyVoiceCloneProfileRepository,
+)
+from packages.application.cosyvoice_service import (
+    CosyVoiceError,
+    CosyVoiceService,
+    CosyVoiceTimeoutError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, max_retries=2, name="worker.process_voice_clone")
+def process_voice_clone(self: Task, profile_id: str) -> dict:
+    """处理音色克隆任务。
+
+    轮询 CosyVoice 克隆任务状态，更新 VoiceCloneProfile。
+    失败时自动重试（最多 2 次）。
+
+    Args:
+        profile_id: VoiceCloneProfile ID
+
+    Returns:
+        dict: {"ok": True, "profile_id": str, "voice_id": str}
+    """
+    session = SessionLocal()
+    try:
+        repo = SQLAlchemyVoiceCloneProfileRepository(session)
+        profile = repo.get(profile_id)
+        if profile is None:
+            raise ValueError(f"VoiceCloneProfile {profile_id} not found")
+
+        # 获取 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)
+        session.commit()
+
+        logger.info(
+            f"Voice clone completed: profile_id={profile_id}, voice_id={voice_id}"
+        )
+        return {"ok": True, "profile_id": profile_id, "voice_id": voice_id}
+
+    except Retry:
+        # Celery Retry 异常必须向上传播，不能被后续 except 捕获
+        raise
+
+    except CosyVoiceTimeoutError as e:
+        logger.error(f"Voice clone timeout for {profile_id}: {e}")
+        session.rollback()
+        # 超时重试
+        raise self.retry(exc=e, countdown=30)
+
+    except CosyVoiceError as e:
+        logger.error(f"Voice clone failed for {profile_id}: {e}")
+        session.rollback()
+        # API 错误，标记为 failed
+        try:
+            repo = SQLAlchemyVoiceCloneProfileRepository(session)
+            profile = repo.get(profile_id)
+            if profile is not None:
+                profile.mark_failed(str(e))
+                repo.update(profile)
+                session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark profile as failed: {inner_e}")
+            session.rollback()
+        return {"ok": False, "profile_id": profile_id, "error": str(e)}
+
+    except Exception as e:
+        logger.error(f"Voice clone unexpected error for {profile_id}: {e}")
+        session.rollback()
+        # 未知错误，标记为 failed
+        try:
+            repo = SQLAlchemyVoiceCloneProfileRepository(session)
+            profile = repo.get(profile_id)
+            if profile is not None:
+                profile.mark_failed(str(e))
+                repo.update(profile)
+                session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark profile as failed: {inner_e}")
+            session.rollback()
+        return {"ok": False, "profile_id": profile_id, "error": str(e)}
+
+    finally:
+        session.close()

packages/application/cosyvoice_service.py

@@ -141,6 +141,105 @@ class CosyVoiceService:
 
     # ── 音色克隆 ─────────────────────────────────────────
 
+    def submit_clone_task(
+        self,
+        audio_url: str,
+        voice_name: str = "",
+        language: str = "zh-CN",
+    ) -> dict:
+        """提交音色克隆任务（非阻塞）。
+
+        只提交任务到 CosyVoice API，不轮询结果。
+        返回的 dict 包含 task_id（异步）或 voice_id（同步）。
+
+        Args:
+            audio_url: 参考音频 URL
+            voice_name: 音色名称（可选）
+            language: 语言代码
+
+        Returns:
+            dict: {"task_id": str, "voice_id": str, "request_id": str}
+                  task_id 和 voice_id 至少有一个非空
+
+        Raises:
+            CosyVoiceError: API 调用失败
+            CosyVoiceAuthError: 认证失败
+            ValueError: 参数无效
+        """
+        if not audio_url:
+            raise ValueError("audio_url 不能为空")
+        if not self._api_key:
+            raise CosyVoiceAuthError("CosyVoice API Key 未配置")
+
+        payload = {
+            "model": self._model,
+            "input": {
+                "audio_url": audio_url,
+            },
+            "parameters": {
+                "language": language,
+            },
+        }
+        if voice_name:
+            payload["parameters"]["voice_name"] = voice_name
+
+        response = self._call_api(
+            method="POST",
+            path="/services/audio/voice-clone",
+            json=payload,
+            timeout=60.0,
+        )
+
+        output = response.get("output", {})
+        task_id = output.get("task_id", "")
+        voice_id = output.get("voice_id", "")
+        request_id = response.get("request_id", "")
+
+        if not task_id and not voice_id:
+            raise CosyVoiceError(
+                f"CosyVoice API 未返回 task_id 或 voice_id: {response}"
+            )
+
+        return {
+            "task_id": task_id,
+            "voice_id": voice_id,
+            "request_id": request_id,
+        }
+
+    def check_task_status(self, task_id: str) -> dict:
+        """查询克隆任务状态（单次查询，不轮询）。
+
+        Args:
+            task_id: 任务 ID
+
+        Returns:
+            dict: {"status": str, "voice_id": str, "message": str}
+                  status 为 SUCCEEDED/FAILED/PENDING/RUNNING
+
+        Raises:
+            CosyVoiceError: API 调用失败
+            CosyVoiceAuthError: 认证失败
+        """
+        if not self._api_key:
+            raise CosyVoiceAuthError("CosyVoice API Key 未配置")
+
+        response = self._call_api(
+            method="GET",
+            path=f"/tasks/{task_id}",
+            timeout=30.0,
+        )
+
+        output = response.get("output", {})
+        status = output.get("task_status", "").upper()
+        voice_id = output.get("voice_id", "")
+        message = output.get("message", "")
+
+        return {
+            "status": status,
+            "voice_id": voice_id,
+            "message": message,
+        }
+
     def clone_voice(
         self,
         audio_url: str,

packages/application/voice_clone/workflow.py

@@ -0,0 +1,265 @@
+"""Voice clone workflow orchestration — Phase 3.
+
+编排音色克隆的完整流程：
+1. 创建 VoiceCloneProfile
+2. 提交 CosyVoice 克隆任务
+3. 处理克隆结果（成功/失败）
+4. 重试失败的克隆
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional
+
+from packages.application.cosyvoice_service import (
+    CosyVoiceAuthError,
+    CosyVoiceError,
+    CosyVoiceService,
+)
+from packages.application.voice_clone.use_cases import (
+    CreateVoiceCloneUseCase,
+    RetryVoiceCloneUseCase,
+    VoiceCloneNotFoundError,
+    VoiceCloneNotRetryableError,
+)
+from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
+from packages.ports.voice_clone_profile_repository import VoiceCloneProfileRepository
+
+logger = logging.getLogger(__name__)
+
+
+class VoiceCloneWorkflowError(Exception):
+    """音色克隆工作流异常。"""
+
+    pass
+
+
+class VoiceCloneWorkflowService:
+    """音色克隆工作流编排服务。
+
+    协调 CreateVoiceCloneUseCase + CosyVoiceService，
+    实现完整的克隆生命周期管理。
+    """
+
+    def __init__(
+        self,
+        repository: VoiceCloneProfileRepository,
+        cosyvoice_service: CosyVoiceService,
+    ) -> None:
+        self.repository = repository
+        self.cosyvoice_service = cosyvoice_service
+
+    def start_clone(
+        self,
+        user_id: str,
+        name: str,
+        *,
+        description: str = "",
+        source_audio_url: str = "",
+        voice_model: str = "",
+        language: str = "zh-CN",
+        gender: str = "unknown",
+        max_retries: int = 3,
+        metadata: Optional[dict] = None,
+    ) -> VoiceCloneProfile:
+        """启动音色克隆流程。
+
+        1. 创建 VoiceCloneProfile (pending)
+        2. 标记为 processing
+        3. 提交 CosyVoice 克隆任务
+        4. 保存 task_id 到 metadata
+        5. 返回 profile（Celery task 由调用方触发）
+
+        Args:
+            user_id: 用户 ID
+            name: 音色名称
+            description: 描述
+            source_audio_url: 参考音频 URL
+            voice_model: 模型名称
+            language: 语言
+            gender: 性别
+            max_retries: 最大重试次数
+            metadata: 扩展元数据
+
+        Returns:
+            VoiceCloneProfile: 已创建的 profile（状态为 processing）
+
+        Raises:
+            VoiceCloneWorkflowError: CosyVoice 提交失败
+        """
+        # 1. 创建 profile
+        create_use_case = CreateVoiceCloneUseCase(self.repository)
+        profile = create_use_case.execute(
+            user_id=user_id,
+            name=name,
+            description=description,
+            source_audio_url=source_audio_url,
+            voice_model=voice_model,
+            language=language,
+            gender=gender,
+            max_retries=max_retries,
+            metadata=metadata,
+        )
+
+        # 2. 标记为 processing
+        profile.mark_processing()
+        profile = self.repository.update(profile)
+
+        # 3. 提交 CosyVoice 克隆任务
+        if source_audio_url:
+            try:
+                submit_result = self.cosyvoice_service.submit_clone_task(
+                    audio_url=source_audio_url,
+                    voice_name=name,
+                    language=language,
+                )
+
+                # 4. 保存 task_id / voice_id 到 metadata
+                task_metadata = dict(profile.metadata)
+                task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
+                task_metadata["cosyvoice_request_id"] = submit_result.get(
+                    "request_id", ""
+                )
+
+                # 如果 CosyVoice 同步返回了 voice_id，直接标记 ready
+                voice_id = submit_result.get("voice_id", "")
+                if voice_id:
+                    profile.mark_ready(voice_id)
+                    profile.metadata = task_metadata
+                    profile = self.repository.update(profile)
+                    logger.info(
+                        f"音色克隆同步完成: profile_id={profile.id}, voice_id={voice_id}"
+                    )
+                    return profile
+
+                profile.metadata = task_metadata
+                profile = self.repository.update(profile)
+                logger.info(
+                    f"音色克隆任务已提交: profile_id={profile.id}, "
+                    f"task_id={submit_result.get('task_id')}"
+                )
+
+            except (CosyVoiceError, CosyVoiceAuthError) as e:
+                # CosyVoice 提交失败，标记为 failed
+                profile.mark_failed(str(e))
+                profile = self.repository.update(profile)
+                logger.error(f"音色克隆提交失败: profile_id={profile.id}, error={e}")
+                return profile
+            except ValueError as e:
+                profile.mark_failed(str(e))
+                profile = self.repository.update(profile)
+                logger.error(f"音色克隆参数错误: profile_id={profile.id}, error={e}")
+                return profile
+        else:
+            # 没有音频 URL，保持 processing 状态等待用户上传
+            logger.info(
+                f"音色克隆已创建但无音频URL: profile_id={profile.id}"
+            )
+
+        return profile
+
+    def process_clone_result(self, profile_id: str, voice_id: str) -> VoiceCloneProfile:
+        """处理克隆成功结果。
+
+        Args:
+            profile_id: Profile ID
+            voice_id: CosyVoice 返回的音色 ID
+
+        Returns:
+            VoiceCloneProfile: 更新后的 profile
+
+        Raises:
+            VoiceCloneNotFoundError: profile 不存在
+        """
+        profile = self.repository.get(profile_id)
+        if profile is None:
+            raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found")
+
+        profile.mark_ready(voice_id)
+        profile = self.repository.update(profile)
+        logger.info(f"音色克隆成功: profile_id={profile_id}, voice_id={voice_id}")
+        return profile
+
+    def process_clone_failure(
+        self, profile_id: str, error_message: str
+    ) -> VoiceCloneProfile:
+        """处理克隆失败结果。
+
+        Args:
+            profile_id: Profile ID
+            error_message: 错误信息
+
+        Returns:
+            VoiceCloneProfile: 更新后的 profile
+
+        Raises:
+            VoiceCloneNotFoundError: profile 不存在
+        """
+        profile = self.repository.get(profile_id)
+        if profile is None:
+            raise VoiceCloneNotFoundError(f"Voice clone {profile_id} not found")
+
+        profile.mark_failed(error_message)
+        profile = self.repository.update(profile)
+        logger.error(f"音色克隆失败: profile_id={profile_id}, error={error_message}")
+        return profile
+
+    def retry_clone(self, clone_id: str, user_id: str) -> VoiceCloneProfile:
+        """重试失败的音色克隆。
+
+        1. 调用 RetryVoiceCloneUseCase 重置状态为 pending
+        2. 标记为 processing
+        3. 重新提交 CosyVoice 克隆任务
+
+        Args:
+            clone_id: Profile ID
+            user_id: 用户 ID
+
+        Returns:
+            VoiceCloneProfile: 更新后的 profile
+
+        Raises:
+            VoiceCloneNotFoundError: profile 不存在
+            VoiceCloneNotRetryableError: 不可重试
+            VoiceCloneWorkflowError: CosyVoice 提交失败
+        """
+        # 1. 重置状态
+        retry_use_case = RetryVoiceCloneUseCase(self.repository)
+        profile = retry_use_case.execute(clone_id, user_id)
+
+        # 2. 标记为 processing
+        profile.mark_processing()
+        profile = self.repository.update(profile)
+
+        # 3. 重新提交 CosyVoice
+        if profile.source_audio_url:
+            try:
+                submit_result = self.cosyvoice_service.submit_clone_task(
+                    audio_url=profile.source_audio_url,
+                    voice_name=profile.name,
+                    language=profile.language,
+                )
+
+                task_metadata = dict(profile.metadata)
+                task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
+                task_metadata["cosyvoice_request_id"] = submit_result.get(
+                    "request_id", ""
+                )
+
+                voice_id = submit_result.get("voice_id", "")
+                if voice_id:
+                    profile.mark_ready(voice_id)
+                    profile.metadata = task_metadata
+                    profile = self.repository.update(profile)
+                    return profile
+
+                profile.metadata = task_metadata
+                profile = self.repository.update(profile)
+
+            except (CosyVoiceError, CosyVoiceAuthError, ValueError) as e:
+                profile.mark_failed(str(e))
+                profile = self.repository.update(profile)
+                logger.error(f"音色克隆重试提交失败: profile_id={clone_id}, error={e}")
+
+        return profile

tests/unit/test_cosyvoice_service.py

@@ -292,6 +292,178 @@ class TestCloneVoice:
             service.clone_voice(audio_url="https://example.com/audio.mp3")
 
 
+# ── submit_clone_task ────────────────────────────────────
+
+
+class TestSubmitCloneTask:
+    """测试 submit_clone_task（非阻塞提交）。"""
+
+    def test_submit_async_returns_task_id(self) -> None:
+        """异步模式：返回 task_id。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={
+                "request_id": "req-001",
+                "output": {"task_id": "task-abc", "task_status": "PENDING"},
+            }
+        )
+
+        service = _make_service(http_client=mock_client)
+        result = service.submit_clone_task(
+            audio_url="https://example.com/audio.mp3",
+            voice_name="测试音色",
+        )
+
+        assert result["task_id"] == "task-abc"
+        assert result["voice_id"] == ""
+        assert result["request_id"] == "req-001"
+        mock_client.request.assert_called_once()
+
+    def test_submit_sync_returns_voice_id(self) -> None:
+        """同步模式：直接返回 voice_id。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={
+                "request_id": "req-002",
+                "output": {"voice_id": "voice-sync-001"},
+            }
+        )
+
+        service = _make_service(http_client=mock_client)
+        result = service.submit_clone_task(
+            audio_url="https://example.com/audio.mp3",
+        )
+
+        assert result["task_id"] == ""
+        assert result["voice_id"] == "voice-sync-001"
+        assert result["request_id"] == "req-002"
+
+    def test_submit_empty_audio_url_raises(self) -> None:
+        """空 audio_url 抛出 ValueError。"""
+        service = _make_service()
+
+        with pytest.raises(ValueError, match="audio_url 不能为空"):
+            service.submit_clone_task(audio_url="")
+
+    def test_submit_no_api_key_raises(self) -> None:
+        """未配置 API Key 抛出 CosyVoiceAuthError。"""
+        service = _make_service(api_key="")
+
+        with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"):
+            service.submit_clone_task(audio_url="https://example.com/audio.mp3")
+
+    def test_submit_no_task_id_or_voice_id_raises(self) -> None:
+        """API 返回无效响应时抛出 CosyVoiceError。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={"output": {}}
+        )
+
+        service = _make_service(http_client=mock_client)
+
+        with pytest.raises(CosyVoiceError, match="未返回 task_id 或 voice_id"):
+            service.submit_clone_task(audio_url="https://example.com/audio.mp3")
+
+    def test_submit_with_voice_name_in_payload(self) -> None:
+        """voice_name 参数包含在请求体中。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={"output": {"task_id": "task-001"}}
+        )
+
+        service = _make_service(http_client=mock_client)
+        service.submit_clone_task(
+            audio_url="https://example.com/audio.mp3",
+            voice_name="我的音色",
+            language="en-US",
+        )
+
+        call_args = mock_client.request.call_args
+        payload = call_args.kwargs.get("json") or call_args[1].get("json")
+        assert payload["parameters"]["voice_name"] == "我的音色"
+        assert payload["parameters"]["language"] == "en-US"
+
+
+# ── check_task_status ────────────────────────────────────
+
+
+class TestCheckTaskStatus:
+    """测试 check_task_status（单次状态查询）。"""
+
+    def test_check_succeeded(self) -> None:
+        """查询成功状态。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={
+                "output": {
+                    "task_status": "SUCCEEDED",
+                    "voice_id": "voice-done-001",
+                },
+            }
+        )
+
+        service = _make_service(http_client=mock_client)
+        result = service.check_task_status("task-abc")
+
+        assert result["status"] == "SUCCEEDED"
+        assert result["voice_id"] == "voice-done-001"
+        assert result["message"] == ""
+
+    def test_check_running(self) -> None:
+        """查询运行中状态。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={
+                "output": {"task_status": "RUNNING"},
+            }
+        )
+
+        service = _make_service(http_client=mock_client)
+        result = service.check_task_status("task-abc")
+
+        assert result["status"] == "RUNNING"
+        assert result["voice_id"] == ""
+
+    def test_check_failed_with_message(self) -> None:
+        """查询失败状态，包含错误消息。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={
+                "output": {
+                    "task_status": "FAILED",
+                    "message": "音频质量不达标",
+                },
+            }
+        )
+
+        service = _make_service(http_client=mock_client)
+        result = service.check_task_status("task-fail")
+
+        assert result["status"] == "FAILED"
+        assert result["message"] == "音频质量不达标"
+
+    def test_check_no_api_key_raises(self) -> None:
+        """未配置 API Key 抛出 CosyVoiceAuthError。"""
+        service = _make_service(api_key="")
+
+        with pytest.raises(CosyVoiceAuthError, match="API Key 未配置"):
+            service.check_task_status("task-abc")
+
+    def test_check_uses_correct_path(self) -> None:
+        """请求路径包含 task_id。"""
+        mock_client = MagicMock(spec=httpx.Client)
+        mock_client.request.return_value = _mock_response(
+            json_data={"output": {"task_status": "PENDING"}}
+        )
+
+        service = _make_service(http_client=mock_client)
+        service.check_task_status("task-xyz-123")
+
+        call_args = mock_client.request.call_args
+        url = call_args.kwargs.get("url") or call_args[1].get("url") or call_args[0][0]
+        assert "/tasks/task-xyz-123" in url
+
+
 # ── synthesize_speech ────────────────────────────────────
 
 

tests/unit/test_voice_clone_task.py

@@ -0,0 +1,239 @@
+"""process_voice_clone Celery 任务单元测试。
+
+关键：voice_clone.py 在模块级别 import worker_app.db.SessionLocal，
+而 worker_app.db 会在导入时调用 ensure_database_exists() 尝试连接 PostgreSQL。
+因此必须在 @patch 装饰器解析模块路径之前，将 worker_app.db 预注入 sys.modules。
+
+Celery 5.x 中 @task(bind=True) 装饰后，task.run 是绑定方法（self 已绑定），
+直接调用 task(profile_id) 即可，不需要手动传 self。
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# worker_app 在 apps/worker 下，需要加入 sys.path
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
+_mock_db_module = MagicMock()
+_mock_db_module.SessionLocal = MagicMock()
+sys.modules.setdefault("worker_app.db", _mock_db_module)
+
+if "worker_app" in sys.modules:
+    sys.modules["worker_app"].db = _mock_db_module
+
+from celery.exceptions import Retry
+
+from packages.application.cosyvoice_service import (
+    CosyVoiceError,
+    CosyVoiceService,
+    CosyVoiceTimeoutError,
+)
+from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
+
+
+def _make_profile(
+    *,
+    status: VoiceCloneStatus = VoiceCloneStatus.PROCESSING,
+    metadata: dict | None = None,
+) -> VoiceCloneProfile:
+    """创建测试用 VoiceCloneProfile。"""
+    if metadata is None:
+        metadata = {"cosyvoice_task_id": "task-abc"}
+    profile = VoiceCloneProfile.create(
+        user_id="user-123",
+        name="测试音色",
+        source_audio_url="https://example.com/audio.wav",
+        max_retries=3,
+        metadata=metadata,
+    )
+    profile.status = status
+    return profile
+
+
+# ── 成功场景 ──────────────────────────────────────────────
+
+
+class TestProcessVoiceCloneSuccess:
+    """测试成功场景。"""
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_success(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """克隆成功：轮询返回 voice_id，profile 标记为 ready。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+        mock_service = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile()
+        mock_repo.get.return_value = profile
+        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_cls.return_value = mock_service
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        # bind=True → run 是绑定方法，直接调用 task(profile_id)
+        result = process_voice_clone("profile-123")
+
+        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_session.commit.assert_called_once()
+        mock_session.close.assert_called_once()
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_profile_not_found(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """profile 不存在时返回 failed。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+        mock_repo_cls.return_value = mock_repo
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        result = process_voice_clone("nonexistent")
+
+        assert result["ok"] is False
+        assert "not found" in result["error"].lower()
+        mock_session.close.assert_called_once()
+
+
+# ── 超时场景 ──────────────────────────────────────────────
+
+
+class TestProcessVoiceCloneTimeout:
+    """测试超时场景。"""
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_timeout_retries(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """超时时调用 self.retry() 进行重试，Retry 异常向上传播。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+        mock_service = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile()
+        mock_repo.get.return_value = profile
+        mock_repo_cls.return_value = mock_repo
+
+        mock_service._poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时")
+        mock_service_cls.return_value = mock_service
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        # mock task.retry 使其抛出 Retry（模拟 Celery 行为）
+        with patch.object(process_voice_clone, "retry", side_effect=Retry("retrying")):
+            with pytest.raises(Retry):
+                process_voice_clone("profile-123")
+
+        mock_session.rollback.assert_called_once()
+        mock_session.close.assert_called_once()
+
+
+# ── 失败场景 ──────────────────────────────────────────────
+
+
+class TestProcessVoiceCloneFailure:
+    """测试失败场景。"""
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_cosyvoice_error(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """CosyVoice 错误：profile 标记为 failed。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+        mock_service = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile()
+        mock_repo.get.return_value = profile
+        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_cls.return_value = mock_service
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        result = process_voice_clone("profile-123")
+
+        assert result["ok"] is False
+        assert "克隆失败" in result["error"]
+        mock_session.close.assert_called_once()
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_unexpected_error(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """意外异常：profile 标记为 failed。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+        mock_service = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile()
+        mock_repo.get.return_value = profile
+        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_cls.return_value = mock_service
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        result = process_voice_clone("profile-123")
+
+        assert result["ok"] is False
+        assert "未知错误" in result["error"]
+        mock_session.close.assert_called_once()
+
+    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
+    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
+    def test_process_voice_clone_no_task_id(
+        self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock
+    ) -> None:
+        """metadata 中没有 cosyvoice_task_id 时返回 failed。"""
+        mock_session = MagicMock()
+        mock_repo = MagicMock()
+
+        # 显式传入空 dict，确保没有 cosyvoice_task_id
+        profile = _make_profile(metadata={})
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+        mock_repo_cls.return_value = mock_repo
+
+        _mock_db_module.SessionLocal.return_value = mock_session
+
+        from worker_app.tasks.voice_clone import process_voice_clone
+
+        result = process_voice_clone("profile-123")
+
+        assert result["ok"] is False
+        assert "task_id" in result["error"]
+        mock_session.close.assert_called_once()

tests/unit/test_voice_clone_workflow.py

@@ -0,0 +1,322 @@
+"""VoiceCloneWorkflowService 单元测试。"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.application.cosyvoice_service import (
+    CosyVoiceAuthError,
+    CosyVoiceError,
+    CosyVoiceService,
+)
+from packages.application.voice_clone.use_cases import (
+    VoiceCloneNotFoundError,
+    VoiceCloneNotRetryableError,
+)
+from packages.application.voice_clone.workflow import VoiceCloneWorkflowService
+from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
+
+
+def _make_profile(
+    *,
+    status: VoiceCloneStatus = VoiceCloneStatus.PENDING,
+    source_audio_url: str = "https://example.com/audio.wav",
+    retry_count: int = 0,
+    max_retries: int = 3,
+    metadata: dict | None = None,
+) -> VoiceCloneProfile:
+    """创建测试用 VoiceCloneProfile。"""
+    profile = VoiceCloneProfile.create(
+        user_id="user-123",
+        name="测试音色",
+        source_audio_url=source_audio_url,
+        max_retries=max_retries,
+        metadata=metadata,
+    )
+    profile.status = status
+    profile.retry_count = retry_count
+    return profile
+
+
+def _make_service(
+    *,
+    repo: MagicMock | None = None,
+    cosyvoice: MagicMock | None = None,
+) -> VoiceCloneWorkflowService:
+    """创建测试用 VoiceCloneWorkflowService。"""
+    mock_repo = repo or MagicMock()
+    mock_cosyvoice = cosyvoice or MagicMock(spec=CosyVoiceService)
+    return VoiceCloneWorkflowService(
+        repository=mock_repo, cosyvoice_service=mock_cosyvoice
+    )
+
+
+# ── start_clone ──────────────────────────────────────────
+
+
+class TestStartClone:
+    """测试 start_clone 方法。"""
+
+    def test_start_clone_with_async_task(self) -> None:
+        """异步模式：提交任务后返回 processing 状态的 profile。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        # CosyVoice 返回 task_id（异步模式）
+        mock_cosyvoice.submit_clone_task.return_value = {
+            "task_id": "task-abc",
+            "voice_id": "",
+            "request_id": "req-123",
+        }
+
+        # repo.create 和 repo.update 返回传入的 profile
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="https://example.com/audio.wav",
+        )
+
+        assert profile.status == VoiceCloneStatus.PROCESSING
+        assert profile.metadata["cosyvoice_task_id"] == "task-abc"
+        assert profile.metadata["cosyvoice_request_id"] == "req-123"
+        mock_cosyvoice.submit_clone_task.assert_called_once()
+        assert mock_repo.create.call_count == 1
+        # update 至少调用 2 次：mark_processing + 保存 task_id
+        assert mock_repo.update.call_count >= 2
+
+    def test_start_clone_with_sync_result(self) -> None:
+        """同步模式：CosyVoice 直接返回 voice_id，profile 变为 ready。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_cosyvoice.submit_clone_task.return_value = {
+            "task_id": "",
+            "voice_id": "voice-sync-123",
+            "request_id": "req-456",
+        }
+
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="https://example.com/audio.wav",
+        )
+
+        assert profile.status == VoiceCloneStatus.READY
+        assert profile.voice_id == "voice-sync-123"
+
+    def test_start_clone_cosyvoice_error(self) -> None:
+        """CosyVoice 提交失败，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceError("API 调用失败")
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="https://example.com/audio.wav",
+        )
+
+        assert profile.status == VoiceCloneStatus.FAILED
+        assert "API 调用失败" in profile.error_message
+
+    def test_start_clone_auth_error(self) -> None:
+        """CosyVoice 认证失败，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceAuthError("认证失败")
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="https://example.com/audio.wav",
+        )
+
+        assert profile.status == VoiceCloneStatus.FAILED
+        assert "认证失败" in profile.error_message
+
+    def test_start_clone_without_audio_url(self) -> None:
+        """没有音频 URL 时，profile 保持 processing 状态。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        mock_repo.create.side_effect = lambda p: p
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        profile = service.start_clone(
+            user_id="user-123",
+            name="测试音色",
+            source_audio_url="",
+        )
+
+        # 没有音频 URL，保持 processing 状态
+        assert profile.status == VoiceCloneStatus.PROCESSING
+        mock_cosyvoice.submit_clone_task.assert_not_called()
+
+
+# ── process_clone_result ─────────────────────────────────
+
+
+class TestProcessCloneResult:
+    """测试 process_clone_result 方法。"""
+
+    def test_process_clone_result_success(self) -> None:
+        """克隆成功，profile 标记为 ready。"""
+        mock_repo = MagicMock()
+        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo)
+        result = service.process_clone_result(profile.id, "voice-xyz")
+
+        assert result.status == VoiceCloneStatus.READY
+        assert result.voice_id == "voice-xyz"
+
+    def test_process_clone_result_not_found(self) -> None:
+        """profile 不存在时抛出异常。"""
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+
+        service = _make_service(repo=mock_repo)
+        with pytest.raises(VoiceCloneNotFoundError):
+            service.process_clone_result("nonexistent", "voice-xyz")
+
+
+# ── process_clone_failure ────────────────────────────────
+
+
+class TestProcessCloneFailure:
+    """测试 process_clone_failure 方法。"""
+
+    def test_process_clone_failure(self) -> None:
+        """克隆失败，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        service = _make_service(repo=mock_repo)
+        result = service.process_clone_failure(profile.id, "超时错误")
+
+        assert result.status == VoiceCloneStatus.FAILED
+        assert result.error_message == "超时错误"
+
+    def test_process_clone_failure_not_found(self) -> None:
+        """profile 不存在时抛出异常。"""
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+
+        service = _make_service(repo=mock_repo)
+        with pytest.raises(VoiceCloneNotFoundError):
+            service.process_clone_failure("nonexistent", "错误")
+
+
+# ── retry_clone ──────────────────────────────────────────
+
+
+class TestRetryClone:
+    """测试 retry_clone 方法。"""
+
+    def test_retry_clone_with_async_task(self) -> None:
+        """重试成功，异步模式。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.FAILED, retry_count=1, max_retries=3
+        )
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        mock_cosyvoice.submit_clone_task.return_value = {
+            "task_id": "task-retry",
+            "voice_id": "",
+            "request_id": "req-retry",
+        }
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        result = service.retry_clone(profile.id, "user-123")
+
+        assert result.status == VoiceCloneStatus.PROCESSING
+        assert result.metadata["cosyvoice_task_id"] == "task-retry"
+        assert result.retry_count == 2  # prepare_retry 增加了一次
+
+    def test_retry_clone_with_sync_result(self) -> None:
+        """重试成功，同步模式。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.FAILED, retry_count=0, max_retries=3
+        )
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        mock_cosyvoice.submit_clone_task.return_value = {
+            "task_id": "",
+            "voice_id": "voice-retry-sync",
+            "request_id": "req-retry",
+        }
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        result = service.retry_clone(profile.id, "user-123")
+
+        assert result.status == VoiceCloneStatus.READY
+        assert result.voice_id == "voice-retry-sync"
+
+    def test_retry_clone_not_found(self) -> None:
+        """profile 不存在时抛出异常。"""
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+
+        service = _make_service(repo=mock_repo)
+        with pytest.raises(VoiceCloneNotFoundError):
+            service.retry_clone("nonexistent", "user-123")
+
+    def test_retry_clone_not_retryable(self) -> None:
+        """不可重试时抛出异常。"""
+        mock_repo = MagicMock()
+        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
+        mock_repo.get.return_value = profile
+
+        service = _make_service(repo=mock_repo)
+        with pytest.raises(VoiceCloneNotRetryableError):
+            service.retry_clone(profile.id, "user-123")
+
+    def test_retry_clone_cosyvoice_error(self) -> None:
+        """重试时 CosyVoice 失败，profile 标记为 failed。"""
+        mock_repo = MagicMock()
+        mock_cosyvoice = MagicMock(spec=CosyVoiceService)
+
+        profile = _make_profile(
+            status=VoiceCloneStatus.FAILED, retry_count=0, max_retries=3
+        )
+        mock_repo.get.return_value = profile
+        mock_repo.update.side_effect = lambda p: p
+
+        mock_cosyvoice.submit_clone_task.side_effect = CosyVoiceError("重试失败")
+
+        service = _make_service(repo=mock_repo, cosyvoice=mock_cosyvoice)
+        result = service.retry_clone(profile.id, "user-123")
+
+        assert result.status == VoiceCloneStatus.FAILED
+        assert "重试失败" in result.error_message