apps/api/app/api/routes/tts.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 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_db_session
 from app.schemas.tts import (
     ListTTSJobResponse,
     TTSSynthesizeRequest,
@@ -19,6 +19,7 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.tts_job_repository import (
     SQLAlchemyTTSJobRepository,
 )
+from packages.application.cosyvoice_service import CosyVoiceService
 from packages.application.tts_job.use_cases import (
     CreateTTSJobUseCase,
     DeleteTTSJobUseCase,
@@ -27,6 +28,7 @@ from packages.application.tts_job.use_cases import (
     ListTTSJobsUseCase,
     TTSJobNotFoundError,
 )
+from packages.application.tts_job.workflow import TTSWorkflowService
 
 router = APIRouter()
 
@@ -67,10 +69,11 @@ def synthesize(
     request: TTSSynthesizeRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
 ) -> TTSSynthesizeResponse:
     """发起 TTS 合成任务。
 
-    创建 TTS 任务，状态为 pending，等待后续 CosyVoice API 调用。
+    创建 TTS 任务 → 提交 CosyVoice 合成 → 触发 Celery 异步轮询。
     """
     user_id = authenticated_user.user.id
     use_case = CreateTTSJobUseCase(repository)
@@ -82,6 +85,26 @@ def synthesize(
         voice_clone_profile_id=request.voice_clone_profile_id,
         metadata=request.metadata_,
     )
+
+    # 提交 CosyVoice 合成任务
+    workflow = TTSWorkflowService(
+        repository=repository, cosyvoice_service=cosyvoice_service,
+    )
+    job = workflow.start_synthesis(job.id)
+
+    # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询
+    if job.status.value == "processing":
+        task_id = (job.metadata or {}).get("cosyvoice_task_id", "")
+        if task_id:
+            try:
+                from worker_app.tasks import process_tts_synthesis
+                process_tts_synthesis.delay(job.id)
+            except Exception as e:
+                # Celery 调度失败，标记 job 为 failed
+                workflow.process_synthesis_failure(
+                    job.id, f"Celery 任务调度失败: {e}"
+                )
+
     return TTSSynthesizeResponse(
         job_id=job.id,
         status=job.status,

apps/api/app/api/routes/voices.py

@@ -24,6 +24,7 @@ from app.schemas.voice_library import (
 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.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository
 from packages.application.voice_library.commands import CreateVoiceLibraryCommand, UpdateVoiceLibraryCommand
 from packages.application.voice_library.use_cases import (
@@ -45,6 +46,10 @@ def _get_voice_repository(session: Session = Depends(get_db_session)) -> SQLAlch
     return SQLAlchemyVoiceLibraryRepository(session)
 
 
+def _get_clone_profile_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyVoiceCloneProfileRepository:
+    return SQLAlchemyVoiceCloneProfileRepository(session)
+
+
 def _to_response(item) -> VoiceLibraryItemResponse:
     return VoiceLibraryItemResponse(
         id=item.id,
@@ -65,8 +70,16 @@ def _to_response(item) -> VoiceLibraryItemResponse:
     )
 
 
-def _to_unified_response(item) -> UnifiedVoiceItemResponse:
-    """将数据库音色转换为统一响应格式。"""
+def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoiceItemResponse:
+    """将数据库音色转换为统一响应格式。
+
+    Args:
+        item: VoiceLibraryItem
+        profile_id_map: voice_id → profile_id 映射，用于填充 voice_clone_profile_id
+    """
+    profile_id = None
+    if profile_id_map and item.voice_id:
+        profile_id = profile_id_map.get(item.voice_id)
     return UnifiedVoiceItemResponse(
         id=item.id,
         type="clone",
@@ -83,6 +96,7 @@ def _to_unified_response(item) -> UnifiedVoiceItemResponse:
         tags=item.tags,
         user_id=item.user_id,
         project_id=item.project_id,
+        voice_clone_profile_id=profile_id,
         created_at=item.created_at,
         updated_at=item.updated_at,
     )
@@ -125,6 +139,7 @@ def list_voices_unified(
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    clone_profile_repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_clone_profile_repository),
 ) -> UnifiedVoiceListResponse:
     """获取配音列表（预置音色 + 用户克隆音色）。
 
@@ -148,9 +163,11 @@ def list_voices_unified(
     # 获取克隆音色
     if has_clone:
         use_case = ListVoiceLibraryUseCase(voice_repository)
-        clone_items_raw = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
-        clone_items = [_to_unified_response(i) for i in clone_items_raw]
-        clone_count = voice_repository.count_by_user(user_id, status=status_filter) if status_filter else voice_repository.count_by_user(user_id)
+        clone_items_raw, clone_count = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
+        # 批量查询 voice_id → profile_id 映射，填充 voice_clone_profile_id
+        voice_ids = [i.voice_id for i in clone_items_raw if i.voice_id]
+        profile_id_map = clone_profile_repository.find_profile_ids_by_voice_ids(voice_ids) if voice_ids else {}
+        clone_items = [_to_unified_response(i, profile_id_map) for i in clone_items_raw]
 
     # 组装结果
     if type == "preset":
@@ -214,8 +231,7 @@ def list_voices_legacy(
     """
     user_id = authenticated_user.user.id
     use_case = ListVoiceLibraryUseCase(voice_repository)
-    items = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
-    total = voice_repository.count_by_user(user_id)
+    items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
     return ListVoiceLibraryResponse(
         items=[_to_response(i) for i in items],
         total=total,

apps/worker/worker_app/celery_app.py

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

apps/worker/worker_app/tasks/__init__.py

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

apps/worker/worker_app/tasks/tts_synthesis.py

@@ -0,0 +1,108 @@
+"""TTS synthesis tasks - process TTS synthesis requests via CosyVoice API."""
+
+import logging
+
+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.tts_job_repository import (
+    SQLAlchemyTTSJobRepository,
+)
+from packages.application.cosyvoice_service import (
+    CosyVoiceError,
+    CosyVoiceService,
+    CosyVoiceTimeoutError,
+)
+from packages.application.tts_job.workflow import TTSWorkflowService
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, max_retries=3, name="worker.process_tts_synthesis")
+def process_tts_synthesis(self: Task, job_id: str) -> dict:
+    """处理 TTS 合成任务。
+
+    通过 TTSWorkflowService.poll_and_process_synthesis() 轮询 CosyVoice
+    合成任务状态，更新 TTSJob。
+    超时自动重试（最多 3 次），其他错误标记 job 为 failed。
+
+    Args:
+        job_id: TTSJob ID
+
+    Returns:
+        dict: {"ok": True, "job_id": str, "audio_url": str} 或
+              {"ok": False, "job_id": str, "error": str}
+    """
+    session = None
+    try:
+        session = SessionLocal()
+        repo = SQLAlchemyTTSJobRepository(session)
+        workflow = TTSWorkflowService(
+            repository=repo, cosyvoice_service=CosyVoiceService(),
+        )
+
+        updated_job = workflow.poll_and_process_synthesis(job_id, timeout=120)
+        session.commit()
+
+        logger.info(
+            f"TTS synthesis completed: job_id={job_id}, "
+            f"audio_url={updated_job.output_audio_url}"
+        )
+        return {
+            "ok": True,
+            "job_id": job_id,
+            "audio_url": updated_job.output_audio_url,
+        }
+
+    except Retry:
+        # Celery Retry 异常必须向上传播，不能被后续 except 捕获
+        raise
+
+    except CosyVoiceTimeoutError as e:
+        logger.warning(f"TTS synthesis timeout for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        # 超时重试，指数退避
+        raise self.retry(exc=e, countdown=30)
+
+    except CosyVoiceError as e:
+        logger.error(f"TTS synthesis failed for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        # 标记 job 为 failed
+        try:
+            if session is not None:
+                job = repo.get(job_id)
+                if job is not None:
+                    job.mark_failed(str(e))
+                    repo.update(job)
+                    session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark job as failed: {inner_e}")
+            if session is not None:
+                session.rollback()
+        return {"ok": False, "job_id": job_id, "error": str(e)}
+
+    except Exception as e:
+        logger.error(f"TTS synthesis unexpected error for {job_id}: {e}")
+        if session is not None:
+            session.rollback()
+        # 标记 job 为 failed
+        try:
+            if session is not None:
+                job = repo.get(job_id)
+                if job is not None:
+                    job.mark_failed(str(e))
+                    repo.update(job)
+                    session.commit()
+        except Exception as inner_e:
+            logger.error(f"Failed to mark job as failed: {inner_e}")
+            if session is not None:
+                session.rollback()
+        return {"ok": False, "job_id": job_id, "error": str(e)}
+
+    finally:
+        if session is not None:
+            session.close()

apps/worker/worker_app/tasks/voice_clone.py

@@ -26,7 +26,13 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
 
     通过 VoiceCloneWorkflowService.poll_and_process_clone() 轮询 CosyVoice
     克隆任务状态，更新 VoiceCloneProfile。
-    失败时自动重试（最多 2 次）。
+
+    重试策略（Celery 5.x bind=True 模式）：
+    - max_retries=2：最多重试 2 次（共执行 3 次），超过后抛出 MaxRetriesExceededError
+    - CosyVoiceTimeoutError：网络超时属于临时性故障，使用 countdown=30 延迟 30 秒后重试
+    - CosyVoiceError：API 业务错误（如任务失败），属于永久性故障，不重试直接标记 failed
+    - Exception：未知错误，不重试直接标记 failed，避免无限重试掩盖 bug
+    - Retry 异常：Celery 内部重试信号，必须向上传播不能被捕获
 
     Args:
         profile_id: VoiceCloneProfile ID
@@ -34,8 +40,11 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
     Returns:
         dict: {"ok": True, "profile_id": str, "voice_id": str}
     """
-    session = SessionLocal()
+    # P2-2 修复：session 初始化为 None，避免 SessionLocal() 抛异常时
+    # finally 块中 session.close() 触发 UnboundLocalError
+    session = None
     try:
+        session = SessionLocal()
         repo = SQLAlchemyVoiceCloneProfileRepository(session)
         workflow = VoiceCloneWorkflowService(
             repository=repo, cosyvoice_service=CosyVoiceService(),
@@ -60,39 +69,47 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
 
     except CosyVoiceTimeoutError as e:
         logger.error(f"Voice clone timeout for {profile_id}: {e}")
-        session.rollback()
-        # 超时重试
+        if session is not None:
+            session.rollback()
+        # 超时属于临时性故障，延迟 30 秒后重试
         raise self.retry(exc=e, countdown=30)
 
     except CosyVoiceError as e:
         logger.error(f"Voice clone failed for {profile_id}: {e}")
-        session.rollback()
-        # 标记 profile 为 failed
+        if session is not None:
+            session.rollback()
+        # API 业务错误属于永久性故障，不重试，标记 profile 为 failed
         try:
-            profile = repo.get(profile_id)
-            if profile is not None:
-                profile.mark_failed(str(e))
-                repo.update(profile)
-                session.commit()
+            if session is not None:
+                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()
+            if session is not None:
+                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()
-        # 标记 profile 为 failed
+        if session is not None:
+            session.rollback()
+        # 未知错误不重试，标记 profile 为 failed，避免无限重试掩盖 bug
         try:
-            profile = repo.get(profile_id)
-            if profile is not None:
-                profile.mark_failed(str(e))
-                repo.update(profile)
-                session.commit()
+            if session is not None:
+                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()
+            if session is not None:
+                session.rollback()
         return {"ok": False, "profile_id": profile_id, "error": str(e)}
 
     finally:
-        session.close()
+        if session is not None:
+            session.close()

packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 from sqlalchemy.orm import Session
 
@@ -130,6 +130,23 @@ class SQLAlchemyVoiceCloneProfileRepository:
             return None
         return self._model_to_entity(model)
 
+    def find_profile_ids_by_voice_ids(self, voice_ids: List[str]) -> Dict[str, str]:
+        """批量查询 voice_id → profile_id 映射。用于填充统一列表的 voice_clone_profile_id。"""
+        if not voice_ids:
+            return {}
+        rows = (
+            self.session.query(
+                VoiceCloneProfileModel.voice_id,
+                VoiceCloneProfileModel.id,
+            )
+            .filter(
+                VoiceCloneProfileModel.voice_id.in_(voice_ids),
+                VoiceCloneProfileModel.status != "deleted",
+            )
+            .all()
+        )
+        return {voice_id: profile_id for voice_id, profile_id in rows}
+
     @staticmethod
     def _model_to_entity(model: VoiceCloneProfileModel) -> VoiceCloneProfile:
         return VoiceCloneProfile(

packages/application/cosyvoice_service.py

@@ -391,6 +391,100 @@ class CosyVoiceService:
 
     # ── 语音合成 ─────────────────────────────────────────
 
+    def submit_synthesize_task(
+        self,
+        text: str,
+        voice_id: str = "",
+        sample_rate: int = 0,
+        format: str = "",
+        speed: float = 1.0,
+    ) -> dict:
+        """提交语音合成任务（非阻塞）。
+
+        只提交任务到 CosyVoice API，不轮询结果。
+        返回的 dict 包含 task_id（异步）或 audio_url（同步）。
+
+        Args:
+            text: 要合成的文本
+            voice_id: 音色 ID（预置音色或克隆音色）
+            sample_rate: 采样率（Hz），0 表示使用配置默认值
+            format: 输出格式（mp3/wav/pcm），空表示使用配置默认值
+            speed: 语速（0.5-2.0），1.0 为正常速度
+
+        Returns:
+            dict: {"task_id": str, "audio_url": str, "request_id": str}
+                  task_id 和 audio_url 至少有一个非空
+
+        Raises:
+            CosyVoiceError: API 调用失败
+            CosyVoiceAuthError: 认证失败
+            ValueError: 参数无效
+        """
+        if not text:
+            raise ValueError("text 不能为空")
+        if not voice_id:
+            raise ValueError("voice_id 不能为空")
+        if not self._api_key:
+            raise CosyVoiceAuthError("CosyVoice API Key 未配置")
+
+        settings = get_shared_settings()
+
+        payload = {
+            "model": self._model,
+            "input": {
+                "text": text,
+            },
+            "parameters": {
+                "voice": voice_id,
+                "sample_rate": sample_rate or settings.cosyvoice_sample_rate,
+                "format": format or settings.cosyvoice_format,
+                "rate": speed,
+            },
+        }
+
+        response = self._call_api(
+            method="POST",
+            path="/services/aigc/text2audio/generation",
+            json=payload,
+            timeout=60.0,
+        )
+
+        output = response.get("output", {})
+        task_id = output.get("task_id", "")
+        audio_url = output.get("audio_url", "")
+        request_id = response.get("request_id", "")
+
+        if not task_id and not audio_url:
+            raise CosyVoiceError(
+                f"CosyVoice API 未返回 task_id 或 audio_url: {response}"
+            )
+
+        return {
+            "task_id": task_id,
+            "audio_url": audio_url,
+            "duration": output.get("duration", 0.0),
+            "file_size": output.get("file_size", 0),
+            "request_id": request_id,
+        }
+
+    def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
+        """轮询语音合成任务状态（公开方法）。
+
+        供 Celery 后台任务调用，轮询直到完成或超时。
+
+        Args:
+            task_id: CosyVoice 任务 ID
+            timeout: 超时时间（秒），默认 120
+
+        Returns:
+            dict: {"audio_url": str, "duration": float, "file_size": int}
+
+        Raises:
+            CosyVoiceError: 任务失败
+            CosyVoiceTimeoutError: 超时
+        """
+        return self._poll_synthesize_task(task_id, timeout=timeout)
+
     def synthesize_speech(
         self,
         text: str,

packages/application/tts_job/workflow.py

@@ -0,0 +1,214 @@
+"""TTS Job workflow orchestration — Phase 3.
+
+编排 TTS 合成的完整流程：
+1. 创建 TTSJob（pending）
+2. 提交 CosyVoice 合成任务
+3. 轮询处理合成结果（成功/失败）
+4. 重试失败的合成
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from packages.application.cosyvoice_service import (
+    CosyVoiceAuthError,
+    CosyVoiceError,
+    CosyVoiceService,
+)
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+from packages.ports.tts_job_repository import TTSJobRepository
+
+logger = logging.getLogger(__name__)
+
+
+class TTSWorkflowError(Exception):
+    """TTS 合成工作流异常。"""
+
+    pass
+
+
+class TTSJobNotFoundError(Exception):
+    """TTS 任务未找到。"""
+
+    pass
+
+
+class TTSWorkflowService:
+    """TTS 合成工作流编排服务。
+
+    协调 TTSJobRepository + CosyVoiceService，
+    实现完整的 TTS 合成生命周期管理。
+    """
+
+    def __init__(
+        self,
+        repository: TTSJobRepository,
+        cosyvoice_service: CosyVoiceService,
+    ) -> None:
+        self.repository = repository
+        self.cosyvoice_service = cosyvoice_service
+
+    def start_synthesis(
+        self,
+        job_id: str,
+    ) -> TTSJob:
+        """启动 TTS 合成流程。
+
+        1. 获取 pending 状态的 TTSJob
+        2. 标记为 processing
+        3. 提交 CosyVoice 合成任务
+        4. 保存 task_id 到 metadata
+        5. 返回 job（Celery task 由调用方触发）
+
+        Args:
+            job_id: TTSJob ID
+
+        Returns:
+            TTSJob: 更新后的 job
+
+        Raises:
+            TTSJobNotFoundError: job 不存在
+            TTSWorkflowError: CosyVoice 提交失败
+        """
+        job = self.repository.get(job_id)
+        if job is None:
+            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
+
+        # 标记为 processing
+        job.mark_processing()
+        job = self.repository.update(job)
+
+        try:
+            submit_result = self.cosyvoice_service.submit_synthesize_task(
+                text=job.input_text,
+                voice_id=job.voice_id,
+                sample_rate=job.sample_rate,
+                format=job.format,
+            )
+
+            # 保存 task_id / request_id 到 metadata
+            job_metadata = dict(job.metadata)
+            job_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
+            job_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "")
+
+            # 如果 CosyVoice 同步返回了 audio_url，直接标记完成
+            audio_url = submit_result.get("audio_url", "")
+            if audio_url:
+                job.mark_completed(
+                    output_audio_url=audio_url,
+                    duration=submit_result.get("duration", 0.0),
+                    file_size=submit_result.get("file_size", 0),
+                )
+                job.metadata = job_metadata
+                job = self.repository.update(job)
+                logger.info(
+                    f"TTS 合成同步完成: job_id={job.id}, audio_url={audio_url}"
+                )
+                return job
+
+            job.metadata = job_metadata
+            job = self.repository.update(job)
+            logger.info(
+                f"TTS 合成任务已提交: job_id={job.id}, "
+                f"task_id={submit_result.get('task_id')}"
+            )
+
+        except (CosyVoiceError, CosyVoiceAuthError) as e:
+            job.mark_failed(str(e))
+            job = self.repository.update(job)
+            logger.error(f"TTS 合成提交失败: job_id={job.id}, error={e}")
+        except ValueError as e:
+            job.mark_failed(str(e))
+            job = self.repository.update(job)
+            logger.error(f"TTS 合成参数错误: job_id={job.id}, error={e}")
+
+        return job
+
+    def poll_and_process_synthesis(
+        self, job_id: str, timeout: float = 120.0
+    ) -> TTSJob:
+        """轮询 CosyVoice 合成任务并处理结果。
+
+        从 job.metadata 获取 task_id，调用 CosyVoiceService.poll_synthesize_task()
+        轮询状态，然后通过 process_synthesis_result / process_synthesis_failure 更新 job。
+
+        供 Celery 后台任务调用。
+        """
+        job = self.repository.get(job_id)
+        if job is None:
+            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
+
+        task_id = (job.metadata or {}).get("cosyvoice_task_id", "")
+        if not task_id:
+            raise ValueError(
+                f"TTSJob {job_id} has no cosyvoice_task_id in metadata"
+            )
+
+        result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout)
+        return self.process_synthesis_result(
+            job_id,
+            audio_url=result["audio_url"],
+            duration=result.get("duration", 0.0),
+            file_size=result.get("file_size", 0),
+        )
+
+    def process_synthesis_result(
+        self,
+        job_id: str,
+        audio_url: str,
+        *,
+        duration: float = 0.0,
+        file_size: int = 0,
+    ) -> TTSJob:
+        """处理合成成功结果。
+
+        Args:
+            job_id: TTSJob ID
+            audio_url: 输出音频 URL
+            duration: 音频时长
+            file_size: 文件大小
+
+        Returns:
+            TTSJob: 更新后的 job
+
+        Raises:
+            TTSJobNotFoundError: job 不存在
+        """
+        job = self.repository.get(job_id)
+        if job is None:
+            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
+
+        job.mark_completed(
+            output_audio_url=audio_url,
+            duration=duration,
+            file_size=file_size,
+        )
+        job = self.repository.update(job)
+        logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={audio_url}")
+        return job
+
+    def process_synthesis_failure(
+        self, job_id: str, error_message: str
+    ) -> TTSJob:
+        """处理合成失败结果。
+
+        Args:
+            job_id: TTSJob ID
+            error_message: 错误信息
+
+        Returns:
+            TTSJob: 更新后的 job
+
+        Raises:
+            TTSJobNotFoundError: job 不存在
+        """
+        job = self.repository.get(job_id)
+        if job is None:
+            raise TTSJobNotFoundError(f"TTS job {job_id} not found")
+
+        job.mark_failed(error_message)
+        job = self.repository.update(job)
+        logger.error(f"TTS 合成失败: job_id={job_id}, error={error_message}")
+        return job

packages/application/voice_library/use_cases.py

@@ -25,8 +25,11 @@ class ListVoiceLibraryUseCase:
         status: Optional[str] = None,
         skip: int = 0,
         limit: int = 50,
-    ) -> List[VoiceLibraryItem]:
-        return self.repository.list_by_user(user_id, status=status, skip=skip, limit=limit)
+    ) -> tuple[List[VoiceLibraryItem], int]:
+        """返回 (items, total_count)，避免调用方再单独查一次 count。"""
+        items = self.repository.list_by_user(user_id, status=status, skip=skip, limit=limit)
+        total = self.repository.count_by_user(user_id, status=status) if status else self.repository.count_by_user(user_id)
+        return items, total
 
 
 class GetVoiceLibraryUseCase:

packages/ports/voice_clone_profile_repository.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Dict, List, Protocol
 
 from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
 
@@ -49,3 +49,7 @@ class VoiceCloneProfileRepository(Protocol):
     def find_by_voice_id(self, voice_id: str) -> VoiceCloneProfile | None:
         """根据 CosyVoice 返回的音色 ID 查找档案。"""
         ...
+
+    def find_profile_ids_by_voice_ids(self, voice_ids: List[str]) -> Dict[str, str]:
+        """批量查询 voice_id → profile_id 映射。"""
+        ...

tests/unit/test_voice_library_use_cases.py

@@ -586,11 +586,13 @@ class TestListVoiceLibraryUseCase:
             VoiceLibraryItem(id="v2", user_id="user-001", name="B"),
         ]
         mock_repo.list_by_user.return_value = items
+        mock_repo.count_by_user.return_value = 2
         use_case = ListVoiceLibraryUseCase(repository=mock_repo)
 
-        result = use_case.execute("user-001")
+        result_items, total = use_case.execute("user-001")
 
-        assert len(result) == 2
+        assert len(result_items) == 2
+        assert total == 2
         mock_repo.list_by_user.assert_called_once_with("user-001", status=None, skip=0, limit=50)
 
     def test_list_with_status_filter(self, mock_repo):
@@ -605,11 +607,13 @@ class TestListVoiceLibraryUseCase:
     def test_list_empty(self, mock_repo):
         """测试空列表"""
         mock_repo.list_by_user.return_value = []
+        mock_repo.count_by_user.return_value = 0
         use_case = ListVoiceLibraryUseCase(repository=mock_repo)
 
-        result = use_case.execute("user-001")
+        items, total = use_case.execute("user-001")
 
-        assert result == []
+        assert items == []
+        assert total == 0
 
 
 # ===========================================================================